package cilogs
import (
"bytes"
"context"
"encoding/json"
"errors"
"io"
"log/slog"
"net"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
"github.com/bluesky-social/indigo/atproto/identity"
"github.com/bluesky-social/indigo/atproto/syntax"
"github.com/gorilla/websocket"
gossh "golang.org/x/crypto/ssh"
"tangled.org/core/api/org_tangled"
"tangled.org/core/repoident"
)
const (
testRepo = "did:plc:cilogsrepo"
testSha = "4f1c2b3a4f1c2b3a4f1c2b3a4f1c2b3a4f1c2b3a"
)
func fakeSpindle(t *testing.T, pipeline map[string]any) *httptest.Server {
upgrader := websocket.Upgrader{}
mux := http.NewServeMux()
reply := func(w http.ResponseWriter, v any) {
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(v)
}
mux.HandleFunc("/xrpc/org.tangled.ci.queryPipelines", func(w http.ResponseWriter, r *http.Request) {
reply(w, map[string]any{"pipelines": []any{pipeline}})
})
mux.HandleFunc("/xrpc/org.tangled.ci.getPipeline", func(w http.ResponseWriter, r *http.Request) {
reply(w, pipeline)
})
mux.HandleFunc("/xrpc/org.tangled.ci.subscribePipelineLogs", func(w http.ResponseWriter, r *http.Request) {
if r.URL.Query().Get("repo") != testRepo {
http.Error(w, `{"error":"InvalidRequest","message":"repo is required"}`, http.StatusBadRequest)
return
}
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
return
}
defer conn.Close()
status := func(s string) *string { return &s }
now := time.Now().Format(time.RFC3339)
for _, ev := range []org_tangled.CiSubscribePipelineLogs_Event{
{Control: &org_tangled.CiSubscribePipelineLogs_Control{Time: now, Workflow: "build", Step: 0, Content: "compile", Status: status("start")}},
{Data: &org_tangled.CiSubscribePipelineLogs_Data{Time: now, Workflow: "build", Step: 0, Content: "hello from the build\n", Stream: "stdout"}},
} {
var buf bytes.Buffer
if err := ev.Serialize(&buf); err != nil {
t.Error(err)
return
}
conn.WriteMessage(websocket.BinaryMessage, buf.Bytes())
}
<-r.Context().Done()
})
srv := httptest.NewServer(mux)
t.Cleanup(srv.Close)
return srv
}
func fakeKnot(t *testing.T, spindleDid string) *httptest.Server {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/xrpc/com.atproto.repo.getRecord" || r.URL.Query().Get("collection") != org_tangled.RepoManifestNSID {
http.Error(w, `{"error":"RecordNotFound"}`, http.StatusNotFound)
return
}
json.NewEncoder(w).Encode(map[string]any{"value": map[string]any{
"$type": org_tangled.RepoManifestNSID,
"declaration": map[string]any{"owner": "did:plc:owner", "name": "proj"},
"spindle": spindleDid,
"createdAt": "2026-10-01T09:12:44Z",
}})
}))
t.Cleanup(srv.Close)
return srv
}
func directoryFor(knot *httptest.Server) identity.Directory {
dir := identity.NewMockDirectory()
dir.Insert(identity.Identity{DID: testRepo, Services: map[string]identity.ServiceEndpoint{
repoident.KnotServiceID: {Type: repoident.KnotServiceType, URL: knot.URL},
}})
return dir
}
func freeAddr(t *testing.T) string {
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
defer ln.Close()
return ln.Addr().String()
}
func dialSSH(t *testing.T, addr string) *gossh.Client {
deadline := time.Now().Add(5 * time.Second)
for {
c, err := gossh.Dial("tcp", addr, &gossh.ClientConfig{User: "anyone", HostKeyCallback: gossh.InsecureIgnoreHostKey()})
if err == nil {
return c
}
if time.Now().After(deadline) {
t.Fatal(err)
}
time.Sleep(50 * time.Millisecond)
}
}
type syncBuffer struct {
mu sync.Mutex
buf bytes.Buffer
}
func (b *syncBuffer) Write(p []byte) (int, error) {
b.mu.Lock()
defer b.mu.Unlock()
return b.buf.Write(p)
}
func (b *syncBuffer) String() string {
b.mu.Lock()
defer b.mu.Unlock()
return b.buf.String()
}
func TestFollowsAWorkflowsLogsFromTheSpindleInTheRepoManifest(t *testing.T) {
name := "build"
pipeline := map[string]any{
"id": syntax.NewTIDNow(0).String(),
"repo": testRepo,
"commit": testSha,
"trigger": map[string]any{
"$type": "org.tangled.event.push", "ref": "refs/heads/main", "newSha": testSha,
"oldSha": strings.Repeat("0", 40), "repo": testRepo,
},
"workflows": []any{map[string]any{"id": name, "name": name, "status": "running", "startedAt": time.Now().Format(time.RFC3339)}},
}
spindle := fakeSpindle(t, pipeline)
spindleDid := "did:web:localhost%3A" + spindle.URL[strings.LastIndex(spindle.URL, ":")+1:]
knot := fakeKnot(t, spindleDid)
dir := directoryFor(knot)
addr := freeAddr(t)
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
srv := New(Config{ListenAddr: addr, HostKeyPath: t.TempDir() + "/host_key"}, NewSpindles(dir, true), slog.New(slog.NewTextHandler(io.Discard, nil)))
go srv.ListenAndServe(ctx)
client := dialSSH(t, addr)
defer client.Close()
sess, err := client.NewSession()
if err != nil {
t.Fatal(err)
}
defer sess.Close()
if err := sess.RequestPty("xterm-256color", 40, 160, gossh.TerminalModes{}); err != nil {
t.Fatal(err)
}
var out syncBuffer
sess.Stdout = &out
sess.Stderr = &out
stdin, err := sess.StdinPipe()
if err != nil {
t.Fatal(err)
}
if err := sess.Start(testRepo + " " + testSha); err != nil {
t.Fatal(err)
}
// wish asks the terminal for its background colour before drawing anything
stdin.Write([]byte("\x1b]11;rgb:0000/0000/0000\x07\x1b[?62c"))
deadline := time.Now().Add(10 * time.Second)
for !strings.Contains(out.String(), "hello from the build") {
if strings.Contains(out.String(), "error:") || time.Now().After(deadline) {
t.Fatalf("never saw the log line, terminal got:\n%s", out.String())
}
time.Sleep(50 * time.Millisecond)
}
if !strings.Contains(out.String(), "refs/heads/main") {
t.Errorf("the top bar doesn't name the pushed ref:\n%s", out.String())
}
stdin.Write([]byte("q"))
}
func TestSpindleURL(t *testing.T) {
for did, want := range map[string]string{
"did:web:spindle.example.com": "https://spindle.example.com",
"did:web:localhost%3A6555": "http://localhost:6555",
"did:web:host:with:a:path": "",
"did:plc:abc": "",
} {
got, err := spindleURL(did)
if (err != nil) != (want == "") || got != want {
t.Errorf("spindleURL(%q) = %q, %v, want %q", did, got, err, want)
}
}
}
func TestNoSpindleInTheManifestIsReportedAsSuch(t *testing.T) {
_, err := NewSpindles(directoryFor(fakeKnot(t, "")), true).Of(context.Background(), testRepo)
if !errors.Is(err, ErrNoSpindle) {
t.Errorf("err = %v, want ErrNoSpindle", err)
}
}
func TestRefusesLoopbackSpindlesOutsideDev(t *testing.T) {
srv := httptest.NewServer(http.NotFoundHandler())
defer srv.Close()
addr := strings.TrimPrefix(srv.URL, "http://")
for dev, wantErr := range map[bool]bool{false: true, true: false} {
conn, err := NewSpindles(identity.NewMockDirectory(), dev).dialer.DialContext(context.Background(), "tcp", addr)
if conn != nil {
conn.Close()
}
if (err != nil) != wantErr {
t.Errorf("dev=%v: dial err = %v, want error: %v", dev, err, wantErr)
}
}
}