package controlplane import ( "bytes" "encoding/json" "log/slog" "net/http" "net/http/httptest" "testing" "git.toki-labs.com/toki/gito/services/core/internal/config" "git.toki-labs.com/toki/gito/services/core/internal/protosocket" ) func TestHealthz(t *testing.T) { router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default()) req := httptest.NewRequest(http.MethodGet, "/healthz", nil) rec := httptest.NewRecorder() router.ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status: got %d", rec.Code) } var body map[string]string if err := json.NewDecoder(rec.Body).Decode(&body); err != nil { t.Fatalf("decode: %v", err) } if body["status"] != "ok" { t.Fatalf("body: %#v", body) } } func TestProtoSocketRegistry(t *testing.T) { router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default()) req := httptest.NewRequest(http.MethodGet, "/proto-socket", nil) rec := httptest.NewRecorder() router.ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status: got %d", rec.Code) } var body ProtoSocketRegistry if err := json.NewDecoder(rec.Body).Decode(&body); err != nil { t.Fatalf("decode: %v", err) } if body.Transport != "proto-socket" { t.Fatalf("transport: got %q", body.Transport) } channels := make(map[string]ProtoSocketChannel, len(body.Channels)) for _, channel := range body.Channels { channels[channel.Name] = channel } assertPlaceholderChannel(t, channels, "operation", []string{"create", "cancel", "get", "stream"}) assertChannelActions(t, channels, "event", "mvp", map[string]string{ "subscribe": "mvp", "list": "mvp", "ack": "placeholder", }) } func assertPlaceholderChannel(t *testing.T, channels map[string]ProtoSocketChannel, name string, actions []string) { t.Helper() channel, ok := channels[name] if !ok { t.Fatalf("missing channel %q", name) } if channel.Status != "placeholder" { t.Fatalf("channel %q status: got %q", name, channel.Status) } actionStatuses := make(map[string]string, len(channel.Actions)) for _, action := range channel.Actions { actionStatuses[action.Name] = action.Status } for _, action := range actions { if actionStatuses[action] != "placeholder" { t.Fatalf("channel %q action %q status: got %q", name, action, actionStatuses[action]) } } } func assertChannelActions(t *testing.T, channels map[string]ProtoSocketChannel, name, status string, actions map[string]string) { t.Helper() channel, ok := channels[name] if !ok { t.Fatalf("missing channel %q", name) } if channel.Status != status { t.Fatalf("channel %q status: got %q want %q", name, channel.Status, status) } actionStatuses := make(map[string]string, len(channel.Actions)) for _, action := range channel.Actions { actionStatuses[action.Name] = action.Status } for action, want := range actions { if actionStatuses[action] != want { t.Fatalf("channel %q action %q status: got %q want %q", name, action, actionStatuses[action], want) } } } func TestForgejoPushCreatesWatchedBranchEvent(t *testing.T) { router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default()) registerBranchListener(t, router, `{"repo_id":"nomadcode","branch":"develop","provider":"forgejo"}`) req := httptest.NewRequest(http.MethodPost, "/callbacks/forgejo/push?repo_id=nomadcode", bytes.NewBufferString(`{ "ref": "refs/heads/develop", "before": "111", "after": "222", "repository": {"name": "nomadcode", "full_name": "toki/nomadcode"}, "commits": [{"id": "222", "modified": ["agent-roadmap/phase/a/milestones/b.md"]}] }`)) req.Header.Set("X-Forgejo-Event", "push") req.Header.Set("X-Forgejo-Delivery", "delivery-1") rec := httptest.NewRecorder() router.ServeHTTP(rec, req) if rec.Code != http.StatusAccepted { t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String()) } var pushed map[string]any if err := json.NewDecoder(rec.Body).Decode(&pushed); err != nil { t.Fatalf("decode push response: %v", err) } if pushed["matched"] != true { t.Fatalf("expected push to match listener: %#v", pushed) } eventsReq := httptest.NewRequest(http.MethodGet, "/api/events", nil) eventsRec := httptest.NewRecorder() router.ServeHTTP(eventsRec, eventsReq) if eventsRec.Code != http.StatusOK { t.Fatalf("events status: got %d", eventsRec.Code) } var body map[string][]map[string]any if err := json.NewDecoder(eventsRec.Body).Decode(&body); err != nil { t.Fatalf("decode events: %v", err) } if len(body["events"]) != 1 { t.Fatalf("events: %#v", body) } event := body["events"][0] if event["type"] != "branch.updated" || event["provider"] != "forgejo" || event["delivery_id"] != "delivery-1" { t.Fatalf("unexpected event: %#v", event) } revision := event["revision"].(map[string]any) if revision["repo_id"] != "nomadcode" || revision["branch"] != "develop" || revision["before"] != "111" || revision["after"] != "222" { t.Fatalf("unexpected revision: %#v", revision) } } func TestForgejoPushIgnoresUnwatchedBranch(t *testing.T) { router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default()) registerBranchListener(t, router, `{"repo_id":"nomadcode","branch":"develop","provider":"forgejo"}`) req := httptest.NewRequest(http.MethodPost, "/callbacks/forgejo/push?repo_id=nomadcode", bytes.NewBufferString(`{ "ref": "refs/heads/feature/a", "before": "111", "after": "222", "repository": {"name": "nomadcode", "full_name": "toki/nomadcode"} }`)) req.Header.Set("X-Forgejo-Event", "push") rec := httptest.NewRecorder() router.ServeHTTP(rec, req) if rec.Code != http.StatusAccepted { t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String()) } var pushed map[string]any if err := json.NewDecoder(rec.Body).Decode(&pushed); err != nil { t.Fatalf("decode push response: %v", err) } if pushed["matched"] != false { t.Fatalf("expected push to be ignored: %#v", pushed) } if _, ok := pushed["event"]; ok { t.Fatalf("unmatched push must not expose an emitted event: %#v", pushed) } } func TestForgejoPushRequiresSignatureBeforeIgnoringUnsupportedEvent(t *testing.T) { router := NewRouter(config.Config{ AppEnv: "test", ProtoSocketPath: "/proto-socket", ForgejoWebhookSecret: "forgejo-secret", }, slog.Default()) req := httptest.NewRequest(http.MethodPost, "/callbacks/forgejo/push", bytes.NewBufferString(`{"unsupported":true}`)) req.Header.Set("X-Forgejo-Event", "repository") rec := httptest.NewRecorder() router.ServeHTTP(rec, req) if rec.Code != http.StatusUnauthorized { t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String()) } } func TestProviderCallbackRecordsReceivedWebhookEvent(t *testing.T) { router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default()) req := httptest.NewRequest(http.MethodPost, "/callbacks/providers/fake", bytes.NewBufferString(`{ "action": "ping", "token": "top-secret" }`)) req.Header.Set("X-Gito-Provider-Event", "ping") req.Header.Set("X-Gito-Delivery", "fake-delivery-1") rec := httptest.NewRecorder() router.ServeHTTP(rec, req) if rec.Code != http.StatusAccepted { t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String()) } if bytes.Contains(rec.Body.Bytes(), []byte("top-secret")) || bytes.Contains(rec.Body.Bytes(), []byte(`"token"`)) { t.Fatalf("provider callback response leaked raw payload: %s", rec.Body.String()) } var created map[string]any if err := json.NewDecoder(rec.Body).Decode(&created); err != nil { t.Fatalf("decode callback response: %v", err) } event, ok := created["event"].(map[string]any) if !ok { t.Fatalf("missing event: %#v", created) } if event["type"] != "provider.webhook.received" || event["provider"] != "fake" || event["delivery_id"] != "fake-delivery-1" { t.Fatalf("unexpected event: %#v", event) } webhook, ok := event["webhook"].(map[string]any) if !ok { t.Fatalf("missing webhook metadata: %#v", event) } if webhook["event_type"] != "ping" || webhook["external_id"] != "fake-delivery-1" { t.Fatalf("unexpected webhook metadata: %#v", webhook) } if _, ok := webhook["payload_size"].(float64); !ok { t.Fatalf("payload_size missing: %#v", webhook) } eventsReq := httptest.NewRequest(http.MethodGet, "/api/events", nil) eventsRec := httptest.NewRecorder() router.ServeHTTP(eventsRec, eventsReq) if eventsRec.Code != http.StatusOK { t.Fatalf("events status: got %d", eventsRec.Code) } if bytes.Contains(eventsRec.Body.Bytes(), []byte("top-secret")) || bytes.Contains(eventsRec.Body.Bytes(), []byte(`"token"`)) { t.Fatalf("event list leaked raw payload: %s", eventsRec.Body.String()) } var listed map[string][]map[string]any if err := json.NewDecoder(eventsRec.Body).Decode(&listed); err != nil { t.Fatalf("decode events response: %v", err) } if len(listed["events"]) != 1 { t.Fatalf("events: %#v", listed) } if listed["events"][0]["type"] != "provider.webhook.received" { t.Fatalf("unexpected listed event: %#v", listed["events"][0]) } } func TestWebhookSubscriptionRegistersFilterWithoutLeakingSecretRef(t *testing.T) { router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default()) req := httptest.NewRequest(http.MethodPost, "/api/webhook-subscriptions", bytes.NewBufferString(`{ "name": "nomadcode-develop-wakeup", "target_url": "https://consumer.example.invalid/gito/events", "events": ["branch.updated"], "repo_id": "nomadcode", "branch": "develop", "secret_ref": "secret://gito/webhooks/nomadcode" }`)) rec := httptest.NewRecorder() router.ServeHTTP(rec, req) if rec.Code != http.StatusCreated { t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String()) } if bytes.Contains(rec.Body.Bytes(), []byte("secret://gito/webhooks/nomadcode")) || bytes.Contains(rec.Body.Bytes(), []byte("secret_ref")) || bytes.Contains(rec.Body.Bytes(), []byte("consumer.example.invalid")) { t.Fatalf("subscription response leaked private subscription input: %s", rec.Body.String()) } var created map[string]map[string]any if err := json.NewDecoder(rec.Body).Decode(&created); err != nil { t.Fatalf("decode subscription response: %v", err) } subscription := created["subscription"] if subscription["id"] == "" { t.Fatalf("missing subscription id: %#v", subscription) } if subscription["name"] != "nomadcode-develop-wakeup" || subscription["repo_id"] != "nomadcode" || subscription["branch"] != "develop" { t.Fatalf("unexpected subscription payload: %#v", subscription) } events, ok := subscription["events"].([]any) if !ok || len(events) != 1 || events[0] != "branch.updated" { t.Fatalf("events: %#v", subscription["events"]) } listReq := httptest.NewRequest(http.MethodGet, "/api/webhook-subscriptions", nil) listRec := httptest.NewRecorder() router.ServeHTTP(listRec, listReq) if listRec.Code != http.StatusOK { t.Fatalf("list status: got %d body=%s", listRec.Code, listRec.Body.String()) } if bytes.Contains(listRec.Body.Bytes(), []byte("secret://gito/webhooks/nomadcode")) || bytes.Contains(listRec.Body.Bytes(), []byte("secret_ref")) || bytes.Contains(listRec.Body.Bytes(), []byte("consumer.example.invalid")) { t.Fatalf("subscription list leaked private subscription input: %s", listRec.Body.String()) } } func TestWebhookSubscriptionValidatesTargetAndEvents(t *testing.T) { router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default()) req := httptest.NewRequest(http.MethodPost, "/api/webhook-subscriptions", bytes.NewBufferString(`{ "name": "invalid", "target_url": "file:///tmp/event", "events": ["branch.updated"] }`)) rec := httptest.NewRecorder() router.ServeHTTP(rec, req) if rec.Code != http.StatusBadRequest { t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String()) } req = httptest.NewRequest(http.MethodPost, "/api/webhook-subscriptions", bytes.NewBufferString(`{ "name": "missing-events", "target_url": "https://consumer.example.invalid/gito/events" }`)) rec = httptest.NewRecorder() router.ServeHTTP(rec, req) if rec.Code != http.StatusBadRequest { t.Fatalf("missing events status: got %d body=%s", rec.Code, rec.Body.String()) } } func TestEventSubscribeRegistersConnectionFilter(t *testing.T) { dispatcher := protosocket.NewDispatcher() runtime := NewRuntime(nil) subscriber := &recordingSubscriber{} registerProtoSocketHandlers(dispatcher, runtime, subscriber) response := dispatcher.Dispatch(t.Context(), protosocket.Envelope{ ID: "msg-1", Type: "request", Channel: "event", Action: "event.subscribe", Payload: map[string]any{ "events": []any{"branch.updated"}, "repo_id": "nomadcode", "branch": "develop", }, Meta: map[string]any{ "connection_id": "conn-1", }, }) if response.Type != "response" { t.Fatalf("response type: got %q error=%+v", response.Type, response.Error) } if subscriber.connectionID != "conn-1" { t.Fatalf("connection id: got %q", subscriber.connectionID) } if len(subscriber.subscription.Events) != 1 || subscriber.subscription.Events[0] != "branch.updated" { t.Fatalf("events: %#v", subscriber.subscription.Events) } if subscriber.subscription.RepoID != "nomadcode" || subscriber.subscription.Branch != "develop" { t.Fatalf("subscription: %+v", subscriber.subscription) } } func TestForgejoPushTreatsMissingEventHeaderAsPush(t *testing.T) { router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default()) registerBranchListener(t, router, `{"repo_id":"nomadcode","branch":"develop","provider":"forgejo"}`) // X-Forgejo-Event header missing but valid push payload should still be processed. req := httptest.NewRequest(http.MethodPost, "/callbacks/forgejo/push?repo_id=nomadcode", bytes.NewBufferString(`{ "ref": "refs/heads/develop", "before": "111", "after": "222", "repository": {"name": "nomadcode", "full_name": "toki/nomadcode"}, "commits": [{"id": "222", "modified": ["README.md"]}] }`)) req.Header.Set("X-Forgejo-Delivery", "delivery-missing-event") // Note: X-Forgejo-Event header is intentionally absent. rec := httptest.NewRecorder() router.ServeHTTP(rec, req) if rec.Code != http.StatusAccepted { t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String()) } var pushed map[string]any if err := json.NewDecoder(rec.Body).Decode(&pushed); err != nil { t.Fatalf("decode push response: %v", err) } if pushed["matched"] != true { t.Fatalf("expected push to match listener with missing event header: %#v", pushed) } } func TestForgejoPushRejectsMalformedPayload(t *testing.T) { router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default()) req := httptest.NewRequest(http.MethodPost, "/callbacks/forgejo/push?repo_id=nomadcode", bytes.NewBufferString(`{malformed json`)) req.Header.Set("X-Forgejo-Event", "push") req.Header.Set("X-Forgejo-Delivery", "delivery-malformed") rec := httptest.NewRecorder() router.ServeHTTP(rec, req) if rec.Code != http.StatusBadRequest { t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String()) } } func TestForgejoPushReturns400WhenNoRepoIDSource(t *testing.T) { router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default()) // Valid payload but repository has no name or full_name. req := httptest.NewRequest(http.MethodPost, "/callbacks/forgejo/push", bytes.NewBufferString(`{ "ref": "refs/heads/develop", "before": "111", "after": "222", "repository": {}, "commits": [{"id": "222"}] }`)) req.Header.Set("X-Forgejo-Event", "push") req.Header.Set("X-Forgejo-Delivery", "delivery-no-repo") rec := httptest.NewRecorder() router.ServeHTTP(rec, req) if rec.Code != http.StatusBadRequest { t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String()) } } func TestEventSubscribeRequiresConnectionID(t *testing.T) { dispatcher := protosocket.NewDispatcher() runtime := NewRuntime(nil) registerProtoSocketHandlers(dispatcher, runtime, &recordingSubscriber{}) response := dispatcher.Dispatch(t.Context(), protosocket.Envelope{ ID: "msg-1", Type: "request", Channel: "event", Action: "event.subscribe", }) if response.Type != "error" || response.Error == nil || response.Error.Code != "event.missing_connection" { t.Fatalf("expected missing connection error, got %+v", response) } } func registerBranchListener(t *testing.T, router http.Handler, body string) { t.Helper() req := httptest.NewRequest(http.MethodPost, "/api/listeners/branches", bytes.NewBufferString(body)) rec := httptest.NewRecorder() router.ServeHTTP(rec, req) if rec.Code != http.StatusCreated { t.Fatalf("listener status: got %d body=%s", rec.Code, rec.Body.String()) } } type recordingSubscriber struct { connectionID string subscription protosocket.EventSubscription } func (r *recordingSubscriber) Subscribe(connectionID string, subscription protosocket.EventSubscription) protosocket.EventSubscription { r.connectionID = connectionID r.subscription = subscription.Normalized() return r.subscription }