From c2391cb2f22526bcb0beb1b84c7764084436f9e7 Mon Sep 17 00:00:00 2001 From: Aly Raffauf Date: Fri, 31 Jul 2026 13:47:03 -0400 Subject: [PATCH] spindle: add tests --- spindle/ci_subscribe_pipeline_logs_test.go | 125 +++++++++++++++++++-- spindle/client_test.go | 38 +++++++ 2 files changed, 153 insertions(+), 10 deletions(-) diff --git a/spindle/ci_subscribe_pipeline_logs_test.go b/spindle/ci_subscribe_pipeline_logs_test.go index 66151fa..8bbc1ba 100644 --- a/spindle/ci_subscribe_pipeline_logs_test.go +++ b/spindle/ci_subscribe_pipeline_logs_test.go @@ -2,27 +2,39 @@ package spindle import ( "bytes" + "context" + "net/http" + "net/http/httptest" "testing" + "time" + + "github.com/gorilla/websocket" ) func cborString(s string) []byte { - var buf bytes.Buffer l := len(s) - if l < 24 { - buf.WriteByte(0x60 | byte(l)) - } else { - buf.WriteByte(0x78) - buf.WriteByte(byte(l)) + switch { + case l < 24: + return append([]byte{0x60 | byte(l)}, s...) + case l < 256: + return append([]byte{0x78, byte(l)}, s...) + default: + return append([]byte{0x79, byte(l), byte(l >> 8)}, s...) } - buf.WriteString(s) - return buf.Bytes() } func cborInt(n int) []byte { if n >= 0 { - return []byte{byte(n)} + if n < 24 { + return []byte{byte(n)} + } + return []byte{0x18, byte(n)} } - return []byte{0x20 | byte(-n-1)} + v := -n - 1 + if v < 24 { + return []byte{0x20 | byte(v)} + } + return []byte{0x38, byte(v)} } func cborMap(pairs ...[2][]byte) []byte { @@ -103,3 +115,96 @@ func TestDecodeLogEventUnknownType(t *testing.T) { t.Fatalf("decodeLogEvent() = %+v, want nil for unknown type", event) } } + +func TestSubscribePipelineLogsStreamsEventsAndCloses(t *testing.T) { + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/xrpc/"+nsidSubscribeLogs.String() { + t.Fatalf("path = %q", request.URL.Path) + } + if request.URL.Query().Get("pipeline") != "3mrvk5dbnep22" { + t.Fatalf("pipeline query = %q", request.URL.Query().Get("pipeline")) + } + if workflows := request.URL.Query()["workflows"]; len(workflows) != 2 || workflows[0] != "build.yml" || workflows[1] != "test.yml" { + t.Fatalf("workflows query = %v", request.URL.Query()["workflows"]) + } + conn, err := upgrader.Upgrade(writer, request, nil) + if err != nil { + t.Fatalf("upgrade: %v", err) + } + defer conn.Close() + header := cborMap( + [2][]byte{cborString("t"), cborString("#data")}, + [2][]byte{cborString("op"), cborInt(1)}, + ) + body := cborMap( + [2][]byte{cborString("step"), cborInt(0)}, + [2][]byte{cborString("time"), cborString("2026-07-31T06:34:10+03:00")}, + [2][]byte{cborString("stream"), cborString("stdout")}, + [2][]byte{cborString("content"), cborString("building")}, + [2][]byte{cborString("workflow"), cborString("build.yml")}, + ) + if err := conn.WriteMessage(websocket.BinaryMessage, append(header, body...)); err != nil { + t.Fatalf("write data frame: %v", err) + } + if err := conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""), time.Now().Add(time.Second)); err != nil { + t.Fatalf("write close frame: %v", err) + } + })) + defer server.Close() + + client, err := New(server.URL, server.Client()) + if err != nil { + t.Fatalf("New() error = %v", err) + } + var events []PipelineLogEvent + if err := client.SubscribePipelineLogs(context.Background(), "3mrvk5dbnep22", []string{"build.yml", "test.yml"}, func(event PipelineLogEvent) error { + events = append(events, event) + return nil + }); err != nil { + t.Fatalf("SubscribePipelineLogs() error = %v", err) + } + if len(events) != 1 || events[0].Data == nil || events[0].Data.Content != "building" { + t.Fatalf("events = %+v", events) + } +} + +func TestSubscribePipelineLogsHonorsCancellation(t *testing.T) { + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + upgraded := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + conn, err := upgrader.Upgrade(writer, request, nil) + if err != nil { + t.Fatalf("upgrade: %v", err) + } + defer conn.Close() + close(upgraded) + for { + if _, _, err := conn.ReadMessage(); err != nil { + return + } + } + })) + defer server.Close() + + client, err := New(server.URL, server.Client()) + if err != nil { + t.Fatalf("New() error = %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + done <- client.SubscribePipelineLogs(ctx, "3mrvk5dbnep22", nil, func(PipelineLogEvent) error { return nil }) + }() + <-upgraded + cancel() + select { + case err := <-done: + if err != context.Canceled { + t.Fatalf("SubscribePipelineLogs() error = %v, want context.Canceled", err) + } + case <-time.After(5 * time.Second): + t.Fatal("SubscribePipelineLogs() did not return after cancellation") + } +} diff --git a/spindle/client_test.go b/spindle/client_test.go index a2299a0..c36e8c7 100644 --- a/spindle/client_test.go +++ b/spindle/client_test.go @@ -121,3 +121,41 @@ func TestQueryLatestPipelineLimitsResults(t *testing.T) { t.Fatalf("QueryLatestPipeline() error = %v", err) } } + +func TestGetPipelineQueriesAndDecodes(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/xrpc/"+nsidGetPipeline.String() { + t.Fatalf("path = %q", request.URL.Path) + } + if request.URL.Query().Get("pipeline") != "3mrvk5dbnep22" { + t.Fatalf("pipeline query = %q", request.URL.Query().Get("pipeline")) + } + writer.Header().Set("Content-Type", "application/json") + _, _ = writer.Write([]byte(`{"id":"3mrvk5dbnep22","commit":"abc","trigger":{"$type":"sh.tangled.ci.trigger#push"},"workflows":[{"id":"test","name":"test","status":"success"}]}`)) + })) + defer server.Close() + + client, err := New(server.URL, server.Client()) + if err != nil { + t.Fatalf("New() error = %v", err) + } + pipeline, err := client.GetPipeline(context.Background(), "3mrvk5dbnep22") + if err != nil { + t.Fatalf("GetPipeline() error = %v", err) + } + if pipeline.ID != "3mrvk5dbnep22" || pipeline.Commit != "abc" || len(pipeline.Workflows) != 1 { + t.Fatalf("pipeline = %+v", pipeline) + } +} + +func TestServiceDID(t *testing.T) { + if got, err := ServiceDID("spindle.example"); err != nil || got != "did:web:spindle.example" { + t.Fatalf("ServiceDID() = %q, %v", got, err) + } + if got, err := ServiceDID("https://spindle.example:443"); err != nil || got != "did:web:spindle.example%3A443" { + t.Fatalf("ServiceDID(port) = %q, %v", got, err) + } + if _, err := ServiceDID(""); err == nil { + t.Fatal("ServiceDID(empty) error = nil") + } +} -- 2.51.2