1package sessionwire
2
3import (
4 "errors"
5 "io"
6 "testing"
7 "time"
8
9 "google.golang.org/protobuf/proto"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13 "github.com/bnema/vev/internal/protocol"
14 "github.com/bnema/vev/internal/protocol/wire"
15 "github.com/stretchr/testify/require"
16)
17
18func TestServerConnectionDecodesEveryClientMessage(t *testing.T) {
19 target := testTarget()
20 exact := protocol.ExactSessionTarget{LifecycleID: target.LifecycleID, SessionName: target.SessionName}
21 tests := []struct {
22 name string
23 want protocol.ClientMessage
24 }{
25 {name: "hello", want: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}}},
26 {name: "input", want: protocol.Input{InputSeq: 1, Data: []byte("x")}},
27 {name: "resize", want: protocol.Resize{Size: domain.Size{Cols: 80, Rows: 24}}},
28 {name: "detach", want: protocol.Detach{}},
29 {name: "closed detach", want: protocol.Detach{Closed: true}},
30 {name: "ping", want: protocol.Ping{}},
31 {name: "list", want: protocol.List{}},
32 {name: "kill", want: protocol.Kill{RequestID: 1, Name: "work"}},
33 {name: "theme", want: protocol.Theme{TrueColor: true}},
34 {name: "ack", want: protocol.Ack{Epoch: 1, State: 2}},
35 {name: "image", want: protocol.ImagePush{InputSeq: 2, Mime: "image/png", Data: []byte{1}}},
36 {name: "notice", want: protocol.ClientNotice{Action: protocol.ClientNoticeLinkConnected}},
37 {name: "command", want: protocol.CommandRequest{Version: protocol.Version, RequestID: 7, Slug: "list-sessions"}},
38 {name: "reset", want: protocol.OutputResetRequest{}},
39 {name: "select tab", want: protocol.SelectTab{TabID: "tab-2"}},
40 {name: "preview", want: protocol.RemotePreviewWatch{Request: protocol.RemotePreviewRequest{Version: protocol.RemotePreviewSchemaVersion, Target: target, Width: 1, Height: 1}, MinInterval: 33 * time.Millisecond}},
41 {name: "attention", want: protocol.RouteAttentionSubscription{}},
42 {name: "same peer", want: protocol.SamePeerSwitchRequest{RequestID: 3, Target: exact}},
43 {name: "snapshot", want: protocol.RecentRouteSnapshot{}},
44 {name: "route failure", want: protocol.RouteNavigationFailure{Key: 1, Generation: 1, Code: protocol.RouteFailureUnavailable}},
45 {name: "creation failure", want: protocol.SessionCreationFailure{RequestID: 1, Code: protocol.RouteFailureUnavailable}},
46 }
47 for _, tt := range tests {
48 t.Run(tt.name, func(t *testing.T) {
49 raw := &scriptedTransport{recv: mustServerPreambleQueue(t, mustEncodeClient(t, tt.want))}
50 got, err := NewServerConnection(raw).ReceiveClient()
51 require.NoError(t, err)
52 require.Equal(t, tt.want, got)
53 })
54 }
55}
56
57func TestServerConnectionEncodesEveryServerMessage(t *testing.T) {
58 exact := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "work"}
59 welcome := protocol.Welcome{SessionID: "session"}
60 errorMessage := protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "error"}
61 output := protocol.Output{Epoch: 1, New: 1, Full: true, Size: domain.Size{Cols: 80, Rows: 24}, Context: testUIOutputContext()}
62 receipt := protocol.UIReceipt{ActionID: 7, Epoch: 1, State: 1, ViewPublication: 1, Outcome: protocol.UIReceiptProcessed}
63 viewUpdate := protocol.UIViewUpdate{Epoch: 1, State: 1, Context: *testUIOutputContext()}
64 detached := protocol.Detached{Reason: protocol.ReasonDetach}
65 pong := protocol.Pong{}
66 sessions := protocol.Sessions{}
67 commandResult := protocol.CommandResult{RequestID: 1, Outcome: protocol.CommandSucceeded}
68 killResult := protocol.KillResult{RequestID: 1, Outcome: protocol.KillSucceeded}
69 killFailure := protocol.KillResult{RequestID: 2, Outcome: protocol.KillFailed, Code: protocol.ErrInternal, Text: "partial", Failures: []protocol.KillFailure{{Class: "stopped", Name: "work", Text: "boom"}}}
70 attachTarget := protocol.AttachTarget{Session: "work", Intent: protocol.IntentAttach}
71 preview := protocol.RemotePreview{Version: protocol.RemotePreviewSchemaVersion, Status: protocol.RemotePreviewUnavailable}
72 identity := protocol.CommittedRouteIdentity{Target: exact}
73 routeAction := protocol.RouteNavigationAction{SnapshotGeneration: 1, Key: 1, Generation: 1}
74 routeCreate := protocol.RouteCreateSessionAction{RequestID: 1, SnapshotGeneration: 1, Key: 1, Generation: 1, SessionName: "example"}
75 routeFailure := protocol.RouteNavigationFailure{Key: 1, Generation: 1, Code: protocol.RouteFailureUnavailable}
76 routePosition := protocol.RoutePosition{Target: exact, ActiveTabID: "tab-1"}
77 switchFailure := protocol.SamePeerSwitchFailure{RequestID: 1, Code: protocol.SamePeerSwitchUnavailable}
78
79 tests := []struct {
80 name string
81 message protocol.ServerMessage
82 }{
83 {name: "welcome", message: welcome},
84 {name: "error", message: errorMessage},
85 {name: "output", message: output},
86 {name: "receipt", message: receipt},
87 {name: "receipt pointer", message: &receipt},
88 {name: "view update", message: viewUpdate},
89 {name: "view update pointer", message: &viewUpdate},
90 {name: "detached", message: detached},
91 {name: "pong", message: pong},
92 {name: "sessions", message: sessions},
93 {name: "command result", message: commandResult},
94 {name: "kill result", message: killResult},
95 {name: "kill result partial failures", message: killFailure},
96 {name: "attach target", message: attachTarget},
97 {name: "preview", message: preview},
98 {name: "identity", message: identity},
99 {name: "route action", message: routeAction},
100 {name: "route create", message: routeCreate},
101 {name: "route create pointer", message: &routeCreate},
102 {name: "route failure", message: routeFailure},
103 {name: "route position", message: routePosition},
104 {name: "switch failure", message: switchFailure},
105 }
106 for _, tt := range tests {
107 t.Run(tt.name, func(t *testing.T) {
108 raw := &scriptedTransport{}
109 require.NoError(t, NewServerConnection(raw).SendServer(tt.message))
110 require.Equal(t, 1, raw.sentLen())
111 got, err := DecodeServerEnvelope(raw.sentPayload(0))
112 require.NoError(t, err)
113 switch message := tt.message.(type) {
114 case *protocol.UIReceipt:
115 require.Equal(t, *message, got)
116 case *protocol.UIViewUpdate:
117 require.Equal(t, *message, got)
118 case *protocol.RouteCreateSessionAction:
119 require.Equal(t, *message, got)
120 default:
121 require.Equal(t, tt.message, got)
122 }
123 })
124 }
125}
126
127func TestServerConnectionPreservesPayloadAgnosticClientControls(t *testing.T) {
128 // Empty control envelopes decode without payload-specific validation.
129 for _, tt := range []struct {
130 name string
131 message protocol.ClientMessage
132 }{
133 {name: "list", message: protocol.List{}},
134 {name: "detach", message: protocol.Detach{}},
135 {name: "ping", message: protocol.Ping{}},
136 } {
137 t.Run(tt.name, func(t *testing.T) {
138 raw := &scriptedTransport{recv: mustServerPreambleQueue(t, mustEncodeClient(t, tt.message))}
139 message, err := NewServerConnection(raw).ReceiveClient()
140 require.NoError(t, err)
141 require.Equal(t, tt.message, message)
142 })
143 }
144}
145
146func helloWithVersion(t *testing.T, version uint32) []byte {
147 t.Helper()
148 converted, err := helloToWire(protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach})
149 require.NoError(t, err)
150 converted.Version = version
151 raw, err := proto.Marshal(&wire.ClientEnvelope{
152 Payload: &wire.ClientEnvelope_Hello{Hello: converted},
153 })
154 require.NoError(t, err)
155 return raw
156}
157
158func TestServerConnectionClassifiesDecodeFailures(t *testing.T) {
159 validCommand, err := commandRequestToWire(protocol.CommandRequest{Version: protocol.Version, RequestID: 9, Slug: "list-sessions"})
160 require.NoError(t, err)
161 validCommandRaw, err := proto.Marshal(&wire.ClientEnvelope{
162 Payload: &wire.ClientEnvelope_CommandRequest{CommandRequest: validCommand},
163 })
164 require.NoError(t, err)
165 tests := []struct {
166 name string
167 payload []byte
168 category protocol.DecodeCategory
169 version uint16
170 request uint64
171 hasRequest bool
172 }{
173 {name: "malformed hello", payload: helloWithVersion(t, 38), category: protocol.DecodeMalformed, version: 38},
174 {name: "truncated command", payload: validCommandRaw[:1], category: protocol.DecodeMalformed},
175
176 {name: "unknown", payload: []byte{0xf8, 0x07}, category: protocol.DecodeUnknownType},
177 }
178 for _, tt := range tests {
179 t.Run(tt.name, func(t *testing.T) {
180 raw := &scriptedTransport{recv: mustServerPreambleQueue(t, tt.payload)}
181 _, err := NewServerConnection(raw).ReceiveClient()
182 var failure *protocol.DecodeFailure
183 require.ErrorAs(t, err, &failure)
184 require.Equal(t, tt.category, failure.Category)
185 require.Equal(t, tt.version, failure.Version)
186 require.Equal(t, tt.request, failure.RequestID)
187 require.Equal(t, tt.hasRequest, failure.HasRequestID)
188 })
189 }
190}
191
192func TestServerConnectionPreservesSendModesCapabilitiesAndErrors(t *testing.T) {
193 sendErr := errors.New("send failed")
194 events := make(chan ports.LinkEvent, 1)
195 raw := &capableTransport{scriptedTransport: scriptedTransport{sendErr: sendErr}, events: events}
196 connection := NewServerConnection(raw)
197
198 caps := connection.Capabilities()
199 require.Equal(t, uint8(1), caps.PreferredOutputWindow)
200 require.Equal(t, protocol.MaxOutputDataLen, caps.OutputDataLimit)
201 require.True(t, caps.AsyncSend)
202 require.True(t, caps.OwnedSynchronousSend)
203 require.True(t, caps.LinkState)
204 require.Equal(t, ports.LinkStateDegraded, connection.LinkState())
205 require.Equal(t, (<-chan ports.LinkEvent)(events), connection.LinkEvents())
206
207 message := protocol.Pong{}
208 require.ErrorIs(t, connection.SendServer(message), sendErr)
209 require.Equal(t, 1, raw.sentLen())
210 got, err := DecodeServerEnvelope(raw.sentPayload(0))
211 require.NoError(t, err)
212 require.Equal(t, message, got)
213 require.ErrorIs(t, connection.SendServerAsync(message), sendErr)
214 require.Equal(t, 1, len(raw.async))
215 require.ErrorIs(t, connection.SendServerSynchronous(message), sendErr)
216 require.Equal(t, 1, len(raw.synchronous))
217
218 output := protocol.Output{Epoch: 1, New: 1, Full: true, Size: domain.Size{Cols: 80, Rows: 24}, Context: testUIOutputContext()}
219 require.ErrorIs(t, connection.SendOutput(output), sendErr)
220 require.ErrorIs(t, connection.SendOutputAsync(output), sendErr)
221 require.ErrorIs(t, connection.SendOutputSynchronous(output), sendErr)
222
223 plain := NewServerConnection(&scriptedTransport{})
224 require.ErrorIs(t, plain.SendServerAsync(message), ErrUnsupportedSend)
225 require.ErrorIs(t, plain.SendServerSynchronous(message), ErrUnsupportedSend)
226 require.ErrorIs(t, plain.SendOutputAsync(output), ErrUnsupportedSend)
227 require.ErrorIs(t, plain.SendOutputSynchronous(output), ErrUnsupportedSend)
228 require.Equal(t, uint8(protocol.MaxOutputWindow), plain.Capabilities().PreferredOutputWindow)
229 require.Equal(t, ports.LinkStateConnected, plain.LinkState())
230 require.Nil(t, plain.LinkEvents())
231}
232
233func TestServerConnectionForwardsReceiveAndCloseErrors(t *testing.T) {
234 recvErr := errors.Join(io.EOF, errors.New("closed"))
235 closeErr := errors.New("close failed")
236 // The queued preamble request is consumed, then the preamble
237 // itself fails on the joined error; the failure surfaces as a
238 // typed decode failure chaining io.EOF, and the failed attempt
239 // closes the transport before the explicit Close.
240 raw := &scriptedTransport{recv: mustServerPreambleQueue(t), recvErr: recvErr, closeErr: closeErr}
241 connection := NewServerConnection(raw)
242 _, err := connection.ReceiveClient()
243 require.ErrorIs(t, err, io.EOF)
244 var failure *protocol.DecodeFailure
245 require.ErrorAs(t, err, &failure)
246 require.Equal(t, protocol.DecodeMalformed, failure.Category)
247 require.ErrorIs(t, connection.Close(), closeErr)
248 require.Equal(t, 2, raw.closeCalls())
249}
250
251func TestServerListenerWrapsEachAcceptedIncarnation(t *testing.T) {
252 raw := &scriptedTransport{recv: mustServerPreambleQueue(t, mustEncodeClient(t, protocol.Ping{}))}
253 listener := &scriptedListener{transport: raw}
254 wrapped := NewServerListener(listener)
255 first, err := wrapped.Accept()
256 require.NoError(t, err)
257 message, err := first.ReceiveClient()
258 require.NoError(t, err)
259 require.Equal(t, protocol.Ping{}, message)
260 require.Equal(t, "test", wrapped.Addr())
261 require.NoError(t, wrapped.Close())
262}
263
264type scriptedListener struct{ transport wire.Transport }
265
266func (l *scriptedListener) Accept() (wire.Transport, error) { return l.transport, nil }
267func (*scriptedListener) Close() error { return nil }
268func (*scriptedListener) Addr() string { return "test" }