Something went wrong. Try again.
An ATProto Powered Reputation Network
Something went wrong. Try again.
5.8 kB · 164 lines
Go
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165package microcosm
import ( "context" "encoding/json" "errors" "net/http" "net/http/httptest" "strconv" "strings" "testing"
"github.com/coder/websocket"
"provenance/internal/atproto" "provenance/internal/domain")
const author = "did:plc:alice"const subject = "did:plc:bob"const recordCID = "bafyreidtest"
func commit(t *testing.T, operation Operation, record any) []byte { t.Helper() value := map[string]any{"did": author, "time_us": int64(10000000), "kind": "commit", "commit": map[string]any{"operation": operation, "collection": domain.Collection, "rkey": domain.RecordKey(subject, "work"), "record": record, "cid": recordCID}} data, err := json.Marshal(value) if err != nil { t.Fatal(err) } return data}
func validRecord() atproto.Record { score := 2 return atproto.Record{Type: domain.Collection, Subject: subject, Tag: "work", Score: &score, CreatedAt: "2026-10-07T00:00:00Z"}}
func TestWebSocketStream(t *testing.T) { stop := errors.New("complete") server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Query().Get("wantedCollections") != domain.Collection || r.URL.Query().Get("cursor") != strconv.FormatInt(10000000-ReplayOverlap.Microseconds(), 10) { t.Error("wrong collection filter or replay cursor") } conn, err := websocket.Accept(w, r, nil) if err != nil { t.Error(err) return } defer conn.CloseNow() for _, operation := range []Operation{Create, Update, Delete} { if err := conn.Write(r.Context(), websocket.MessageText, commit(t, operation, validRecord())); err != nil { return } } conn.Read(r.Context()) })) defer server.Close() client, err := New("ws"+strings.TrimPrefix(server.URL, "http")+"/subscribe", server.URL) if err != nil { t.Fatal(err) } var changes []domain.Change err = client.Consume(context.Background(), 10000000, nil, func(_ context.Context, change domain.Change) error { changes = append(changes, change) if len(changes) == 3 { return stop } return nil }) if !errors.Is(err, stop) || len(changes) != 3 { t.Fatalf("received %d changes: %v", len(changes), err) } if changes[0].Vouch == nil || changes[0].Vouch.Score != 2 || changes[2].Vouch != nil { t.Fatalf("incorrect commit mapping: %+v", changes) }}
func TestSlingshotHydration(t *testing.T) { calls := 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { calls++ if r.URL.Path != "/xrpc/com.atproto.repo.getRecord" || r.URL.Query().Get("repo") != author || r.URL.Query().Get("cid") != recordCID { t.Error("wrong Slingshot query") } json.NewEncoder(w).Encode(map[string]any{"uri": "at://" + author + "/" + domain.Collection + "/" + domain.RecordKey(subject, "work"), "cid": recordCID, "value": validRecord()}) })) defer server.Close() client, err := New("ws://127.0.0.1:1/subscribe", server.URL) if err != nil { t.Fatal(err) } change, err := client.decode(context.Background(), commit(t, Create, nil), nil) if err != nil || change.Vouch == nil || calls != 1 { t.Fatalf("hydration: %+v %v", change, err) } change, err = client.decode(context.Background(), commit(t, Create, nil), func(context.Context, string, int64) (bool, error) { return false, nil }) if err != nil || change.Vouch != nil || calls != 1 { t.Fatal("non-participant triggered record hydration") }}
func TestInvalidUpdateRemovesAssertion(t *testing.T) { client, err := New(DefaultJetstreamURL, DefaultSlingshotURL) if err != nil { t.Fatal(err) } record := validRecord() score := domain.SelfScore record.Score = &score change, err := client.decode(context.Background(), commit(t, Update, record), nil) if err != nil || change.Kind != domain.RecordChange || change.Vouch != nil || change.RecordKey == "" { t.Fatalf("invalid update retained its assertion: %+v %v", change, err) } if _, err := client.decode(context.Background(), []byte(`{"kind":"commit"}`), nil); err == nil { t.Fatal("accepted malformed event envelope") }}
func TestNonCanonicalRecords(t *testing.T) { wrongSubject := validRecord() wrongSubject.Subject = "did:plc:carol" wrongTag := validRecord() wrongTag.Tag = "Work" for _, tt := range []struct { name string record atproto.Record }{ {"subject mismatch", wrongSubject}, {"tag mismatch", wrongTag}, } { t.Run(tt.name, func(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { key := domain.RecordKey(subject, "work") if r.URL.Query().Get("rkey") != key { t.Error("wrong hydration record key") } json.NewEncoder(w).Encode(map[string]any{"uri": "at://" + author + "/" + domain.Collection + "/" + key, "cid": recordCID, "value": tt.record}) })) defer server.Close() client, err := New("ws://127.0.0.1:1/subscribe", server.URL) if err != nil { t.Fatal(err) } for _, operation := range []Operation{Create, Update} { for _, payload := range []any{tt.record, nil} { change, err := client.decode(context.Background(), commit(t, operation, payload), nil) if err != nil || change.Vouch != nil || change.RecordKey != domain.RecordKey(subject, "work") || change.Kind != domain.RecordChange { t.Fatalf("non-canonical %s entered the index: %+v %v", operation, change, err) } } } }) }}
func TestEndpointValidation(t *testing.T) { for _, address := range []string{"ws://remote.example.com/subscribe", "wss://user:pass@example.com/subscribe", "wss://example.com/subscribe?compress=true"} { if _, err := New(address, DefaultSlingshotURL); err == nil { t.Fatalf("accepted %s", address) } } if _, err := New("wss://stream.internal.example/subscribe", "https://records.internal.example"); err != nil { t.Fatal("rejected configured self-hosted endpoints") }}