Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234package 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) } }}