eventlog_test.go
1package eventlog
2
3import (
4 "context"
5 "net/http/httptest"
6 "path/filepath"
7 "strings"
8 "sync"
9 "testing"
10 "time"
11
12 "git.kilimanjaro.io/ci/internal/lex"
13)
14
15func open(t *testing.T) *Log {
16 t.Helper()
17 l, err := Open(filepath.Join(t.TempDir(), "events.db"))
18 if err != nil {
19 t.Fatal(err)
20 }
21 t.Cleanup(func() { l.Close() })
22 return l
23}
24
25func TestAppendListMonotonic(t *testing.T) {
26 l := open(t)
27 var prev int64
28 for i := range 100 {
29 ev, err := l.Append("x.test", "", map[string]int{"i": i})
30 if err != nil {
31 t.Fatal(err)
32 }
33 if ev.Created <= prev {
34 t.Fatalf("created not increasing: %d after %d", ev.Created, prev)
35 }
36 prev = ev.Created
37 }
38 all, _ := l.List(0, 1000)
39 if len(all) != 100 {
40 t.Fatalf("%d events", len(all))
41 }
42 tail, _ := l.List(all[89].Created, 1000)
43 if len(tail) != 10 || string(tail[0].Event) != `{"i":90}` {
44 t.Errorf("after cursor: %d %s", len(tail), tail[0].Event)
45 }
46 // reopening keeps the clock ahead of what's stored
47 path := filepath.Join(t.TempDir(), "e.db")
48 a, _ := Open(path)
49 e1, _ := a.Append("x", "", 1)
50 a.Close()
51 b, _ := Open(path)
52 defer b.Close()
53 if b.last != e1.Created {
54 t.Errorf("reopened last %d want %d", b.last, e1.Created)
55 }
56}
57
58func TestStreamReplayTailAndResume(t *testing.T) {
59 l := open(t)
60 for i := range 3 {
61 l.Append("x.test", "", i)
62 }
63 srv := httptest.NewServer(RequireToken("s3cret", l))
64 defer srv.Close()
65 url := "ws" + strings.TrimPrefix(srv.URL, "http")
66
67 var mu sync.Mutex
68 var got []lex.Event
69 ctx, cancel := context.WithCancel(context.Background())
70 done := make(chan error)
71 go func() {
72 done <- Subscribe(ctx, url, 0, "s3cret", func(ev lex.Event) error {
73 mu.Lock()
74 got = append(got, ev)
75 mu.Unlock()
76 return nil
77 })
78 }()
79 wait := func(n int) {
80 t.Helper()
81 for range 200 {
82 mu.Lock()
83 k := len(got)
84 mu.Unlock()
85 if k >= n {
86 return
87 }
88 time.Sleep(10 * time.Millisecond)
89 }
90 t.Fatalf("got %d events, want %d", len(got), n)
91 }
92 wait(3)
93 l.Append("x.test", "", 3) // tail
94 wait(4)
95 cancel()
96 <-done
97
98 // resume from the last cursor: only newer events
99 l.Append("x.test", "", 4)
100 var resumed []lex.Event
101 ctx2, cancel2 := context.WithCancel(context.Background())
102 go func() {
103 _ = Subscribe(ctx2, url, got[3].Created, "s3cret", func(ev lex.Event) error {
104 resumed = append(resumed, ev)
105 cancel2()
106 return nil
107 })
108 }()
109 <-ctx2.Done()
110 time.Sleep(20 * time.Millisecond)
111 if len(resumed) != 1 || string(resumed[0].Event) != "4" {
112 t.Errorf("resumed %+v", resumed)
113 }
114}
115
116func TestStreamNeedsToken(t *testing.T) {
117 l := open(t)
118 srv := httptest.NewServer(RequireToken("s3cret", l))
119 defer srv.Close()
120 url := "ws" + strings.TrimPrefix(srv.URL, "http")
121 var c int64
122 if err := follow(context.Background(), url, &c, "wrong", func(lex.Event) error { return nil }); err == nil || !strings.Contains(err.Error(), "401") {
123 t.Errorf("wrong token: %v", err)
124 }
125}
