Something went wrong. Try again.
Go boilerplate library for building atproto apps
atproto go
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334package jetstream
import ( "context" "encoding/json" "testing"
"github.com/klauspost/compress/zstd")
func TestNew(t *testing.T) { c := New(&Config{ WantedCollections: []string{"app.bsky.feed.post"}, }, func(ctx context.Context, evt *Event) error { return nil }) if c == nil { t.Fatal("expected non-nil consumer") } if c.IsConnected() { t.Fatal("should not be connected before Start") }}
func TestConfig_Defaults(t *testing.T) { cfg := &Config{} if eps := cfg.endpoints(); len(eps) == 0 { t.Fatal("expected default endpoints") } if cfg.cursorPersistEvery() != 1000 { t.Fatalf("expected 1000, got %d", cfg.cursorPersistEvery()) }}
func TestConfig_CustomEndpoints(t *testing.T) { cfg := &Config{Endpoints: []string{"wss://custom.example.com/subscribe"}} eps := cfg.endpoints() if len(eps) != 1 || eps[0] != "wss://custom.example.com/subscribe" { t.Fatalf("unexpected endpoints: %v", eps) }}
func TestBuildURL_Collections(t *testing.T) { c := New(&Config{ Endpoints: []string{"wss://jetstream1.us-east.bsky.network/subscribe"}, WantedCollections: []string{"app.bsky.feed.post", "app.bsky.feed.like"}, }, func(ctx context.Context, evt *Event) error { return nil })
u, err := c.buildURL("wss://jetstream1.us-east.bsky.network/subscribe") if err != nil { t.Fatal(err) } if u == "" { t.Fatal("expected non-empty URL") } // Should contain wantedCollections params if !contains(u, "wantedCollections=app.bsky.feed.post") { t.Errorf("URL missing wantedCollections: %s", u) }}
func TestBuildURL_Cursor(t *testing.T) { c := New(&Config{ Endpoints: []string{"wss://jetstream1.us-east.bsky.network/subscribe"}, }, func(ctx context.Context, evt *Event) error { return nil })
c.cursor.Store(1000000) u, err := c.buildURL("wss://jetstream1.us-east.bsky.network/subscribe") if err != nil { t.Fatal(err) } if !contains(u, "cursor=") { t.Errorf("URL missing cursor: %s", u) }}
func TestProcess_ValidEvent(t *testing.T) { var received *Event c := New(&Config{}, func(ctx context.Context, evt *Event) error { received = evt return nil })
evt := Event{ DID: "did:plc:test", TimeUS: 1234567890, Kind: "commit", Commit: &Commit{ Operation: "create", Collection: "app.bsky.feed.post", RKey: "abc123", }, } data, _ := json.Marshal(evt)
if err := c.process(context.Background(), data); err != nil { t.Fatal(err) } if received == nil { t.Fatal("expected handler to be called") } if received.DID != "did:plc:test" { t.Fatalf("got DID %q", received.DID) }}
func TestProcess_UpdatesCursor(t *testing.T) { c := New(&Config{}, func(ctx context.Context, evt *Event) error { return nil })
data, _ := json.Marshal(Event{DID: "did:plc:test", TimeUS: 9999999}) c.process(context.Background(), data)
if c.cursor.Load() != 9999999 { t.Fatalf("cursor not updated: %d", c.cursor.Load()) }}
func TestProcess_InvalidJSON(t *testing.T) { c := New(&Config{}, func(ctx context.Context, evt *Event) error { return nil }) err := c.process(context.Background(), []byte("{bad json")) if err == nil { t.Fatal("expected error for invalid JSON") }}
func TestProcess_CompressedRoundTrip(t *testing.T) { var received *Event c := New(&Config{Compress: true}, func(ctx context.Context, evt *Event) error { received = evt return nil })
enc, err := zstd.NewWriter(nil, zstd.WithEncoderDict(zstdDictionary)) if err != nil { t.Fatalf("create encoder: %v", err) } defer enc.Close()
evt := Event{ DID: "did:plc:test", TimeUS: 1234567890, Kind: "commit", Commit: &Commit{ Operation: "create", Collection: "app.bsky.feed.post", RKey: "abc123", }, } raw, _ := json.Marshal(evt) compressed := enc.EncodeAll(raw, nil)
if len(compressed) == 0 { t.Fatal("expected non-empty compressed payload") } if err := c.process(context.Background(), compressed); err != nil { t.Fatalf("process: %v", err) } if received == nil { t.Fatal("expected handler to be called") } if received.DID != "did:plc:test" { t.Fatalf("got DID %q", received.DID) } if received.Commit == nil || received.Commit.RKey != "abc123" { t.Fatalf("commit not round-tripped: %+v", received.Commit) }}
func TestProcess_CompressedGarbage(t *testing.T) { c := New(&Config{Compress: true}, func(ctx context.Context, evt *Event) error { return nil })
if err := c.process(context.Background(), []byte("not zstd")); err == nil { t.Fatal("expected error decompressing garbage") }}
func TestEvent_IdentityRoundTrip(t *testing.T) { raw := []byte(`{"did":"did:plc:test","time_us":1,"kind":"identity","identity":{"did":"did:plc:test","handle":"alice.bsky.social","seq":42,"time":"2026-01-01T00:00:00Z"}}`) var evt Event if err := json.Unmarshal(raw, &evt); err != nil { t.Fatal(err) } if evt.Identity == nil { t.Fatal("expected identity payload") } if evt.Identity.Handle != "alice.bsky.social" { t.Fatalf("got handle %q", evt.Identity.Handle) } if evt.Identity.Seq != 42 { t.Fatalf("got seq %d", evt.Identity.Seq) }}
func TestEvent_AccountRoundTrip(t *testing.T) { raw := []byte(`{"did":"did:plc:test","time_us":1,"kind":"account","account":{"active":false,"did":"did:plc:test","seq":7,"status":"deleted","time":"2026-01-01T00:00:00Z"}}`) var evt Event if err := json.Unmarshal(raw, &evt); err != nil { t.Fatal(err) } if evt.Account == nil { t.Fatal("expected account payload") } if evt.Account.Active { t.Fatal("expected active=false") } if evt.Account.Status != "deleted" { t.Fatalf("got status %q", evt.Account.Status) }}
func TestBuildURL_WantedDIDs(t *testing.T) { c := New(&Config{ Endpoints: []string{"wss://example.com/subscribe"}, WantedDIDs: []string{"did:plc:alice", "did:plc:bob"}, }, func(ctx context.Context, evt *Event) error { return nil })
u, err := c.buildURL("wss://example.com/subscribe") if err != nil { t.Fatal(err) } if !contains(u, "wantedDids=did%3Aplc%3Aalice") { t.Errorf("URL missing wantedDids=alice: %s", u) } if !contains(u, "wantedDids=did%3Aplc%3Abob") { t.Errorf("URL missing wantedDids=bob: %s", u) }}
func TestUpdateOptions_NotConnected(t *testing.T) { c := New(&Config{ Endpoints: []string{"wss://example.com/subscribe"}, WantedCollections: []string{"app.bsky.feed.post"}, }, func(ctx context.Context, evt *Event) error { return nil })
// Not connected — UpdateOptions should succeed and apply on next reconnect. if err := c.UpdateOptions([]string{"app.bsky.feed.like"}, []string{"did:plc:alice"}); err != nil { t.Fatalf("UpdateOptions: %v", err) }
u, err := c.buildURL("wss://example.com/subscribe") if err != nil { t.Fatal(err) } if !contains(u, "wantedCollections=app.bsky.feed.like") { t.Errorf("URL missing new wantedCollections: %s", u) } if contains(u, "wantedCollections=app.bsky.feed.post") { t.Errorf("URL still contains old wantedCollections: %s", u) } if !contains(u, "wantedDids=did%3Aplc%3Aalice") { t.Errorf("URL missing new wantedDids: %s", u) }}
func TestUpdateOptions_NilSlicesClear(t *testing.T) { c := New(&Config{ Endpoints: []string{"wss://example.com/subscribe"}, WantedCollections: []string{"app.bsky.feed.post"}, WantedDIDs: []string{"did:plc:alice"}, }, func(ctx context.Context, evt *Event) error { return nil })
if err := c.UpdateOptions(nil, nil); err != nil { t.Fatalf("UpdateOptions: %v", err) }
u, err := c.buildURL("wss://example.com/subscribe") if err != nil { t.Fatal(err) } if contains(u, "wantedCollections=") { t.Errorf("URL still has wantedCollections after clear: %s", u) } if contains(u, "wantedDids=") { t.Errorf("URL still has wantedDids after clear: %s", u) }}
func TestConfig_PingDefaults(t *testing.T) { cfg := &Config{} if got := cfg.pingInterval(); got != defaultPingInterval { t.Fatalf("expected default ping %v, got %v", defaultPingInterval, got) }
cfg.PingInterval = 5 * 1_000_000_000 // 5s in ns if got := cfg.pingInterval(); got != cfg.PingInterval { t.Fatalf("expected configured ping, got %v", got) }
cfg.PingInterval = -1 if got := cfg.pingInterval(); got != 0 { t.Fatalf("expected 0 for disabled, got %v", got) }}
func TestConfig_ReadDeadlineDefaults(t *testing.T) { cfg := &Config{} if got := cfg.readDeadline(); got != defaultReadDeadline { t.Fatalf("expected default deadline %v, got %v", defaultReadDeadline, got) }
cfg.ReadDeadline = 1 if got := cfg.readDeadline(); got != 1 { t.Fatalf("expected configured deadline, got %v", got) }}
func TestCurrentEndpoint_EmptyBeforeConnect(t *testing.T) { c := New(&Config{}, func(ctx context.Context, evt *Event) error { return nil }) if ep := c.CurrentEndpoint(); ep != "" { t.Fatalf("expected empty endpoint, got %q", ep) }}
func TestStats(t *testing.T) { c := New(&Config{}, func(ctx context.Context, evt *Event) error { return nil }) evts, bytes := c.Stats() if evts != 0 || bytes != 0 { t.Fatalf("expected zero stats, got events=%d bytes=%d", evts, bytes) }}
func contains(s, substr string) bool { return len(s) >= len(substr) && (s == substr || len(s) > 0 && containsStr(s, substr))}
func containsStr(s, substr string) bool { for i := 0; i <= len(s)-len(substr); i++ { if s[i:i+len(substr)] == substr { return true } } return false}