Something went wrong. Try again.
An ATProto Powered Reputation Network
Something went wrong. Try again.
3.7 kB · 123 lines
Go
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124package appview
import ( "context" "errors" "path/filepath" "testing" "time"
"provenance/internal/domain" "provenance/internal/storage" "provenance/internal/trust")
type reconnectStream struct { cursors chan int64 change domain.Change}
func (f *reconnectStream) Consume(ctx context.Context, cursor int64, wanted func(context.Context, string, int64) (bool, error), apply func(context.Context, domain.Change) error) error { f.cursors <- cursor if f.change.TimeUS == 0 { f.change.TimeUS = cursor + time.Second.Microseconds() selected, err := wanted(ctx, f.change.DID, f.change.TimeUS) if err != nil || !selected { return errors.New("participant filter rejected the event") } if err := apply(ctx, f.change); err != nil { return err } return errors.New("connection closed") } <-ctx.Done() return ctx.Err()}
func TestReconnectUsesCommittedCursor(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() store, err := storage.Open(filepath.Join(t.TempDir(), "index.db")) if err != nil { t.Fatal(err) } defer store.Close() repo := &repository{} service := New(store, repo, trust.Limits{States: 100, Edges: 100, Penalty: .1}) if err := service.Join(ctx, alice); err != nil { t.Fatal(err) } v := domain.Vouch{Author: alice, Subject: bob, Score: 2} stream := &reconnectStream{cursors: make(chan int64, 2), change: domain.Change{Kind: domain.RecordChange, DID: alice, RecordKey: domain.RecordKey(bob, ""), Vouch: &v}} done := make(chan struct{}) go func() { defer close(done); service.Run(ctx, stream) }() var first, second int64 select { case first = <-stream.cursors: case <-ctx.Done(): t.Fatal("stream did not start") } select { case second = <-stream.cursors: case <-ctx.Done(): t.Fatal("stream did not reconnect") } cancel() <-done if second != first+time.Second.Microseconds() { t.Fatalf("reconnect cursor = %d, initial = %d", second, first) } result, err := service.Reputation(context.Background(), alice, bob, "", domain.Default) if err != nil || result.Score == nil || *result.Score != 2 { t.Fatal("stream event did not reach the graph") }}
type blockingRepository struct { repository entered chan struct{} release chan struct{} vouch domain.Vouch}
func (r *blockingRepository) Snapshot(ctx context.Context, _ string) ([]domain.Vouch, error) { close(r.entered) select { case <-r.release: return []domain.Vouch{r.vouch}, nil case <-ctx.Done(): return nil, ctx.Err() }}
func TestBootstrapAndStreamUpdate(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() store, err := storage.Open(filepath.Join(t.TempDir(), "index.db")) if err != nil { t.Fatal(err) } defer store.Close() v := domain.Vouch{Author: alice, Subject: bob, Score: 2} repo := &blockingRepository{entered: make(chan struct{}), release: make(chan struct{}), vouch: v} service := New(store, repo, trust.Limits{States: 100, Edges: 100, Penalty: .1}) joined := make(chan error, 1) go func() { joined <- service.Join(ctx, alice) }() <-repo.entered v.Score = domain.Blocked applied := make(chan error, 1) go func() { applied <- service.Apply(ctx, domain.Change{Kind: domain.RecordChange, DID: alice, TimeUS: time.Now().Add(time.Second).UnixMicro(), RecordKey: domain.RecordKey(bob, ""), Vouch: &v}) }() close(repo.release) if err := <-joined; err != nil { t.Fatal(err) } if err := <-applied; err != nil { t.Fatal(err) } result, err := service.Reputation(ctx, alice, bob, "", domain.Default) if err != nil || result.Score == nil || *result.Score != domain.Blocked { t.Fatal("bootstrap overwrote a concurrent block") }}