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