1package client_test
2
3import (
4 "bytes"
5 "context"
6 "reflect"
7 "sync"
8 "sync/atomic"
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 portsmocks "github.com/bnema/vev/internal/ports/mocks"
17 "github.com/bnema/vev/internal/usecase/client"
18)
19
20func TestRuntimeObserverDependencyUsesSerializedContract(t *testing.T) {
21 deps := reflect.TypeFor[client.Dependencies]()
22 field, ok := deps.FieldByName("RuntimeObserver")
23 if !ok || field.Type != reflect.TypeFor[ports.SerializedRuntimeObserver]() {
24 t.Fatalf("Dependencies.RuntimeObserver is %v, want %v; raw blocking sinks must not enter the hot path", field.Type, reflect.TypeFor[ports.SerializedRuntimeObserver]())
25 }
26}
27
28func TestBlockingRuntimeObserverDoesNotDelayTerminalFlushOrACK(t *testing.T) {
29 var out bytes.Buffer
30 var restores atomic.Int32
31 resizeCh := make(chan domain.Size)
32 flushed := make(chan struct{})
33 term := portsmocks.NewMockTerminal(t)
34 term.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
35 term.EXPECT().EnterRaw().Return(func() error { restores.Add(1); return nil }, nil).Once()
36 in := newBlockingReader()
37 defer in.unblock()
38 term.EXPECT().In().Return(in).Maybe()
39 term.EXPECT().Out().Return(&out).Maybe()
40 term.EXPECT().Flush().Run(func() { closeOnce(flushed) }).Return(nil).Maybe()
41 term.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
42
43 observer := &blockingClientRuntimeObserver{entered: make(chan struct{}), release: make(chan struct{})}
44 reporter := ports.NewSerializedRuntimeObserver(observer, 1)
45 defer reporter.Close()
46 acked := make(chan struct{})
47 transport := portsmocks.NewMockTransport(t)
48 transport.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
49 transport.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
50 transport.EXPECT().Send(isType(ports.MsgAck)).Run(func(ports.Frame) { close(acked) }).Return(nil).Once()
51 unblock := scriptRecv(transport,
52 recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
53 recvItem{f: frameOf(ports.MsgOutput, mustMarshalOutput(ports.Output{Epoch: 1, Size: domain.Size{Cols: 1, Rows: 1}, Full: true, Data: []byte("flush-before-observe"), New: 3}))},
54 )
55 defer unblock()
56 transport.EXPECT().Close().Return(nil).Once()
57
58 ctx, cancel := context.WithCancel(context.Background())
59 result := make(chan error, 1)
60 go func() {
61 result <- runTestClient(ctx, testDependencies(transportDialer{transport: transport}, term, realClock{}, nil, reporter), client.AttachRequest{Intent: ports.IntentEphemeral})
62 }()
63 awaitClientRuntime(t, observer.entered, "blocking observer")
64 awaitClientRuntime(t, flushed, "terminal flush")
65 awaitClientRuntime(t, acked, "ACK")
66 cancel()
67 close(observer.release)
68 select {
69 case err := <-result:
70 require.NoError(t, err)
71 case <-time.After(time.Second):
72 t.Fatal("Attach did not complete after observer release")
73 }
74 require.Equal(t, "\x1b]10;?\x07\x1b]11;?\x07\x1b]4;0;?;1;?;2;?;3;?;4;?;5;?;6;?;7;?;8;?;9;?;10;?;11;?;12;?;13;?;14;?;15;?\x07\x1b[?2031$pflush-before-observe", out.String())
75 require.Equal(t, int32(1), restores.Load())
76}
77
78func TestTerminalFlushBoundaryTransportObservability(t *testing.T) {
79 var out bytes.Buffer
80 var restores atomic.Int32
81 resizeCh := make(chan domain.Size)
82 term, in := newHappyTerminal(t, &out, &restores, resizeCh)
83 defer in.unblock()
84
85 observer := &clientRuntimeObserver{}
86 reporter := ports.NewSerializedRuntimeObserver(observer, 64)
87 defer reporter.Close()
88 transport := portsmocks.NewMockTransport(t)
89 transport.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
90 transport.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
91 unblock := scriptRecv(transport,
92 recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
93 recvItem{f: frameOf(ports.MsgOutput, mustMarshalOutput(ports.Output{Epoch: 1, Size: domain.Size{Cols: 1, Rows: 1}, Full: true, Data: []byte("unchanged-by-observer"), New: 3}))},
94 recvItem{f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
95 )
96 defer unblock()
97 transport.EXPECT().Send(isType(ports.MsgAck)).Return(nil).Maybe()
98 transport.EXPECT().Close().Return(nil).Once()
99
100 err := runTestClient(context.Background(), testDependencies(transportDialer{transport: transport}, term, realClock{}, nil, reporter), client.AttachRequest{Intent: ports.IntentEphemeral})
101 require.NoError(t, err)
102 reporter.Flush()
103 require.Equal(t, "\x1b]10;?\x07\x1b]11;?\x07\x1b]4;0;?;1;?;2;?;3;?;4;?;5;?;6;?;7;?;8;?;9;?;10;?;11;?;12;?;13;?;14;?;15;?\x07\x1b[?2031$punchanged-by-observer", out.String(), "observer must not alter terminal bytes")
104 require.Equal(t, int32(1), restores.Load())
105 // Carriage spans belong only to concrete adapters. The client owns the
106 // post-successful-flush terminal boundary, before its ACK is queued.
107 observer.requireOrdered(t, ports.RuntimeTerminalFlushed)
108}
109
110type blockingClientRuntimeObserver struct {
111 entered chan struct{}
112 release chan struct{}
113 once sync.Once
114}
115
116func (o *blockingClientRuntimeObserver) ObserveRuntime(ports.RuntimeMark) {
117 o.once.Do(func() {
118 close(o.entered)
119 <-o.release
120 })
121}
122
123func awaitClientRuntime(t *testing.T, ch <-chan struct{}, what string) {
124 t.Helper()
125 select {
126 case <-ch:
127 case <-time.After(time.Second):
128 t.Fatalf("timed out waiting for %s", what)
129 }
130}
131
132type clientRuntimeObserver struct {
133 mu sync.Mutex
134 marks []ports.RuntimeMark
135}
136
137func (o *clientRuntimeObserver) ObserveRuntime(mark ports.RuntimeMark) {
138 o.mu.Lock()
139 defer o.mu.Unlock()
140 o.marks = append(o.marks, mark)
141}
142
143func (o *clientRuntimeObserver) requireOrdered(t *testing.T, kinds ...ports.RuntimeMarkKind) {
144 t.Helper()
145 o.mu.Lock()
146 defer o.mu.Unlock()
147 next := 0
148 for _, mark := range o.marks {
149 if next < len(kinds) && mark.Kind == kinds[next] {
150 next++
151 }
152 }
153 if next != len(kinds) {
154 t.Fatalf("runtime marks %#v do not contain ordered %#v", o.marks, kinds)
155 }
156}