1package client
2
3import (
4 "bytes"
5 "context"
6 "io"
7 "strings"
8 "sync"
9 "testing"
10 "time"
11
12 "github.com/stretchr/testify/require"
13
14 "github.com/bnema/vev/internal/domain"
15 "github.com/bnema/vev/internal/ports"
16 "github.com/bnema/vev/internal/protocol/catalogue"
17 "github.com/bnema/vev/internal/usecase/ui"
18)
19
20// Client-owned picker controller tests (Plan 001 P5.2b): rendering, local
21// search/cursor behaviour across live updates, and the input-ownership proof
22// that picker input never reaches a session.
23
24// pickerChunkReader yields pre-pushed chunks to the supervisor's single input
25// lifetime. push blocks until the lifetime reads the chunk, so a test can
26// order a read deterministically without a sleep.
27type pickerChunkReader struct {
28 chunks chan []byte
29}
30
31func newPickerChunkReader() *pickerChunkReader {
32 return &pickerChunkReader{chunks: make(chan []byte)}
33}
34
35func (r *pickerChunkReader) push(data []byte) {
36 r.chunks <- append([]byte(nil), data...)
37}
38
39func (r *pickerChunkReader) close() { close(r.chunks) }
40
41func (r *pickerChunkReader) Read(buffer []byte) (int, error) {
42 chunk, ok := <-r.chunks
43 if !ok {
44 return 0, io.EOF
45 }
46 return copy(buffer, chunk), nil
47}
48
49// pickerInputSpy forwards each read to the real consumer and signals that it
50// was delivered, so a test can wait exactly for consumption.
51type pickerInputSpy struct {
52 inner pickerInputConsumer
53 done chan struct{}
54}
55
56func newPickerInputSpy(inner pickerInputConsumer) *pickerInputSpy {
57 return &pickerInputSpy{inner: inner, done: make(chan struct{}, 1)}
58}
59
60func (s *pickerInputSpy) ConsumeTerminalRead(data []byte) bool {
61 consumed := s.inner.ConsumeTerminalRead(data)
62 select {
63 case s.done <- struct{}{}:
64 default:
65 }
66 return consumed
67}
68
69func (s *pickerInputSpy) await(t *testing.T) {
70 t.Helper()
71 select {
72 case <-s.done:
73 case <-time.After(5 * time.Second):
74 t.Fatal("terminal input was not consumed")
75 }
76}
77
78// pickerRecordingTerminal records every byte written to the terminal output,
79// which is where a leaked picker key would surface if it reached a session.
80type pickerRecordingTerminal struct {
81 in io.Reader
82 out bytes.Buffer
83 mu sync.Mutex
84 enters int
85}
86
87func (t *pickerRecordingTerminal) EnableKittyKeyboard(int) error { return nil }
88
89func (t *pickerRecordingTerminal) EnterRaw() (func() error, error) {
90 t.mu.Lock()
91 t.enters++
92 t.mu.Unlock()
93 return func() error { return nil }, nil
94}
95
96func (t *pickerRecordingTerminal) Geometry() (domain.Geometry, error) {
97 return domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}}, nil
98}
99
100func (t *pickerRecordingTerminal) ResizeEvents() <-chan domain.Geometry { return nil }
101
102func (t *pickerRecordingTerminal) In() io.Reader { return t.in }
103
104func (t *pickerRecordingTerminal) Out() io.Writer { return &t.out }
105
106func (t *pickerRecordingTerminal) Flush() error { return nil }
107
108func pickerTestSnapshot(epoch ports.BrokerEpoch, revision ports.BrokerRevision, now time.Time, sessions ...string) ports.BrokerSnapshot {
109 catalogSessions := make([]catalogue.RemoteCatalogSession, 0, len(sessions))
110 for i, name := range sessions {
111 catalogSessions = append(catalogSessions, pickerTestSession(name, byte(i+1), catalogue_Up))
112 }
113 return ports.BrokerSnapshot{Epoch: epoch, Revision: revision, Daemons: []ports.BrokerDaemonObservation{
114 pickerTestLocalObservation(now, catalogSessions...),
115 }}
116}
117
118func pickerTestController(t *testing.T) (*pickerController, *supervisorTestClock) {
119 t.Helper()
120 clock := newSupervisorTestClock()
121 return newPickerController(clock, pickerTestFreshness, true), clock
122}
123
124func pickerRowKeyByLabel(t *testing.T, controller *pickerController, label string) string {
125 t.Helper()
126 line, ok := pickerLineByLabel(controller.Catalogue().Lines(), label)
127 require.True(t, ok, "row %q must be projected", label)
128 return line.Key
129}
130
131func mustCursorKey(t *testing.T, controller *pickerController) string {
132 t.Helper()
133 key, ok := controller.CursorKey()
134 require.True(t, ok, "the picker must have a selected row")
135 return key
136}
137
138func (p *pickerController) modelState(t *testing.T) pickerState {
139 t.Helper()
140 state := p.state()
141 require.True(t, state.hasModel, "the picker must have a model")
142 return state
143}
144
145func (p *pickerController) modelQuery(t *testing.T) string {
146 t.Helper()
147 return p.modelState(t).query
148}
149
150func (p *pickerController) modelSearchActive(t *testing.T) bool {
151 t.Helper()
152 return p.modelState(t).searchActive
153}
154
155func TestPickerControllerRendersLatestSnapshotImmediately(t *testing.T) {
156 controller, clock := pickerTestController(t)
157 controller.ApplySnapshot(pickerTestSnapshot(3, 1, clock.Now(), "alpha", "beta"))
158 firstRender := controller.Render(domain.Size{Cols: 80, Rows: 24})
159 require.NotEmpty(t, firstRender)
160 alphaKey := pickerRowKeyByLabel(t, controller, "alpha")
161 require.Equal(t, alphaKey, mustCursorKey(t, controller))
162
163 // A newer publication is folded synchronously: the very next render already
164 // reflects it, with no debounce or intervening frame.
165 controller.ApplySnapshot(pickerTestSnapshot(3, 2, clock.Now(), "beta"))
166 require.Equal(t, pickerRowKeyByLabel(t, controller, "beta"), mustCursorKey(t, controller))
167 require.NotEmpty(t, controller.Render(domain.Size{Cols: 80, Rows: 24}))
168}
169
170func TestPickerControllerSearchSurvivesHostUpdate(t *testing.T) {
171 controller, clock := pickerTestController(t)
172 controller.ApplySnapshot(pickerTestSnapshot(3, 1, clock.Now(), "alpha", "beta"))
173
174 spy := newPickerInputSpy(controller)
175 reader := newPickerChunkReader()
176 t.Cleanup(reader.close)
177 _ = startTerminalInputLifetime(reader, spy)
178
179 reader.push([]byte("/alp"))
180 spy.await(t)
181 require.Equal(t, "alp", controller.modelQuery(t))
182 require.True(t, controller.modelSearchActive(t))
183
184 // A host update while the query is typed keeps the editor and the matched
185 // cursor; it never resets the local search.
186 controller.ApplySnapshot(pickerTestSnapshot(3, 2, clock.Now(), "alpha", "beta", "gamma"))
187 require.Equal(t, "alp", controller.modelQuery(t))
188 require.True(t, controller.modelSearchActive(t))
189 require.Equal(t, pickerRowKeyByLabel(t, controller, "alpha"), mustCursorKey(t, controller))
190}
191
192// TestPickerControllerInputNeverReachesPTY feeds every input shape a picker can
193// receive and proves the only effects are local: each batch is consumed, the
194// terminal receives no bytes, and the decisions are presentation facts, never a
195// session write.
196func TestPickerControllerInputNeverReachesPTY(t *testing.T) {
197 controller, clock := pickerTestController(t)
198 controller.ApplySnapshot(pickerTestSnapshot(3, 1, clock.Now(), "alpha", "beta"))
199
200 alphaKey := pickerRowKeyByLabel(t, controller, "alpha")
201 betaKey := pickerRowKeyByLabel(t, controller, "beta")
202
203 spy := newPickerInputSpy(controller)
204 reader := newPickerChunkReader()
205 t.Cleanup(reader.close)
206 terminal := &pickerRecordingTerminal{in: reader}
207 _ = startTerminalInputLifetime(terminal.In(), spy)
208
209 chunks := [][]byte{
210 []byte("\x1b[B"), // Down -> beta
211 []byte("k"), // up -> alpha
212 []byte("j"), // down -> beta
213 []byte("\x1b[A"), // Up -> alpha
214 []byte("\x1b[<0;10;5M"), // mouse report: consumed and dropped
215 []byte("\x1b[200~rm -rf /\x1b[201~"), // bracketed paste: content dropped
216 []byte("\x00\x01\x7f"), // control bytes: consumed and dropped
217 []byte("\x1b"), // lone ESC withheld for disambiguation
218 }
219 for _, chunk := range chunks {
220 reader.push(chunk)
221 spy.await(t)
222 }
223 require.Equal(t, alphaKey, mustCursorKey(t, controller), "arrow keys moved the local cursor")
224 require.False(t, controller.modelSearchActive(t), "mouse, paste, and control bytes never opened search")
225 require.Equal(t, betaKey, pickerRowKeyByLabel(t, controller, "beta"))
226
227 // A withheld lone ESC resolves through the bounded window as one Escape.
228 controller.FlushPending()
229 op, _ := controller.TakeOp()
230 require.True(t, op.close, "Escape retires the picker locally")
231
232 require.Empty(t, terminal.out.Bytes(), "the picker wrote nothing to the terminal")
233}
234
235func TestPickerControllerInputRouting(t *testing.T) {
236 tests := []struct {
237 name string
238 chunk []byte
239 flush bool
240 wantCursor string
241 wantNoSelection bool
242 wantSearch bool
243 wantQuery string
244 wantOp pickerOp
245 }{
246 {name: "down", chunk: []byte("\x1b[B"), wantCursor: "beta"},
247 {name: "up skips informational host row", chunk: []byte("\x1b[A"), wantCursor: "alpha"},
248 {name: "search entry", chunk: []byte("/"), wantCursor: "alpha", wantSearch: true},
249 {name: "search text", chunk: []byte("/al"), wantCursor: "alpha", wantSearch: true, wantQuery: "al"},
250 {name: "enter commits", chunk: []byte("/al\r"), wantCursor: "alpha", wantSearch: true, wantQuery: "al", wantOp: pickerOp{commit: true}},
251 {name: "kill", chunk: []byte("x"), wantCursor: "alpha", wantOp: pickerOp{kill: true}},
252 {name: "escape exits search first", chunk: []byte("/al\x1b"), flush: true, wantCursor: "alpha"},
253 {name: "escape closes", chunk: []byte("\x1b"), flush: true, wantCursor: "alpha", wantOp: pickerOp{close: true}},
254 {name: "f1 is consumed", chunk: []byte("\x1b[11~"), wantCursor: "alpha"},
255 }
256
257 for _, tt := range tests {
258 t.Run(tt.name, func(t *testing.T) {
259 controller, clock := pickerTestController(t)
260 controller.ApplySnapshot(pickerTestSnapshot(3, 1, clock.Now(), "alpha", "beta"))
261 alphaKey := pickerRowKeyByLabel(t, controller, "alpha")
262 spy := newPickerInputSpy(controller)
263 reader := newPickerChunkReader()
264 t.Cleanup(reader.close)
265 _ = startTerminalInputLifetime(reader, spy)
266
267 reader.push(tt.chunk)
268 spy.await(t)
269 if tt.flush {
270 controller.FlushPending()
271 }
272 if tt.wantOp != (pickerOp{}) {
273 op, _ := controller.TakeOp()
274 require.Equal(t, tt.wantOp, op)
275 }
276 require.Equal(t, tt.wantSearch, controller.modelSearchActive(t))
277 if tt.wantQuery != "" {
278 require.Equal(t, tt.wantQuery, controller.modelQuery(t))
279 }
280 if tt.wantNoSelection {
281 _, ok := controller.CursorKey()
282 require.False(t, ok, "the host row is a cursor destination but not a selection")
283 return
284 }
285 want := alphaKey
286 if tt.wantCursor == "beta" {
287 want = pickerRowKeyByLabel(t, controller, "beta")
288 }
289 require.Equal(t, want, mustCursorKey(t, controller))
290 })
291 }
292}
293
294// TestPickerControllerNotOwningInputLeavesReadUnconsumed pins the ownership
295// boundary: once the picker releases input, a read is not consumed, so a later
296// attach slice keeps its ordinary pipeline.
297func TestPickerControllerNotOwningInputLeavesReadUnconsumed(t *testing.T) {
298 controller, clock := pickerTestController(t)
299 controller.ApplySnapshot(pickerTestSnapshot(3, 1, clock.Now(), "alpha"))
300 controller.SetOwnsInput(false)
301 require.False(t, controller.ConsumeTerminalRead([]byte("j")))
302 controller.SetOwnsInput(true)
303 require.True(t, controller.ConsumeTerminalRead([]byte("j")))
304}
305
306func TestPickerControllerKeepsProgressInRowsAndSurfacesFailuresAsNotices(t *testing.T) {
307 controller, clock := pickerTestController(t)
308 checking := pickerTestRemoteObservation("user@arch", 1, 1, clock.Now(), pickerTestSession("alpha", 1, catalogue_Up))
309 checking.Checking = true
310 controller.ApplySnapshot(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{checking}})
311 require.Empty(t, controller.RenderNotice(domain.Size{Cols: 80, Rows: 24}), "ordinary refreshing progress stays in the rows")
312
313 unreachable := pickerTestRemoteObservation("user@arch", 1, 1, clock.Now())
314 unreachable.Availability = domain.RemoteAvailabilityUnreachable
315 unreachable.FailureEpisode = 1
316 controller.ApplySnapshot(ports.BrokerSnapshot{Epoch: 3, Revision: 2, Daemons: []ports.BrokerDaemonObservation{unreachable}})
317 require.NotEmpty(t, controller.RenderNotice(domain.Size{Cols: 80, Rows: 24}), "an unavailable host is surfaced as a failure")
318}
319
320// TestPickerControllerResolveInspectableRowStillResolvesExplicitly: a stale row
321// the picker displays for inspection still resolves to one exact request,
322// because freshness is presentation information and only the destination may
323// refuse the exact identity. The informational row stays dimmed and badged.
324func TestPickerControllerResolveInspectableRowStillResolvesExplicitly(t *testing.T) {
325 controller, clock := pickerTestController(t)
326 stale := pickerTestLocalObservation(clock.Now().Add(-2*pickerTestFreshness), pickerTestSession("alpha", 1, catalogue_Up))
327 controller.ApplySnapshot(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{stale}})
328
329 // The stale session row stays a cursor destination and admits an explicit
330 // attempt; only the host status row itself is never committable.
331 key, ok := controller.CursorKey()
332 require.True(t, ok, "a stale session row stays a cursor destination")
333 require.NotEmpty(t, key)
334 require.False(t, controller.modelSearchActive(t))
335
336 request, err := controller.Resolve(pickerTestBase())
337 require.NoError(t, err)
338 require.Equal(t, ports.BrokerAdmissionExact, request.Admission)
339 require.Equal(t, "alpha", request.Target.SessionName)
340}
341
342// TestPickerControllerResolveUsesLatestSnapshot proves resolution revalidates
343// against the snapshot in force at resolve time, not the one displayed.
344func TestPickerControllerResolveUsesLatestSnapshot(t *testing.T) {
345 controller, clock := pickerTestController(t)
346 controller.ApplySnapshot(pickerTestSnapshot(3, 1, clock.Now(), "alpha"))
347 key := mustCursorKey(t, controller)
348
349 // The host disappears before the selection resolves.
350 controller.ApplySnapshot(ports.BrokerSnapshot{Epoch: 3, Revision: 2})
351
352 _, err := controller.Resolve(pickerTestBase())
353 require.Error(t, err)
354 require.True(t, pickerCatalogueErrorIs(err, pickerCatalogueNoSelection), "the vanished row leaves no selection")
355 _, err = controller.Catalogue().Resolve(key, pickerTestBase())
356 require.Error(t, err)
357 require.True(t, pickerCatalogueErrorIs(err, pickerCatalogueGone))
358 require.NotEmpty(t, controller.RenderNotice(domain.Size{Cols: 80, Rows: 24}), "the refusal is surfaced as a bounded notice")
359}
360
361func TestPickerControllerResolveReturnsExactRequest(t *testing.T) {
362 controller, clock := pickerTestController(t)
363 controller.ApplySnapshot(pickerTestSnapshot(3, 1, clock.Now(), "alpha"))
364
365 request, err := controller.Resolve(pickerTestBase())
366 require.NoError(t, err)
367 require.NoError(t, request.Validate())
368 require.Equal(t, ports.BrokerAdmissionExact, request.Admission)
369 require.Equal(t, ports.BrokerStreamAttachment, request.Purpose)
370 require.Equal(t, "alpha", request.Target.SessionName)
371}
372
373// TestPickerControllerConcurrentUpdatesAndInput exercises the mutex boundaries
374// under -race: publications, input consumption, rendering, and resolution run
375// concurrently and must not race or panic.
376func TestPickerControllerConcurrentUpdatesAndInput(t *testing.T) {
377 const iterations = 200
378 controller, clock := pickerTestController(t)
379 controller.ApplySnapshot(pickerTestSnapshot(3, 1, clock.Now(), "alpha"))
380
381 var wait sync.WaitGroup
382 wait.Add(3)
383
384 go func() {
385 defer wait.Done()
386 for revision := ports.BrokerRevision(2); revision < 2+iterations; revision++ {
387 controller.ApplySnapshot(pickerTestSnapshot(3, revision, clock.Now(), "alpha", "beta"))
388 }
389 }()
390
391 go func() {
392 defer wait.Done()
393 for range iterations {
394 controller.ConsumeTerminalRead([]byte("\x1b[B\x1b[Aj"))
395 controller.FlushPending()
396 controller.TakeOp()
397 }
398 }()
399
400 go func() {
401 defer wait.Done()
402 for range iterations {
403 _ = controller.Render(domain.Size{Cols: 80, Rows: 24})
404 _ = controller.RenderNotice(domain.Size{Cols: 80, Rows: 24})
405 _, _ = controller.Resolve(pickerTestBase())
406 _, _ = controller.CursorKey()
407 }
408 }()
409
410 wait.Wait()
411}
412
413// TestSupervisorPickerFoldsPublicationsAndInput proves the supervisor wiring:
414// a broker publication reaches the picker through the established connection's
415// subscription, and terminal input is consumed by the picker from the same
416// single input lifetime the supervisor uses for EOF, never written to the
417// terminal.
418func TestSupervisorPickerFoldsPublicationsAndInput(t *testing.T) {
419 ctx, cancel := context.WithCancel(context.Background())
420 defer cancel()
421
422 clock := newSupervisorTestClock()
423 reader := newPickerChunkReader()
424 t.Cleanup(reader.close)
425 terminal := &pickerRecordingTerminal{in: reader}
426 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) {
427 service := newSupervisorTestService(ports.BrokerConnectionID{1})
428 service.hub.publish(pickerTestSnapshot(3, 1, clock.Now(), "alpha", "beta"))
429 return service, nil
430 })
431 controller := newPickerController(clock, pickerTestFreshness, true)
432 sup := mustSupervisor(t, SupervisorConfig{Connector: connector, Terminal: terminal, Clock: clock, Picker: controller})
433
434 runErr := make(chan error, 1)
435 go func() { runErr <- sup.Run(ctx) }()
436
437 require.Eventually(t, func() bool {
438 return controller.Catalogue().Epoch() == 3
439 }, time.Second, time.Millisecond, "the supervisor folds the committed publication into the picker")
440 require.Equal(t, []string{"alpha", "beta"}, pickerSessionLabels(controller.Catalogue().Lines()))
441 require.True(t, controller.OwnsInput())
442
443 // The same input lifetime the supervisor uses for EOF delivers the read to
444 // the picker: the cursor moves locally and nothing is written anywhere.
445 betaKey := pickerRowKeyByLabel(t, controller, "beta")
446 reader.push([]byte("\x1b[B"))
447 require.Eventually(t, func() bool {
448 key, ok := controller.CursorKey()
449 return ok && key == betaKey
450 }, time.Second, time.Millisecond, "the picker consumed the terminal read")
451
452 cancel()
453 require.ErrorIs(t, <-runErr, context.Canceled)
454 require.Empty(t, terminal.out.Bytes(), "the picker never wrote a session or terminal byte")
455}
456
457// TestSupervisorPickerFoldsLaterPublication covers the second-and-later
458// publication: a snapshot delivered after the connection is established must
459// reach the picker through the subscription awaitLoss parks on. This is the
460// core of P5.2b and the fake hub's non-blocking, never-replaced channel is what
461// keeps it from busy-spinning.
462func TestSupervisorPickerFoldsLaterPublication(t *testing.T) {
463 ctx, cancel := context.WithCancel(context.Background())
464 defer cancel()
465
466 clock := newSupervisorTestClock()
467 reader := newPickerChunkReader()
468 t.Cleanup(reader.close)
469 terminal := &pickerRecordingTerminal{in: reader}
470
471 service := newSupervisorTestService(ports.BrokerConnectionID{1})
472 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) {
473 service.hub.publish(pickerTestSnapshot(3, 1, clock.Now(), "alpha"))
474 return service, nil
475 })
476 controller := newPickerController(clock, pickerTestFreshness, true)
477 sup := mustSupervisor(t, SupervisorConfig{Connector: connector, Terminal: terminal, Clock: clock, Picker: controller})
478
479 runErr := make(chan error, 1)
480 go func() { runErr <- sup.Run(ctx) }()
481
482 require.Eventually(t, func() bool {
483 return controller.Catalogue().Revision() == 1
484 }, time.Second, time.Millisecond, "the first committed publication reaches the picker")
485 require.Equal(t, []string{"alpha"}, pickerSessionLabels(controller.Catalogue().Lines()))
486
487 service.publishSnapshot(pickerTestSnapshot(3, 2, clock.Now(), "alpha", "beta"))
488 require.Eventually(t, func() bool {
489 return controller.Catalogue().Revision() == 2
490 }, time.Second, time.Millisecond, "a publication delivered after the connection is established reaches the picker")
491 require.Equal(t, []string{"alpha", "beta"}, pickerSessionLabels(controller.Catalogue().Lines()))
492
493 cancel()
494 require.ErrorIs(t, <-runErr, context.Canceled)
495}
496
497// TestPickerControllerResolveKeyUsesCommittedRow proves commit and resolve are
498// one action: the key captured with the commit decision resolves the committed
499// row even when a concurrent publication moved the cursor afterwards.
500func TestPickerControllerResolveKeyUsesCommittedRow(t *testing.T) {
501 controller, clock := pickerTestController(t)
502 controller.ApplySnapshot(pickerTestSnapshot(3, 1, clock.Now(), "alpha", "beta"))
503 alphaKey := pickerRowKeyByLabel(t, controller, "alpha")
504
505 // The user commits the row under the cursor.
506 require.True(t, controller.ConsumeTerminalRead([]byte("\r")))
507 op, committedKey := controller.TakeOp()
508 require.True(t, op.commit)
509 require.Equal(t, alphaKey, committedKey)
510
511 // A concurrent publication moves the cursor to another row, and the driver
512 // commits a row different from the one the user selected.
513 controller.ApplySnapshot(pickerTestSnapshot(3, 2, clock.Now(), "alpha", "beta", "gamma"))
514 require.True(t, controller.ConsumeTerminalRead([]byte("j")))
515
516 // Plain Resolve describes the row displayed now...
517 displayed, err := controller.Resolve(pickerTestBase())
518 require.NoError(t, err)
519 require.Equal(t, "beta", displayed.Target.SessionName)
520
521 // ...while the captured key resolves exactly the committed row.
522 committed, _, err := controller.ResolveCommit(committedKey, pickerTestBase())
523 require.NoError(t, err)
524 require.Equal(t, "alpha", committed.Target.SessionName)
525}
526
527// TestPickerControllerCommittedKeySurvivesConcurrentApply runs the commit path
528// under -race against a driver that keeps republishing. The captured key must
529// always resolve the row the user committed, never the row a publication left
530// under the cursor.
531func TestPickerControllerCommittedKeySurvivesConcurrentApply(t *testing.T) {
532 controller, clock := pickerTestController(t)
533 controller.ApplySnapshot(pickerTestSnapshot(3, 1, clock.Now(), "alpha", "beta"))
534
535 stop := make(chan struct{})
536 var applying sync.WaitGroup
537 applying.Add(1)
538 go func() {
539 defer applying.Done()
540 revision := ports.BrokerRevision(2)
541 for {
542 select {
543 case <-stop:
544 return
545 default:
546 }
547 controller.ApplySnapshot(pickerTestSnapshot(3, revision, clock.Now(), "alpha", "beta"))
548 revision++
549 }
550 }()
551
552 for range 200 {
553 require.True(t, controller.ConsumeTerminalRead([]byte("\r")))
554 op, key := controller.TakeOp()
555 if !op.commit || key == "" {
556 continue
557 }
558 request, _, err := controller.ResolveCommit(key, pickerTestBase())
559 if err != nil {
560 continue
561 }
562 require.Equal(t, "alpha", request.Target.SessionName, "the committed key must never resolve another row")
563 }
564 close(stop)
565 applying.Wait()
566}
567
568// TestPickerControllerApplySnapshotRejectsInvalidSnapshot proves the controller
569// keeps the previous projection and surfaces a bounded notice when a publication
570// fails the broker snapshot contract.
571func TestPickerControllerApplySnapshotRejectsInvalidSnapshot(t *testing.T) {
572 controller, clock := pickerTestController(t)
573 controller.ApplySnapshot(pickerTestSnapshot(3, 1, clock.Now(), "alpha"))
574 before := controller.Catalogue().Lines()
575
576 invalid := pickerTestSnapshot(3, 2, clock.Now(), "beta")
577 invalid.Daemons[0].Availability = 0
578 controller.ApplySnapshot(invalid)
579
580 require.Equal(t, before, controller.Catalogue().Lines(), "the previous projection is kept")
581 require.Equal(t, ports.BrokerRevision(1), controller.Catalogue().Revision())
582
583 active := controller.state().notices
584 require.Equal(t, []string{"broker catalogue update rejected"}, active)
585}
586
587// TestPickerControllerUnobservedDaemonIsRefreshingNotVersionMismatch pins the
588// availability-first classification at the controller boundary: an unobserved
589// daemon (Availability Unknown, ProtocolVersion 0) keeps refreshing on its
590// row and surfaces no failure toast.
591func TestPickerControllerUnobservedDaemonIsRefreshingNotVersionMismatch(t *testing.T) {
592 controller, clock := pickerTestController(t)
593 unobserved := pickerTestUnobservedObservation(clock.Now())
594 controller.ApplySnapshot(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{unobserved}})
595
596 require.Empty(t, controller.state().notices)
597 lines := controller.Catalogue().Lines()
598 host, ok := pickerHostLine(lines)
599 require.True(t, ok)
600 require.Equal(t, domain.RemoteReasonRefreshing, host.StatusDetail)
601 require.NotEqual(t, domain.RemoteReasonVersionMismatch, host.StatusDetail)
602}
603
604// TestPickerControllerRenderNoticeStacksQueuesAndBlanks pins the picker notice
605// stack: the newest notice sits at the top-right corner, a burst beyond the
606// visible slots waits for a free slot, and a box that goes away is blanked.
607func TestPickerControllerRenderNoticeStacksQueuesAndBlanks(t *testing.T) {
608 size := domain.Size{Cols: 80, Rows: 24}
609 notice := func(msg string) domain.Notification {
610 return domain.Notification{Code: domain.NoticeSessionKill, Severity: domain.NoticeInfo, Message: msg, Scope: msg}
611 }
612 tests := []struct {
613 name string
614 notices []string
615 advance time.Duration
616 wantShown []string
617 wantHidden []string
618 wantBlanked bool
619 }{
620 {name: "distinct subjects stack", notices: []string{"killed session 1", "killed session 2"}, wantShown: []string{"killed session 1", "killed session 2"}},
621 {name: "burst waits for a slot", notices: []string{"killed session 1", "killed session 2", "killed session 3", "killed session 4"}, wantShown: []string{"killed session 1", "killed session 2", "killed session 3"}, wantHidden: []string{"killed session 4"}},
622 {name: "expiry promotes the next notice", notices: []string{"killed session 1", "killed session 2", "killed session 3", "killed session 4"}, advance: pickerNoticeLifetime, wantShown: []string{"killed session 4"}, wantBlanked: true},
623 {name: "expiry of the last notice blanks it", notices: []string{"killed session 1"}, advance: pickerNoticeLifetime, wantHidden: []string{"killed session 1"}, wantBlanked: true},
624 }
625 for _, tt := range tests {
626 t.Run(tt.name, func(t *testing.T) {
627 controller, clock := pickerTestController(t)
628 for _, msg := range tt.notices {
629 controller.Notify(notice(msg))
630 }
631 first := string(controller.RenderNotice(size))
632 if tt.advance > 0 {
633 clock.mu.Lock()
634 clock.now = clock.now.Add(tt.advance)
635 clock.mu.Unlock()
636 }
637 got := first
638 if tt.advance > 0 {
639 got = string(controller.RenderNotice(size))
640 }
641 for _, msg := range tt.wantShown {
642 require.Contains(t, got, msg)
643 }
644 for _, msg := range tt.wantHidden {
645 require.NotContains(t, got, msg)
646 }
647 blank := strings.Repeat(" ", ui.ToastBounds(size, ui.Toast{Message: "killed session 1", Anchor: domain.AnchorTopRight}).Width)
648 require.Equal(t, tt.wantBlanked, tt.advance > 0 && strings.Contains(got, blank))
649 })
650 }
651}