1package client
2
3import (
4 "context"
5 "testing"
6 "time"
7
8 "github.com/bnema/vev/internal/ports"
9 "github.com/stretchr/testify/require"
10)
11
12func TestResumeCancelDuringStreamOpen(t *testing.T) {
13 for _, tt := range []struct {
14 name, input string
15 escape bool
16 }{
17 {name: "ctrl-c", input: "\x03"},
18 {name: "kitty escape", input: "\x1b[27u"},
19 {name: "legacy escape", input: "\x1b", escape: true},
20 } {
21 t.Run(tt.name, func(t *testing.T) {
22 picker := newAttachTestPicker()
23 h := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { cfg.Jitter = resumeTestJitter })
24 first := newSessionTestStream()
25 opened := make(chan struct{})
26 cancelled := make(chan struct{})
27 count := 0
28 h.service.setOpenStream(func(ctx context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
29 count++
30 if count == 1 {
31 return first, nil
32 }
33 close(opened)
34 <-ctx.Done()
35 close(cancelled)
36 return nil, ctx.Err()
37 })
38 picker.commit(sessionTestRequest(false))
39 deliverReadyStream(t, first)
40 awaitAttachedState(t, h.sup)
41 first.fail(testStreamLoss())
42 fireResumeTimerAfter(t, h.clock, 1, func() {})
43 select {
44 case <-opened:
45 case <-time.After(5 * time.Second):
46 t.Fatal("resume did not open")
47 }
48 h.reader.push([]byte(tt.input))
49 if tt.escape {
50 for {
51 timer := h.clock.awaitTimer(t)
52 if timer.delay == 50*time.Millisecond {
53 timer.fire()
54 break
55 }
56 }
57 }
58 select {
59 case <-cancelled:
60 case <-time.After(5 * time.Second):
61 t.Fatal("cancel did not reach open context")
62 }
63 require.Eventually(t, picker.owns, 5*time.Second, time.Millisecond)
64 require.Equal(t, 2, len(h.service.openedRequests()))
65 })
66 }
67}
68
69// TestResumeHeldKeysNeverReachThePicker types while a lost attachment
70// resumes, then returns to the picker either by cancelling or because the
71// resume fails for good, and proves the picker never receives the held keys
72// while it still receives keys typed after it took the terminal back.
73func TestResumeHeldKeysNeverReachThePicker(t *testing.T) {
74 for _, tt := range []struct {
75 name string
76 cancel bool
77 }{
78 {name: "cancelled", cancel: true},
79 {name: "resume refused"},
80 } {
81 t.Run(tt.name, func(t *testing.T) {
82 assertHeldKeysDroppedAtPicker(t, tt.cancel)
83 })
84 }
85}
86
87func assertHeldKeysDroppedAtPicker(t *testing.T, cancel bool) {
88 picker := newAttachTestPicker()
89 h := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { cfg.Jitter = resumeTestJitter })
90 first := newSessionTestStream()
91 opened := make(chan struct{}, 1)
92 refuse := make(chan struct{})
93 count := 0
94 h.service.setOpenStream(func(ctx context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
95 count++
96 if count == 1 {
97 return first, nil
98 }
99 opened <- struct{}{}
100 select {
101 case <-ctx.Done():
102 return nil, ctx.Err()
103 case <-refuse:
104 return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Text: "refused"}
105 }
106 })
107 picker.commit(sessionTestRequest(false))
108 deliverReadyStream(t, first)
109 awaitAttachedState(t, h.sup)
110 first.fail(testStreamLoss())
111 require.Eventually(t, func() bool { return h.sup.State().Presentation == PresentConnecting }, 5*time.Second, time.Millisecond)
112 pushKeys(t, h, "HELD\r")
113 if cancel {
114 h.reader.push([]byte("\x03"))
115 } else {
116 fireResumeTimerAfter(t, h.clock, 1, func() {})
117 select {
118 case <-opened:
119 case <-time.After(5 * time.Second):
120 t.Fatal("resume did not open")
121 }
122 close(refuse)
123 }
124 require.Eventually(t, picker.owns, 5*time.Second, time.Millisecond)
125 h.reader.push([]byte("j"))
126 require.Eventually(t, func() bool {
127 picker.mu.Lock()
128 defer picker.mu.Unlock()
129 return len(picker.consumedBytes) > 0
130 }, 5*time.Second, time.Millisecond)
131 picker.mu.Lock()
132 defer picker.mu.Unlock()
133 require.Equal(t, "j", string(picker.consumedBytes), "held resume keys reached the picker")
134}