Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206package migration
import ( "context" "errors" "io" "log/slog" "path/filepath" "sync/atomic" "testing"
"github.com/bluesky-social/indigo/atproto/atclient" "github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/appview/db" "tangled.org/core/appview/models" "tangled.org/core/appview/oauth")
func newTestDB(t *testing.T) *db.DB { t.Helper() d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "test.db")) if err != nil { t.Fatalf("db.Make: %v", err) } t.Cleanup(func() { d.Close() }) return d}
func seedMigration(t *testing.T, d *db.DB, mig *models.PDSMigration) { t.Helper() if err := db.EnqueuePdsRecordMigration(context.Background(), d, mig.Name, mig.Did, mig.Collection, mig.Rkey); err != nil { t.Fatalf("EnqueuePdsRecordMigration: %v", err) }}
func fetch(t *testing.T, d *db.DB, did syntax.DID) *models.PDSMigration { t.Helper() rows, err := d.QueryContext(context.Background(), `select name, did, collection, rkey, status, error_msg, retry_count, retry_after from pds_migration where did = ?`, did) if err != nil { t.Fatalf("query: %v", err) } defer rows.Close() if !rows.Next() { t.Fatalf("no row for did %s", did) } var m models.PDSMigration if err := rows.Scan(&m.Name, &m.Did, &m.Collection, &m.Rkey, &m.Status, &m.ErrorMsg, &m.RetryCount, &m.RetryAfter); err != nil { t.Fatalf("scan: %v", err) } return &m}
func newTestMigration(t *testing.T, mig migrator, onPerm permAuthErrHandler) *Migration { t.Helper() return &Migration{ db: newTestDB(t), logger: slog.New(slog.NewTextHandler(io.Discard, nil)), sem: make(chan struct{}, maxConcurrentMigrations), migrators: map[string]migrator{"add-repo-did": mig}, onPermAuthErr: onPerm, }}
func newPDSMigration(did syntax.DID) *models.PDSMigration { return &models.PDSMigration{ Name: "add-repo-did", Did: did, Collection: "sh.tangled.repo", Rkey: "abc", Status: models.PDSMigrationStatusRunning, }}
func TestMigrateInvalidGrantMarksFailed(t *testing.T) { did := syntax.DID("did:plc:boltless") var permCalled atomic.Int32 m := newTestMigration(t, func(context.Context, *atclient.APIClient, syntax.DID, syntax.ATURI) error { return errors.New("put record: failed to refresh OAuth tokens: token refresh failed: auth server request failed (HTTP 400): invalid_grant") }, func(_ context.Context, _ syntax.DID, _ string, err error) bool { permCalled.Add(1) return oauth.IsPermanentAuthErr(err) }, ) seedMigration(t, m.db, newPDSMigration(did)) pm := newPDSMigration(did)
if err := m.migrate(context.Background(), &atclient.APIClient{}, "sess1", pm); err != nil { t.Fatalf("migrate: %v", err) }
got := fetch(t, m.db, did) if got.Status != models.PDSMigrationStatusFailed { t.Fatalf("status = %s, want failed", got.Status) } if got.RetryAfter != 0 { t.Fatalf("RetryAfter = %d, want 0", got.RetryAfter) } if permCalled.Load() != 1 { t.Fatalf("onPermAuthErr called %d times, want 1", permCalled.Load()) }}
func TestMigrateTransientErrorStaysPending(t *testing.T) { did := syntax.DID("did:plc:akshay") m := newTestMigration(t, func(context.Context, *atclient.APIClient, syntax.DID, syntax.ATURI) error { return errors.New("put record: failed to refresh OAuth tokens: token refresh failed (HTTP 429): rate_limited") }, func(_ context.Context, _ syntax.DID, _ string, err error) bool { return oauth.IsPermanentAuthErr(err) }, ) seedMigration(t, m.db, newPDSMigration(did)) pm := newPDSMigration(did)
if err := m.migrate(context.Background(), &atclient.APIClient{}, "sess1", pm); err != nil { t.Fatalf("migrate: %v", err) }
got := fetch(t, m.db, did) if got.Status != models.PDSMigrationStatusPending { t.Fatalf("status = %s, want pending", got.Status) } if got.RetryCount != 1 { t.Fatalf("RetryCount = %d, want 1", got.RetryCount) } if got.RetryAfter == 0 { t.Fatalf("RetryAfter not scheduled") }}
func TestMigrateSuccess(t *testing.T) { did := syntax.DID("did:plc:boltless") m := newTestMigration(t, func(context.Context, *atclient.APIClient, syntax.DID, syntax.ATURI) error { return nil }, func(context.Context, syntax.DID, string, error) bool { return false }, ) seedMigration(t, m.db, newPDSMigration(did)) pm := newPDSMigration(did)
if err := m.migrate(context.Background(), &atclient.APIClient{}, "sess1", pm); err != nil { t.Fatalf("migrate: %v", err) }
got := fetch(t, m.db, did) if got.Status != models.PDSMigrationStatusDone { t.Fatalf("status = %s, want done", got.Status) }}
func TestEnqueueResetsFailedToPending(t *testing.T) { d := newTestDB(t) did := syntax.DID("did:plc:boltless") seed := newPDSMigration(did) seedMigration(t, d, seed)
errMsg := "some prior failure" failed := *seed failed.Status = models.PDSMigrationStatusFailed failed.RetryCount = 7 failed.ErrorMsg = &errMsg if err := db.UpdatePdsRecordMigration(context.Background(), d, &failed); err != nil { t.Fatalf("UpdatePdsRecordMigration: %v", err) }
if err := db.EnqueuePdsRecordMigration(context.Background(), d, seed.Name, seed.Did, seed.Collection, seed.Rkey); err != nil { t.Fatalf("re-enqueue: %v", err) }
got := fetch(t, d, did) if got.Status != models.PDSMigrationStatusPending { t.Fatalf("status = %s, want pending", got.Status) } if got.RetryCount != 0 { t.Fatalf("RetryCount = %d, want 0", got.RetryCount) } if got.ErrorMsg != nil { t.Fatalf("ErrorMsg = %v, want nil", got.ErrorMsg) }}
func TestReapStaleRunning(t *testing.T) { d := newTestDB(t) did := syntax.DID("did:plc:akshay") seed := newPDSMigration(did) seedMigration(t, d, seed) running := *seed running.Status = models.PDSMigrationStatusRunning if err := db.UpdatePdsRecordMigration(context.Background(), d, &running); err != nil { t.Fatalf("update: %v", err) }
if err := db.ReapStaleRunningMigrations(context.Background(), d); err != nil { t.Fatalf("reap: %v", err) }
got := fetch(t, d, did) if got.Status != models.PDSMigrationStatusPending { t.Fatalf("status = %s, want pending", got.Status) }}