hyperguild/ingestion/internal/webhook/webhook_test.go

package webhook

import (
	"bytes"
	"context"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"os"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/kubernetes/fake"
	k8stesting "k8s.io/client-go/testing"
)

// newFakeClientset simulates the real API server's GenerateName expansion,
// which the plain fake clientset tracker does not do on its own -- without
// this, every created Job keeps an empty Name (a fake-clientset limitation,
// not real API server behavior).
func newFakeClientset(objects ...runtime.Object) *fake.Clientset {
	cs := fake.NewSimpleClientset(objects...)
	cs.PrependReactor("create", "jobs", func(action k8stesting.Action) (bool, runtime.Object, error) {
		createAction := action.(k8stesting.CreateAction)
		job, ok := createAction.GetObject().(*batchv1.Job)
		if ok && job.Name == "" && job.GenerateName != "" {
			job.Name = job.GenerateName + "test0001"
		}
		return false, nil, nil // not "handled" -- let the default reactor store it
	})
	return cs
}

func sign(t *testing.T, payload []byte, secret string) string {
	t.Helper()
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write(payload)
	return hex.EncodeToString(mac.Sum(nil))
}

func testLogger() *slog.Logger {
	return slog.New(slog.NewTextHandler(os.Stderr, nil))
}

func fakeCronJob(namespace, name string) *batchv1.CronJob {
	return &batchv1.CronJob{
		ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace},
		Spec: batchv1.CronJobSpec{
			JobTemplate: batchv1.JobTemplateSpec{
				Spec: batchv1.JobSpec{
					Template: corev1.PodTemplateSpec{
						Spec: corev1.PodSpec{
							Containers: []corev1.Container{
								{Name: "sync", Image: "alpine/git:v2.47.2"},
							},
							RestartPolicy: corev1.RestartPolicyNever,
						},
					},
				},
			},
		},
	}
}

// --------------------------------------------------------------------------- #
// VerifySignature
// --------------------------------------------------------------------------- #
func TestVerifySignature_AcceptsCorrectHMAC(t *testing.T) {
	payload := []byte(`{"ref":"refs/heads/main"}`)
	secret := "s3cret"
	sig := sign(t, payload, secret)
	assert.True(t, VerifySignature(payload, sig, secret))
}

func TestVerifySignature_RejectsWrongSecret(t *testing.T) {
	payload := []byte(`{"ref":"refs/heads/main"}`)
	sig := sign(t, payload, "right-secret")
	assert.False(t, VerifySignature(payload, sig, "wrong-secret"))
}

func TestVerifySignature_RejectsTamperedPayload(t *testing.T) {
	secret := "s3cret"
	sig := sign(t, []byte(`{"ref":"refs/heads/main"}`), secret)
	assert.False(t, VerifySignature([]byte(`{"ref":"refs/heads/evil"}`), sig, secret))
}

func TestVerifySignature_RejectsEmptySignature(t *testing.T) {
	assert.False(t, VerifySignature([]byte("payload"), "", "secret"))
}

// --------------------------------------------------------------------------- #
// TriggerJobFromCronJob
// --------------------------------------------------------------------------- #
func TestTriggerJobFromCronJob_CreatesJobMatchingTemplate(t *testing.T) {
	cs := newFakeClientset(fakeCronJob("brain", "brain-sync"))

	jobName, err := TriggerJobFromCronJob(context.Background(), cs, "brain", "brain-sync")
	require.NoError(t, err)
	assert.NotEmpty(t, jobName)

	jobs, err := cs.BatchV1().Jobs("brain").List(context.Background(), metav1.ListOptions{})
	require.NoError(t, err)
	require.Len(t, jobs.Items, 1)
	assert.Equal(t, "alpine/git:v2.47.2", jobs.Items[0].Spec.Template.Spec.Containers[0].Image)
}

func TestTriggerJobFromCronJob_ErrorsWhenCronJobMissing(t *testing.T) {
	cs := fake.NewSimpleClientset()
	_, err := TriggerJobFromCronJob(context.Background(), cs, "brain", "brain-sync")
	assert.Error(t, err)
}

// --------------------------------------------------------------------------- #
// Handler.ServeHTTP
// --------------------------------------------------------------------------- #
func newTestHandler(cs *fake.Clientset) *Handler {
	return &Handler{
		Secret:      "s3cret",
		Clientset:   cs,
		Namespace:   "brain",
		CronJobName: "brain-sync",
		WatchRepo:   "mathias/brain",
		Logger:      testLogger(),
	}
}

func pushPayload(t *testing.T, repo, ref string) []byte {
	t.Helper()
	body := map[string]any{
		"ref": ref,
		"repository": map[string]any{
			"full_name": repo,
		},
	}
	b, err := json.Marshal(body)
	require.NoError(t, err)
	return b
}

func TestHandler_RejectsMissingSignature(t *testing.T) {
	cs := fake.NewSimpleClientset(fakeCronJob("brain", "brain-sync"))
	h := newTestHandler(cs)
	payload := pushPayload(t, "mathias/brain", "refs/heads/main")
	req := httptest.NewRequest(http.MethodPost, "/webhooks/brain-sync", bytes.NewReader(payload))
	rec := httptest.NewRecorder()

	h.ServeHTTP(rec, req)

	assert.Equal(t, http.StatusUnauthorized, rec.Code)
	jobs, _ := cs.BatchV1().Jobs("brain").List(context.Background(), metav1.ListOptions{})
	assert.Empty(t, jobs.Items, "must not trigger a job on an unsigned request")
}

func TestHandler_RejectsWrongSignature(t *testing.T) {
	cs := fake.NewSimpleClientset(fakeCronJob("brain", "brain-sync"))
	h := newTestHandler(cs)
	payload := pushPayload(t, "mathias/brain", "refs/heads/main")
	req := httptest.NewRequest(http.MethodPost, "/webhooks/brain-sync", bytes.NewReader(payload))
	req.Header.Set("X-Gitea-Signature", sign(t, payload, "not-the-real-secret"))
	rec := httptest.NewRecorder()

	h.ServeHTTP(rec, req)

	assert.Equal(t, http.StatusUnauthorized, rec.Code)
	jobs, _ := cs.BatchV1().Jobs("brain").List(context.Background(), metav1.ListOptions{})
	assert.Empty(t, jobs.Items)
}

func TestHandler_IgnoresOtherRepos(t *testing.T) {
	cs := fake.NewSimpleClientset(fakeCronJob("brain", "brain-sync"))
	h := newTestHandler(cs)
	payload := pushPayload(t, "mathias/some-other-repo", "refs/heads/main")
	req := httptest.NewRequest(http.MethodPost, "/webhooks/brain-sync", bytes.NewReader(payload))
	req.Header.Set("X-Gitea-Signature", sign(t, payload, "s3cret"))
	rec := httptest.NewRecorder()

	h.ServeHTTP(rec, req)

	assert.Equal(t, http.StatusOK, rec.Code)
	jobs, _ := cs.BatchV1().Jobs("brain").List(context.Background(), metav1.ListOptions{})
	assert.Empty(t, jobs.Items, "must not trigger for a push to an unrelated repo")
}

func TestHandler_IgnoresNonMainBranch(t *testing.T) {
	cs := fake.NewSimpleClientset(fakeCronJob("brain", "brain-sync"))
	h := newTestHandler(cs)
	payload := pushPayload(t, "mathias/brain", "refs/heads/some-feature-branch")
	req := httptest.NewRequest(http.MethodPost, "/webhooks/brain-sync", bytes.NewReader(payload))
	req.Header.Set("X-Gitea-Signature", sign(t, payload, "s3cret"))
	rec := httptest.NewRecorder()

	h.ServeHTTP(rec, req)

	assert.Equal(t, http.StatusOK, rec.Code)
	jobs, _ := cs.BatchV1().Jobs("brain").List(context.Background(), metav1.ListOptions{})
	assert.Empty(t, jobs.Items, "must not trigger for a push to a non-main branch")
}

func TestHandler_TriggersJobOnValidMainPush(t *testing.T) {
	cs := fake.NewSimpleClientset(fakeCronJob("brain", "brain-sync"))
	h := newTestHandler(cs)
	payload := pushPayload(t, "mathias/brain", "refs/heads/main")
	req := httptest.NewRequest(http.MethodPost, "/webhooks/brain-sync", bytes.NewReader(payload))
	req.Header.Set("X-Gitea-Signature", sign(t, payload, "s3cret"))
	rec := httptest.NewRecorder()

	h.ServeHTTP(rec, req)

	assert.Equal(t, http.StatusOK, rec.Code)
	jobs, err := cs.BatchV1().Jobs("brain").List(context.Background(), metav1.ListOptions{})
	require.NoError(t, err)
	assert.Len(t, jobs.Items, 1, "a valid push to main on the watched repo must trigger exactly one job")
}