Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247package appview
import ( "context" "database/sql" "encoding/json" "errors" "log/slog" "path/filepath" "testing"
"github.com/bluesky-social/indigo/atproto/syntax" jmodels "github.com/bluesky-social/jetstream/pkg/models" "tangled.org/core/api/tangled" "tangled.org/core/appview/db" "tangled.org/core/appview/knotacl" "tangled.org/core/appview/models" "tangled.org/core/orm")
func newStateIngester(t *testing.T) *Ingester { t.Helper() path := filepath.Join(t.TempDir(), "test.db") d, err := db.Make(t.Context(), path) if err != nil { t.Fatalf("db.Make: %v", err) } t.Cleanup(func() { d.Close() }) return &Ingester{ Ctx: t.Context(), Db: d, Logger: slog.New(slog.DiscardHandler), }}
type stubAcl struct { allow bool err error}
func (s stubAcl) HasRepoPermissionErr(ctx context.Context, repo *models.Repo, userDid, perm string) (bool, error) { return s.allow, s.err}
func seedTx(t *testing.T, d *db.DB, fn func(tx *sql.Tx) error) { t.Helper() tx, err := d.Begin() if err != nil { t.Fatalf("Begin: %v", err) } defer tx.Rollback() if err := fn(tx); err != nil { t.Fatalf("seed: %v", err) } if err := tx.Commit(); err != nil { t.Fatalf("Commit: %v", err) }}
func seedRepo(t *testing.T, d *db.DB, ownerDid, repoDid string) { t.Helper() seedTx(t, d, func(tx *sql.Tx) error { return db.AddRepo(tx, &models.Repo{ Did: ownerDid, Name: "anemone", Knot: "knot.example", Rkey: "anemone", RepoDid: repoDid, }) })}
func seedIssue(t *testing.T, d *db.DB, ownerDid, repoDid, issueRkey string) syntax.ATURI { t.Helper() issue := &models.Issue{ Did: ownerDid, Rkey: issueRkey, RepoDid: syntax.DID(repoDid), Title: "title", Body: "body", Open: true, } seedTx(t, d, func(tx *sql.Tx) error { return db.PutIssue(tx, issue) }) return issue.AtUri()}
func seedRepoAndIssue(t *testing.T, d *db.DB, ownerDid, repoDid, issueRkey string) syntax.ATURI { t.Helper() seedRepo(t, d, ownerDid, repoDid) return seedIssue(t, d, ownerDid, repoDid, issueRkey)}
func issueStateEvent(t *testing.T, op, did, rkey, subject, state, createdAt string) *jmodels.Event { t.Helper() raw, err := json.Marshal(tangled.RepoIssueState{ Issue: subject, State: state, CreatedAt: createdAt, }) if err != nil { t.Fatalf("marshal: %v", err) } return &jmodels.Event{ Did: did, Kind: jmodels.EventKindCommit, Commit: &jmodels.Commit{ Operation: op, Collection: tangled.RepoIssueStateNSID, RKey: rkey, Record: raw, }, }}
func ingestedIssueOpen(t *testing.T, d *db.DB, subject syntax.ATURI) bool { t.Helper() issues, err := db.GetIssues(d, orm.FilterEq("at_uri", subject)) if err != nil || len(issues) != 1 { t.Fatalf("GetIssues: %v len %d", err, len(issues)) } return issues[0].Open}
func TestIngestState_Authorization(t *testing.T) { owner := "did:plc:boltless" cases := []struct { name string author string acl RepoPermissionChecker wantOpen bool }{ {"author closes own issue", owner, stubAcl{err: errors.New("acl must not be consulted for the author")}, false}, {"collaborator with push closes", "did:plc:squid", stubAcl{allow: true}, false}, {"stranger without push rejected", "did:plc:squid", stubAcl{allow: false}, true}, {"unreachable knot fails open", "did:plc:squid", stubAcl{allow: false, err: knotacl.ErrKnotUnreachable}, false}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { ing := newStateIngester(t) ing.Acl = tc.acl at := seedRepoAndIssue(t, ing.Db, owner, "did:plc:anemone", "issue1")
ev := issueStateEvent(t, jmodels.CommitOperationCreate, tc.author, "s1", string(at), tangled.RepoIssueStateClosed, "2026-06-01T00:00:00Z") if err := ing.ingestState(context.Background(), ev, ing.Logger, issueStateSpec); err != nil { t.Fatalf("ingestState: %v", err) }
if got := ingestedIssueOpen(t, ing.Db, at); got != tc.wantOpen { t.Fatalf("issue open = %v, want %v", got, tc.wantOpen) } if pending, _ := db.PendingStateRecordsForSubject(ing.Db, at); len(pending) != 0 { t.Fatalf("a record whose subject exists must resolve, not park: pending=%d", len(pending)) } }) }}
func TestIngestState_ParkThenReconcileDrains(t *testing.T) { ing := newStateIngester(t) owner := "did:plc:boltless" subject := syntax.ATURI("at://" + owner + "/" + tangled.RepoIssueNSID + "/issue1")
ev := issueStateEvent(t, jmodels.CommitOperationCreate, owner, "s1", string(subject), tangled.RepoIssueStateClosed, "2026-06-01T00:00:00Z") if err := ing.ingestState(ing.Ctx, ev, ing.Logger, issueStateSpec); err != nil { t.Fatalf("park: %v", err) } if pending, _ := db.PendingStateRecordsForSubject(ing.Db, subject); len(pending) != 1 { t.Fatalf("a state record whose subject is missing must be parked, pending=%d", len(pending)) }
at := seedRepoAndIssue(t, ing.Db, owner, "did:plc:anemone", "issue1") if at != subject { t.Fatalf("subject mismatch: %s vs %s", at, subject) } ing.ReconcilePendingState()
if ingestedIssueOpen(t, ing.Db, at) { t.Fatal("the reconciler must drain a parked record once its subject exists") } if pending, _ := db.PendingStateRecordsForSubject(ing.Db, at); len(pending) != 0 { t.Fatalf("a drained record must be unparked, pending=%d", len(pending)) }}
func TestIngestState_DeleteUnparksBeforeSubjectArrives(t *testing.T) { ing := newStateIngester(t) ctx := context.Background() owner := "did:plc:boltless" subject := syntax.ATURI("at://" + owner + "/" + tangled.RepoIssueNSID + "/issue1")
create := issueStateEvent(t, jmodels.CommitOperationCreate, owner, "s1", string(subject), tangled.RepoIssueStateClosed, "2026-06-01T00:00:00Z") if err := ing.ingestState(ctx, create, ing.Logger, issueStateSpec); err != nil { t.Fatalf("park: %v", err) }
del := &jmodels.Event{ Did: owner, Kind: jmodels.EventKindCommit, Commit: &jmodels.Commit{ Operation: jmodels.CommitOperationDelete, Collection: tangled.RepoIssueStateNSID, RKey: "s1", }, } if err := ing.ingestState(ctx, del, ing.Logger, issueStateSpec); err != nil { t.Fatalf("delete: %v", err) }
if pending, _ := db.PendingStateRecordsForSubject(ing.Db, subject); len(pending) != 0 { t.Fatalf("deleting a parked record must unpark it, pending=%d", len(pending)) }
at := seedRepoAndIssue(t, ing.Db, owner, "did:plc:anemone", "issue1") ing.drainPendingState(ctx, at, ing.Logger) if !ingestedIssueOpen(t, ing.Db, at) { t.Fatal("a deleted parked record must not apply after the subject arrives") }}
func TestIngestState_ParkedUnauthorizedDroppedOnDrain(t *testing.T) { ing := newStateIngester(t) ing.Acl = stubAcl{allow: false} ctx := context.Background() owner := "did:plc:boltless" subject := syntax.ATURI("at://" + owner + "/" + tangled.RepoIssueNSID + "/issue1")
ev := issueStateEvent(t, jmodels.CommitOperationCreate, "did:plc:squid", "s1", string(subject), tangled.RepoIssueStateClosed, "2026-06-01T00:00:00Z") if err := ing.ingestState(ctx, ev, ing.Logger, issueStateSpec); err != nil { t.Fatalf("park: %v", err) } if pending, _ := db.PendingStateRecordsForSubject(ing.Db, subject); len(pending) != 1 { t.Fatalf("a record whose subject is missing parks before any auth check: pending=%d", len(pending)) }
at := seedRepoAndIssue(t, ing.Db, owner, "did:plc:anemone", "issue1") ing.drainPendingState(ctx, at, ing.Logger)
if !ingestedIssueOpen(t, ing.Db, at) { t.Fatal("a parked record that fails authorization on drain must not apply") } if pending, _ := db.PendingStateRecordsForSubject(ing.Db, at); len(pending) != 0 { t.Fatalf("a parked record rejected on drain must be unparked, not left to re-accumulate: pending=%d", len(pending)) }}