272 lines
8.8 KiB
Go
272 lines
8.8 KiB
Go
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)
|
|
}
|
|
}
|
|
|
|
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 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 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
|
|
}
|