1package sessionwire
2
3import (
4 "io"
5 "sync"
6 "testing"
7
8 "google.golang.org/protobuf/proto"
9
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"
12 "github.com/bnema/vev/internal/protocol"
13 "github.com/bnema/vev/internal/protocol/wire"
14 "github.com/stretchr/testify/require"
15)
16
17// scriptedTransport is an in-memory wire.Transport. Recv serves queued
18// envelopes in order; tests interleave canned preamble envelopes via
19// mustClientPreambleQueue/mustServerPreambleQueue so typed connections
20// complete their lazy preamble before application traffic.
21//
22// The mutex exists because the preamble runs in a spawned goroutine: a
23// test that abandons the preamble (deadline, refused response) must not
24// race the goroutine's trailing Send/Close. Never read sent/recv/closeCall
25// after an abandoned preamble without holding the mutex via sentLen or
26// closeCalls.
27type scriptedTransport struct {
28 mu sync.Mutex
29 recv []wire.Envelope
30 recvErr error
31 sent []wire.Envelope
32 sendErr error
33 closeErr error
34 closeCall int
35}
36
37func (t *scriptedTransport) Send(envelope wire.Envelope) error {
38 t.mu.Lock()
39 defer t.mu.Unlock()
40 t.sent = append(t.sent, envelope)
41 return t.sendErr
42}
43
44func (t *scriptedTransport) Recv() (wire.Envelope, error) {
45 t.mu.Lock()
46 defer t.mu.Unlock()
47 if len(t.recv) == 0 {
48 if t.recvErr != nil {
49 return wire.Envelope{}, t.recvErr
50 }
51 return wire.Envelope{}, io.EOF
52 }
53 next := t.recv[0]
54 t.recv = t.recv[1:]
55 return next, nil
56}
57
58func (t *scriptedTransport) Close() error {
59 t.mu.Lock()
60 defer t.mu.Unlock()
61 t.closeCall++
62 return t.closeErr
63}
64
65// sentLen reports recorded sends for assertions after the preamble
66// goroutine has joined (successful preamble or connection close).
67func (t *scriptedTransport) sentLen() int {
68 t.mu.Lock()
69 defer t.mu.Unlock()
70 return len(t.sent)
71}
72
73// sentPayload returns the payload of one recorded send.
74func (t *scriptedTransport) sentPayload(index int) []byte {
75 t.mu.Lock()
76 defer t.mu.Unlock()
77 return t.sent[index].Payload
78}
79
80// closeCalls reports recorded closes.
81func (t *scriptedTransport) closeCalls() int {
82 t.mu.Lock()
83 defer t.mu.Unlock()
84 return t.closeCall
85}
86
87func mustPreambleRequest(t *testing.T) wire.Envelope {
88 t.Helper()
89 raw, err := proto.Marshal(preambleRequestToWire(defaultProtoCeilings()))
90 require.NoError(t, err)
91 return wire.Envelope{Payload: raw}
92}
93
94func mustPreambleResponse(t *testing.T) wire.Envelope {
95 t.Helper()
96 raw, err := proto.Marshal(preambleResponseToWire(true, defaultProtoCeilings(), 0))
97 require.NoError(t, err)
98 return wire.Envelope{Payload: raw}
99}
100
101// mustClientPreambleQueue interleaves one server PreambleResponse before
102// every queued application envelope: each fresh client connection on the
103// transport completes its preamble first. Never reuse one transport for
104// two connections expecting a single preamble: every ReceiveServer/SendClient
105// on a NEW connection consumes the next canned response.
106func mustClientPreambleQueue(t *testing.T, payloads ...[]byte) []wire.Envelope {
107 t.Helper()
108 queue := make([]wire.Envelope, 0, 2*len(payloads))
109 for _, payload := range payloads {
110 queue = append(queue, mustPreambleResponse(t), wire.Envelope{Payload: payload})
111 }
112 return queue
113}
114
115// mustServerPreambleQueue interleaves one client PreambleRequest before
116// every queued application envelope.
117func mustServerPreambleQueue(t *testing.T, payloads ...[]byte) []wire.Envelope {
118 t.Helper()
119 queue := make([]wire.Envelope, 0, 2*len(payloads))
120 for _, payload := range payloads {
121 queue = append(queue, mustPreambleRequest(t), wire.Envelope{Payload: payload})
122 }
123 return queue
124}
125
126func mustEncodeClient(t *testing.T, message protocol.ClientMessage) []byte {
127 t.Helper()
128 raw, err := EncodeClientMessage(message)
129 require.NoError(t, err)
130 return raw
131}
132
133func mustEncodeServer(t *testing.T, message protocol.ServerMessage) []byte {
134 t.Helper()
135 raw, err := EncodeServerMessage(message)
136 require.NoError(t, err)
137 return raw
138}
139
140// mustSingleAppPayload asserts a transport recorded exactly the preamble
141// exchange plus one application envelope, and returns that payload.
142func mustSingleAppPayload(t *testing.T, raw *scriptedTransport) []byte {
143 t.Helper()
144 require.Len(t, raw.sent, 2, "want preamble request plus one application envelope")
145 return raw.sentPayload(1)
146}
147
148type capableTransport struct {
149 scriptedTransport
150 async []wire.Envelope
151 synchronous []wire.Envelope
152 events chan ports.LinkEvent
153}
154
155func (*capableTransport) DatagramTransport() {}
156func (t *capableTransport) SendAsync(envelope wire.Envelope) error {
157 t.scriptedTransport.mu.Lock()
158 defer t.scriptedTransport.mu.Unlock()
159 t.async = append(t.async, envelope)
160 return t.sendErr
161}
162func (t *capableTransport) SendSynchronous(envelope wire.Envelope) error {
163 t.scriptedTransport.mu.Lock()
164 defer t.scriptedTransport.mu.Unlock()
165 t.synchronous = append(t.synchronous, envelope)
166 return t.sendErr
167}
168func (*capableTransport) LinkState() ports.LinkState { return ports.LinkStateDegraded }
169func (t *capableTransport) LinkEvents() <-chan ports.LinkEvent { return t.events }
170
171func testTarget() domain.RemoteSessionTarget {
172 return domain.RemoteSessionTarget{
173 Endpoint: "remote", DisplayOrigin: "remote", LifecycleID: domain.SessionLifecycleID{1},
174 SessionName: "work", LiveTabID: "tab-1",
175 }
176}