amadan

oficina / meet Public

Clone 
git clone https://amadan.net/oficina/meet

Plain git — no account needed to clone.

Download

Download this file

1package callscore
2
3import (
4 "encoding/json"
5 "fmt"
6 "net/http"
7 "net/http/httptest"
8 "sync"
9 "testing"
10 "time"
11
12 "amadan.net/oficina/calls/protocol"
13)
14
15func TestChatOutcomeWaitsForServer(t *testing.T) {
16 for _, status := range []int{http.StatusOK, http.StatusBadRequest, http.StatusRequestEntityTooLarge} {
17 t.Run(fmt.Sprint(status), func(t *testing.T) {
18 received := make(chan protocol.Signal, 1)
19 release := make(chan struct{})
20 events := make(chan Event, 1)
21 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
22 var signal protocol.Signal
23 if err := json.NewDecoder(req.Body).Decode(&signal); err != nil {
24 t.Error(err)
25 }
26 received <- signal
27 select {
28 case <-release:
29 w.WriteHeader(status)
30 case <-req.Context().Done():
31 }
32 }))
33 defer srv.Close()
34 r, err := New(srv.URL, "room", srv.Client(), func(e Event) { events <- e })
35 if err != nil {
36 t.Fatal(err)
37 }
38 r.started = true
39 r.state.Peer = "self"
40 defer r.Close()
41 const payload = `{"text":"hello"}`
42 if err := r.Send(protocol.TypeChat, "", 0, payload); err != nil {
43 t.Fatal(err)
44 }
45 select {
46 case signal := <-received:
47 if signal.Type != protocol.TypeChat || signal.To != "" || signal.Seq != 0 || signal.Gen != 0 {
48 t.Fatalf("invalid broadcast: %+v", signal)
49 }
50 case <-time.After(time.Second):
51 t.Fatal("chat was not posted")
52 }
53 select {
54 case e := <-events:
55 t.Fatalf("outcome before server response: %+v", e)
56 default:
57 }
58 close(release)
59 select {
60 case e := <-events:
61 want := "send-failed"
62 if status == http.StatusOK {
63 want = "sent"
64 }
65 if e.Type != want || e.Signal == nil || string(e.Signal.Payload) != payload {
66 t.Fatalf("outcome = %+v, want %s with original text", e, want)
67 }
68 case <-time.After(time.Second):
69 t.Fatal("missing chat outcome")
70 }
71 })
72 }
73}
74
75func TestReregisterReportsEveryQueuedChatAsFailed(t *testing.T) {
76 received := make(chan struct{}, 1)
77 events := make(chan Event, 8)
78 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
79 // Hold the first POST so the remaining chats must still be queued.
80 var signal protocol.Signal
81 if err := json.NewDecoder(req.Body).Decode(&signal); err != nil {
82 t.Error(err)
83 }
84 received <- struct{}{}
85 <-req.Context().Done()
86 }))
87 defer server.Close()
88 r, err := New(server.URL, "room", server.Client(), func(e Event) {
89 if e.Signal != nil {
90 events <- e
91 }
92 })
93 if err != nil {
94 t.Fatal(err)
95 }
96 defer r.Close()
97 r.started = true
98 r.state.Peer = "old"
99 for i := range 3 {
100 if err := r.Send(protocol.TypeChat, "", 0, fmt.Sprintf(`{"text":"message %d"}`, i)); err != nil {
101 t.Fatal(err)
102 }
103 if i == 0 {
104 select {
105 case <-received:
106 case <-time.After(time.Second):
107 t.Fatal("first chat was not posted")
108 }
109 }
110 }
111 r.fresh()
112 r.Close() // Wait for every cancellation callback, including the queue drain.
113 if len(events) != 3 {
114 t.Fatalf("got %d outcomes, want all three accepted chats", len(events))
115 }
116 for i := range 3 {
117 e := <-events
118 if e.Type != "send-failed" || string(e.Signal.Payload) != fmt.Sprintf(`{"text":"message %d"}`, i) {
119 t.Fatalf("outcome %d lost its text or order: %+v", i, e)
120 }
121 }
122}
123
124func TestResumeRetainsParticipationAndDedupe(t *testing.T) {
125 var mu sync.Mutex
126 opens, joins := 0, 0
127 messages := make(chan Event, 64)
128 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
129 switch {
130 case req.URL.Path == "/m/room/ice":
131 fmt.Fprint(w, `{"iceServers":[],"expiresAt":0,"renewAt":0}`)
132 case req.URL.Path == "/m/room/signal":
133 var s protocol.Signal
134 _ = json.NewDecoder(req.Body).Decode(&s)
135 mu.Lock()
136 if s.Type == "join" {
137 joins++
138 }
139 mu.Unlock()
140 w.WriteHeader(200)
141 case req.URL.Path == "/m/room/stream":
142 mu.Lock()
143 opens++
144 n := opens
145 mu.Unlock()
146 if req.Header.Get("Accept") != "text/event-stream" {
147 t.Error("missing SSE Accept")
148 }
149 if n == 2 && (req.URL.Query().Get("peer") != "self" || req.URL.Query().Get("last") != `{"other":1}`) {
150 t.Error("resume lost peer or cursor")
151 }
152 w.Header().Set("Content-Type", "text/event-stream")
153 fmt.Fprint(w, "data: {\"type\":\"hello\",\"peer\":\"self\",\"boot\":\"boot\"}\n\n")
154 fmt.Fprint(w, "data: {\"type\":\"roster\",\"roster\":{\"peers\":[{\"peer\":\"self\",\"kind\":\"member\"},{\"peer\":\"other\",\"gen\":1}]}}\n\n")
155 fmt.Fprint(w, "data: {\"type\":\"offer\",\"from\":\"other\",\"seq\":1,\"gen\":1,\"payload\":{}}\n\n")
156 w.(http.Flusher).Flush()
157 if n == 1 {
158 time.Sleep(50 * time.Millisecond)
159 return
160 }
161 fmt.Fprint(w, "data: {\"type\":\"candidate\",\"from\":\"other\",\"seq\":2,\"gen\":1,\"payload\":{}}\n\n")
162 w.(http.Flusher).Flush()
163 <-req.Context().Done()
164 default:
165 w.WriteHeader(404)
166 }
167 }))
168 defer server.Close()
169 r, err := New(server.URL, "room", server.Client(), func(e Event) { messages <- e })
170 if err != nil {
171 t.Fatal(err)
172 }
173 if err = r.Start(); err != nil {
174 t.Fatal(err)
175 }
176 defer r.Close()
177 timer := time.NewTimer(5 * time.Second)
178 defer timer.Stop()
179 offers := 0
180 for {
181 select {
182 case e := <-messages:
183 if e.Message != nil {
184 if e.Message.Type == "offer" {
185 offers++
186 }
187 if e.Message.Type == "candidate" {
188 mu.Lock()
189 j := joins
190 mu.Unlock()
191 if offers != 1 || j != 1 {
192 t.Fatalf("offers=%d joins=%d", offers, j)
193 }
194 return
195 }
196 }
197 case <-timer.C:
198 t.Fatal("resume never delivered")
199 }
200 }
201}
202func TestGenerationFenceAndMonotonicSequence(t *testing.T) {
203 received := make(chan protocol.Signal, 10)
204 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
205 var s protocol.Signal
206 _ = json.NewDecoder(req.Body).Decode(&s)
207 received <- s
208 }))
209 defer srv.Close()
210 r, _ := New(srv.URL, "room", srv.Client(), func(Event) {})
211 // Seed a live participant without a stream to isolate outgoing generation
212 // fencing, including callbacks that arrive after the native PC was replaced.
213 r.started = true
214 r.state.Peer = "self"
215 r.state.Peers = []protocol.RosterPeer{{Peer: "other", Gen: 1}}
216 defer r.Close()
217 if err := r.Send("offer", "other", 1, `{}`); err != nil {
218 t.Fatal(err)
219 }
220 select {
221 case s := <-received:
222 if s.Seq != 1 {
223 t.Fatal(s.Seq)
224 }
225 case <-time.After(time.Second):
226 t.Fatal("missing offer")
227 }
228 if err := r.receive(protocol.Message{Type: "pair", Peer: "other", Gen: 2}); err != nil {
229 t.Fatal(err)
230 }
231 if err := r.Send("candidate", "other", 1, `{}`); err == nil {
232 t.Fatal("accepted callback from replaced PC")
233 }
234 if err := r.Send("offer", "other", 2, `{}`); err != nil {
235 t.Fatal(err)
236 }
237 select {
238 case s := <-received:
239 if s.Seq != 2 || s.Gen != 2 {
240 t.Fatal(s)
241 }
242 case <-time.After(time.Second):
243 t.Fatal("missing new offer")
244 }
245}
246func TestTerminalLossClearsMediaAndDoesNotRejoin(t *testing.T) {
247 r, _ := New("https://calls.example.com", "room", http.DefaultClient, func(Event) {})
248 r.started = true
249 r.state.Peer = "self"
250 r.state.Peers = []protocol.RosterPeer{{Peer: "other", Gen: 1}}
251 _ = r.receive(protocol.Message{Type: "lost", Reason: protocol.ReasonRemoved})
252 s := r.Snapshot()
253 if s.Phase != "ended" || s.Reason != "removed" || len(s.Peers) != 0 {
254 t.Fatal(s)
255 }
256 if err := r.Send("join", "", 0, ""); err == nil {
257 t.Fatal("terminal participant rejoined")
258 }
259 r.Close()
260}
261
262func TestTargetQueueRetriesSameSequenceBeforeNextMessage(t *testing.T) {
263 received := make(chan protocol.Signal, 8)
264 attempts := 0
265 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
266 var s protocol.Signal
267 _ = json.NewDecoder(req.Body).Decode(&s)
268 attempts++
269 received <- s
270 if attempts == 1 {
271 w.WriteHeader(503)
272 return
273 }
274 if attempts == 2 {
275 w.Header().Set("Retry-After", "1")
276 w.WriteHeader(429)
277 return
278 }
279 }))
280 defer srv.Close()
281 r, _ := New(srv.URL, "room", srv.Client(), func(Event) {})
282 r.started = true
283 r.state.Peer = "self"
284 r.state.Peers = []protocol.RosterPeer{{Peer: "other", Gen: 1}}
285 defer r.Close()
286 if err := r.Send("offer", "other", 1, `{}`); err != nil {
287 t.Fatal(err)
288 }
289 if err := r.Send("candidate", "other", 1, `{}`); err != nil {
290 t.Fatal(err)
291 }
292 deadline := time.NewTimer(5 * time.Second)
293 defer deadline.Stop()
294 for i, want := range []uint64{1, 1, 1, 2} {
295 select {
296 case s := <-received:
297 if s.Seq != want || (i < 3 && s.Type != "offer") || (i == 3 && s.Type != "candidate") {
298 t.Fatalf("attempt %d: %+v", i, s)
299 }
300 case <-deadline.C:
301 t.Fatal("ordered retry stalled")
302 }
303 }
304}
305
306// A fresh open refused with 409 takes the session's own participation back
307// once, and a refused takeover — or a second 409 — is "replaced" as before.
308func TestFreshOpenConflictTakesOverOnce(t *testing.T) {
309 for _, tc := range []struct {
310 name string
311 takeover int
312 location string
313 conflicts int
314 want string
315 }{
316 {"taken back", 303, "/m/room", 1, "admitted"},
317 {"takeover refused", 404, "", 1, "replaced"},
318 {"sign-in redirect is a refusal", 303, "/signin", 1, "replaced"},
319 {"only once", 303, "/m/room", 2, "replaced"},
320 } {
321 t.Run(tc.name, func(t *testing.T) {
322 var mu sync.Mutex
323 opens, takeovers := 0, 0
324 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
325 switch req.URL.Path {
326 case "/m/room/ice":
327 fmt.Fprint(w, `{"iceServers":[],"expiresAt":0,"renewAt":0}`)
328 case "/m/room/signal":
329 w.WriteHeader(200)
330 case "/m/room/takeover":
331 mu.Lock()
332 takeovers++
333 mu.Unlock()
334 if req.Method != http.MethodPost || req.Header.Get("Origin") == "" {
335 t.Error("takeover must be a POST carrying Origin")
336 }
337 if tc.location != "" {
338 w.Header().Set("Location", tc.location)
339 }
340 w.WriteHeader(tc.takeover)
341 case "/m/room/stream":
342 mu.Lock()
343 opens++
344 n := opens
345 mu.Unlock()
346 if n <= tc.conflicts {
347 w.Header().Set("Content-Type", "application/json")
348 w.WriteHeader(409)
349 fmt.Fprint(w, `{"peer":"ghost"}`)
350 return
351 }
352 w.Header().Set("Content-Type", "text/event-stream")
353 fmt.Fprint(w, "data: {\"type\":\"hello\",\"peer\":\"self\",\"boot\":\"boot\"}\n\n")
354 fmt.Fprint(w, "data: {\"type\":\"roster\",\"roster\":{\"peers\":[{\"peer\":\"self\",\"kind\":\"member\"}]}}\n\n")
355 w.(http.Flusher).Flush()
356 <-req.Context().Done()
357 default:
358 w.WriteHeader(404)
359 }
360 }))
361 defer server.Close()
362 states := make(chan Snapshot, 64)
363 r, err := New(server.URL, "room", server.Client(), func(e Event) {
364 if e.State != nil {
365 states <- *e.State
366 }
367 })
368 if err != nil {
369 t.Fatal(err)
370 }
371 if err = r.Start(); err != nil {
372 t.Fatal(err)
373 }
374 defer r.Close()
375 timer := time.NewTimer(5 * time.Second)
376 defer timer.Stop()
377 for {
378 select {
379 case s := <-states:
380 got := ""
381 if s.Phase == "admitted" {
382 got = "admitted"
383 } else if s.Phase == "ended" {
384 got = s.Reason
385 }
386 if got == "" {
387 continue
388 }
389 if got != tc.want {
390 t.Fatalf("got %q, want %q", got, tc.want)
391 }
392 mu.Lock()
393 defer mu.Unlock()
394 if takeovers != 1 {
395 t.Fatalf("takeovers = %d, want exactly 1", takeovers)
396 }
397 return
398 case <-timer.C:
399 t.Fatal("no outcome")
400 }
401 }
402 })
403 }
404}
405