1package client
2
3import (
4 "context"
5 "errors"
6 "io"
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/domain/terminalcap"
16 "github.com/bnema/vev/internal/ports"
17 "github.com/bnema/vev/internal/protocol"
18)
19
20// Real attachment worker tests (Plan 001 P5.3b). They are deterministic and
21// drive the typed session protocol directly: no wall-clock sleeps, no real
22// terminal, and no broker process.
23
24var errSessionTestClosed = errors.New("session test stream is closed")
25
26// sessionTestStream is one scripted broker logical connection. It records every
27// typed client message, delivers scripted server messages, and unblocks
28// ReceiveServer on Close exactly as the ports.BrokerLogicalConnection contract
29// requires.
30type sessionTestStream struct {
31 mu sync.Mutex
32 sent []protocol.ClientMessage
33 incoming chan protocol.ServerMessage
34 closed chan struct{}
35 closeOnce sync.Once
36 err error
37}
38
39func newSessionTestStream() *sessionTestStream {
40 return &sessionTestStream{
41 incoming: make(chan protocol.ServerMessage, 8),
42 closed: make(chan struct{}),
43 }
44}
45
46func (s *sessionTestStream) SendClient(message protocol.ClientMessage) error {
47 select {
48 case <-s.closed:
49 return errSessionTestClosed
50 default:
51 }
52 s.mu.Lock()
53 s.sent = append(s.sent, message)
54 s.mu.Unlock()
55 return nil
56}
57
58func (s *sessionTestStream) ReceiveServer() (protocol.ServerMessage, error) {
59 select {
60 case message := <-s.incoming:
61 return message, nil
62 case <-s.closed:
63 s.mu.Lock()
64 err := s.err
65 s.mu.Unlock()
66 if err != nil {
67 return nil, errors.Join(io.EOF, err)
68 }
69 return nil, io.EOF
70 }
71}
72
73func (s *sessionTestStream) Capabilities() protocol.ConnectionCapabilities {
74 return protocol.ConnectionCapabilities{}
75}
76
77func (s *sessionTestStream) LinkState() ports.LinkState { return ports.LinkStateConnected }
78
79func (s *sessionTestStream) LinkEvents() <-chan ports.LinkEvent { return nil }
80
81func (s *sessionTestStream) Close() error {
82 s.closeOnce.Do(func() { close(s.closed) })
83 return nil
84}
85
86func (s *sessionTestStream) Done() <-chan struct{} { return s.closed }
87
88func (s *sessionTestStream) Err() error {
89 s.mu.Lock()
90 defer s.mu.Unlock()
91 return s.err
92}
93
94type blockingSendSessionStream struct {
95 *sessionTestStream
96 entered chan struct{}
97 finished chan struct{}
98}
99
100func (s *blockingSendSessionStream) SendClient(protocol.ClientMessage) error {
101 close(s.entered)
102 <-s.closed
103 close(s.finished)
104 return errSessionTestClosed
105}
106
107func TestSessionAttachmentSendCancellationClosesStreamAndJoinsSend(t *testing.T) {
108 stream := &blockingSendSessionStream{sessionTestStream: newSessionTestStream(), entered: make(chan struct{}), finished: make(chan struct{})}
109 fg := &attachmentForeground{done: make(chan struct{})}
110 ctx, cancel := context.WithCancel(context.Background())
111 returned := make(chan error, 1)
112 go func() {
113 returned <- (&sessionAttachmentWorker{}).send(ctx, fg, stream, protocol.Detach{})
114 }()
115
116 <-stream.entered
117 cancel()
118 require.ErrorIs(t, <-returned, context.Canceled)
119 select {
120 case <-stream.Done():
121 default:
122 t.Fatal("cancellation did not close stream")
123 }
124 select {
125 case <-stream.finished:
126 default:
127 t.Fatal("send goroutine was not joined")
128 }
129}
130
131func TestSessionAttachmentSettleReportsClosedProcess(t *testing.T) {
132 tests := []struct {
133 name string
134 cause error
135 wantClosed bool
136 }{
137 {name: "closed process sends closed detach", cause: ErrClientClosed, wantClosed: true},
138 {name: "plain cancellation sends nothing", cause: context.Canceled},
139 }
140 for _, tt := range tests {
141 t.Run(tt.name, func(t *testing.T) {
142 stream := newSessionTestStream()
143 ctx, cancel := context.WithCancelCause(context.Background())
144 cancel(tt.cause)
145 w := &sessionAttachmentWorker{cfg: sessionAttachmentConfig{Clock: systemClock{}}}
146
147 event := w.settle(ctx, &attachmentForeground{done: make(chan struct{})}, stream, AttachmentToken{}, nil)
148
149 require.Equal(t, AttachmentEventFailed, event.Kind)
150 stream.mu.Lock()
151 defer stream.mu.Unlock()
152 if tt.wantClosed {
153 require.Equal(t, []protocol.ClientMessage{protocol.Detach{Closed: true}}, stream.sent)
154 } else {
155 require.Empty(t, stream.sent)
156 }
157 })
158 }
159}
160
161func (s *sessionTestStream) deliver(message protocol.ServerMessage) {
162 select {
163 case s.incoming <- message:
164 case <-s.closed:
165 }
166}
167
168// fail publishes the stable stream cause and closes, exactly as a transported
169// logical stream does on loss.
170func (s *sessionTestStream) fail(err error) {
171 s.mu.Lock()
172 if s.err == nil {
173 s.err = err
174 }
175 s.mu.Unlock()
176 _ = s.Close()
177}
178
179func (s *sessionTestStream) closedNow() bool {
180 select {
181 case <-s.closed:
182 return true
183 default:
184 return false
185 }
186}
187
188func (s *sessionTestStream) messages() []protocol.ClientMessage {
189 s.mu.Lock()
190 defer s.mu.Unlock()
191 return append([]protocol.ClientMessage(nil), s.sent...)
192}
193
194// sessionTestStreamWithDeadline additionally implements
195// ports.HandshakeDeadlineProvider, so the supervisor-level deadline tests can
196// prove an accepted absolute deadline is adopted verbatim.
197type sessionTestStreamWithDeadline struct {
198 *sessionTestStream
199 deadline time.Time
200}
201
202func (s *sessionTestStreamWithDeadline) HandshakeDeadline() time.Time { return s.deadline }
203
204func sessionTestOutput(publication uint64, data string) protocol.Output {
205 return protocol.Output{
206 Epoch: 1, New: 1, Full: true,
207 Size: domain.Size{Cols: 80, Rows: 24},
208 Context: &protocol.ViewContext{
209 Publication: publication,
210 Route: protocol.CommittedRouteIdentity{Target: protocol.ExactSessionTarget{
211 LifecycleID: domain.SessionLifecycleID{1},
212 SessionName: "alpha",
213 }},
214 TabID: "t_abc123",
215 FocusedPaneID: "p_def456",
216 },
217 Data: []byte(data),
218 }
219}
220
221func sessionTestRequest(local bool) ports.BrokerOpenStreamRequest {
222 request := ports.BrokerOpenStreamRequest{
223 Epoch: 3,
224 Purpose: ports.BrokerStreamAttachment,
225 Admission: ports.BrokerAdmissionExact,
226 Local: local,
227 Target: protocol.ExactSessionTarget{
228 LifecycleID: domain.SessionLifecycleID{1},
229 SessionName: "alpha",
230 },
231 Policy: ports.BrokerPolicy{
232 ProtocolVersion: protocol.Version,
233 CatalogSchemaVersion: 1,
234 EnvironmentPolicy: protocol.EnvironmentPolicyClientOwned,
235 Transport: "transport",
236 Trust: "trust",
237 Launch: "launch",
238 Isolation: "isolation",
239 },
240 }
241 if !local {
242 request.Endpoint = "user@example"
243 request.Registration = domain.RemoteRegistration{
244 Endpoint: "user@example",
245 Incarnation: [16]byte{1},
246 Generation: 1,
247 }
248 }
249 return request
250}
251
252func sessionTestWorkerConfig(request ports.BrokerOpenStreamRequest) sessionAttachmentConfig {
253 return sessionAttachmentConfig{
254 Request: request,
255 ClientID: [16]byte{0xAB},
256 Geometry: domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}},
257 SessionEnvironment: SessionEnvironment{Provenance: SessionEnvironmentLocalCLI, Env: append([]string(nil), request.Env...)},
258 }
259}
260
261// sessionTestAttachments records every successful MarkAttached without
262// draining: a wait observes the transition but never consumes it, so the test
263// can assert attachment after a script already synchronized on it.
264type sessionTestAttachments struct {
265 count atomic.Int64
266 tokens chan AttachmentToken
267}
268
269func newSessionTestAttachments() *sessionTestAttachments {
270 return &sessionTestAttachments{tokens: make(chan AttachmentToken, 8)}
271}
272
273func (a *sessionTestAttachments) record(token AttachmentToken) {
274 a.count.Add(1)
275 select {
276 case a.tokens <- token:
277 default:
278 }
279}
280
281func (a *sessionTestAttachments) attached() bool { return a.count.Load() > 0 }
282
283func (a *sessionTestAttachments) token(t *testing.T) AttachmentToken {
284 t.Helper()
285 select {
286 case token := <-a.tokens:
287 return token
288 case <-time.After(5 * time.Second):
289 t.Fatal("the worker never attached")
290 return AttachmentToken{}
291 }
292}
293
294// sessionTestRun boots one worker through the real foreground host.
295func sessionTestRun(t *testing.T, stream ports.BrokerLogicalConnection, terminal *workerTestTerminal, cfg sessionAttachmentConfig, attached *sessionTestAttachments) <-chan AttachmentEvent {
296 t.Helper()
297 host := newWorkerTestHost(terminal, nil, attached.record)
298 worker, err := newSessionAttachmentWorker(cfg)
299 require.NoError(t, err)
300 events := make(chan AttachmentEvent, 1)
301 go func() {
302 event, _ := host.Run(context.Background(), AttachmentToken{Generation: 4, Attempt: 1}, worker, stream)
303 events <- event
304 }()
305 return events
306}
307
308// awaitHello blocks until the worker sent its Hello.
309func awaitHello(t *testing.T, stream *sessionTestStream) protocol.Hello {
310 t.Helper()
311 deadline := time.After(5 * time.Second)
312 for {
313 for _, message := range stream.messages() {
314 if hello, ok := message.(protocol.Hello); ok {
315 return hello
316 }
317 }
318 select {
319 case <-deadline:
320 t.Fatal("the worker never sent Hello")
321 return protocol.Hello{}
322 case <-time.After(time.Millisecond):
323 }
324 }
325}
326
327// TestSessionAttachmentWorkerStaysConnectingUntilPublication proves Welcome is
328// never attachment: only a validated, committed initial publication marks
329// attached, and the frame is written and flushed through the UI transaction.
330func TestSessionAttachmentWorkerStaysConnectingUntilPublication(t *testing.T) {
331 stream := newSessionTestStream()
332 terminal := newWorkerTestTerminal()
333 attached := newSessionTestAttachments()
334 events := sessionTestRun(t, stream, terminal, sessionTestWorkerConfig(sessionTestRequest(true)), attached)
335
336 hello := awaitHello(t, stream)
337 require.Equal(t, protocol.Version, hello.Version)
338 require.Equal(t, protocol.IntentAttach, hello.Intent)
339 require.NotNil(t, hello.ExactTarget)
340 require.False(t, hello.Remote)
341 // An exact attach carries its target's session name: the strict Hello
342 // contract requires Name to equal the exact target's name, so an empty Name
343 // would be rejected before the daemon ever routes it.
344 require.Equal(t, hello.ExactTarget.SessionName, hello.Name)
345 require.NoError(t, protocol.ValidateHello(hello))
346
347 stream.deliver(protocol.Welcome{SessionName: "alpha"})
348 // Welcome alone must not attach, write, or publish.
349 require.Never(t, attached.attached, 20*time.Millisecond, 2*time.Millisecond, "Welcome alone must never attach")
350 require.Zero(t, terminal.flushCount(), "Welcome alone writes nothing")
351 require.Empty(t, terminal.publications(), "Welcome alone publishes nothing")
352
353 stream.deliver(sessionTestOutput(1, "\x1b[Hready"))
354 require.Equal(t, AttachmentToken{Generation: 4, Attempt: 1}, attached.token(t))
355 require.Equal(t, "\x1b[Hready", terminal.written())
356 require.Equal(t, 1, terminal.flushCount())
357 require.Len(t, terminal.publications(), 2)
358 // The frame commits the pre-attach Connecting presentation with a public
359 // generation of zero, then the committed Attached presentation carries the
360 // real action generation without writing a second frame.
361 require.Equal(t, ports.UIStatusConnecting, terminal.publications()[0].Status)
362 require.Zero(t, terminal.publications()[0].Generation)
363 require.Equal(t, ports.UIStatusAttached, terminal.publications()[1].Status)
364 require.Equal(t, uint64(4), terminal.publications()[1].Generation)
365 require.Equal(t, 1, terminal.flushCount(), "the attached presentation writes no second frame")
366 require.Eventually(t, func() bool {
367 for _, message := range stream.messages() {
368 if ack, ok := message.(protocol.Ack); ok {
369 return ack == (protocol.Ack{Epoch: 1, State: 1})
370 }
371 }
372 return false
373 }, time.Second, time.Millisecond, "committed initial output must release the daemon output window")
374
375 require.NoError(t, stream.Close())
376 select {
377 case event := <-events:
378 require.Equal(t, AttachmentEventEnded, event.Kind)
379 require.Equal(t, AttachmentToken{Generation: 4, Attempt: 1}, event.Token)
380 case <-time.After(5 * time.Second):
381 t.Fatal("the worker never settled after the stream closed")
382 }
383}
384
385// TestSessionAttachmentWorkerFailureStages is the EOF/refusal coverage table:
386// before Welcome, after Welcome but before publication, and after attachment,
387// plus a protocol refusal and a publication failure.
388func TestSessionAttachmentWorkerFailureStages(t *testing.T) {
389 lossErr := ports.BrokerStreamLost{
390 Connection: ports.BrokerConnectionID{1},
391 Stream: 1,
392 Epoch: 3,
393 Cause: domain.RemoteFailureTransport,
394 Err: errors.New("transport lost"),
395 }
396 publishErr := errors.New("ui publication failed")
397
398 tests := []struct {
399 name string
400 script func(stream *sessionTestStream, terminal *workerTestTerminal, attached *sessionTestAttachments)
401 wantKind AttachmentEventKind
402 wantAttached bool
403 wantErr error
404 wantProtocolCode uint16
405 }{
406 {
407 name: "eof_before_welcome",
408 script: func(stream *sessionTestStream, _ *workerTestTerminal, _ *sessionTestAttachments) { _ = stream.Close() },
409 wantKind: AttachmentEventEnded,
410 },
411 {
412 name: "refusal_before_welcome",
413 script: func(stream *sessionTestStream, _ *workerTestTerminal, _ *sessionTestAttachments) {
414 stream.deliver(protocol.ErrorMsg{Code: protocol.ErrNoSuchSession, Text: "no such session"})
415 },
416 wantKind: AttachmentEventFailed,
417 wantProtocolCode: protocol.ErrNoSuchSession,
418 },
419 {
420 name: "unexpected_before_welcome",
421 script: func(stream *sessionTestStream, _ *workerTestTerminal, _ *sessionTestAttachments) {
422 stream.deliver(protocol.Sessions{})
423 },
424 wantKind: AttachmentEventFailed,
425 },
426 {
427 name: "eof_after_welcome_before_publication",
428 script: func(stream *sessionTestStream, _ *workerTestTerminal, _ *sessionTestAttachments) {
429 stream.deliver(protocol.Welcome{SessionName: "alpha"})
430 _ = stream.Close()
431 },
432 wantKind: AttachmentEventEnded,
433 },
434 {
435 name: "invalid_initial_publication",
436 script: func(stream *sessionTestStream, _ *workerTestTerminal, _ *sessionTestAttachments) {
437 stream.deliver(protocol.Welcome{SessionName: "alpha"})
438 // A full frame with no context never commits.
439 stream.deliver(protocol.Output{Epoch: 1, New: 1, Full: true})
440 },
441 wantKind: AttachmentEventFailed,
442 wantErr: errAttachmentPublication,
443 },
444 {
445 name: "publication_failure_before_attachment",
446 script: func(stream *sessionTestStream, terminal *workerTestTerminal, _ *sessionTestAttachments) {
447 terminal.publishErr = publishErr
448 stream.deliver(protocol.Welcome{SessionName: "alpha"})
449 stream.deliver(sessionTestOutput(1, "\x1b[Hready"))
450 },
451 wantKind: AttachmentEventFailed,
452 wantErr: publishErr,
453 },
454 {
455 name: "unavailable_observation_commits_attachment",
456 script: func(stream *sessionTestStream, terminal *workerTestTerminal, attached *sessionTestAttachments) {
457 terminal.publishErr = ports.ErrUIUnavailable
458 stream.deliver(protocol.Welcome{SessionName: "alpha"})
459 stream.deliver(sessionTestOutput(1, "\x1b[Hready"))
460 attached.token(t)
461 _ = stream.Close()
462 },
463 wantKind: AttachmentEventEnded,
464 wantAttached: true,
465 },
466 {
467 name: "eof_after_attachment",
468 script: func(stream *sessionTestStream, _ *workerTestTerminal, attached *sessionTestAttachments) {
469 stream.deliver(protocol.Welcome{SessionName: "alpha"})
470 stream.deliver(sessionTestOutput(1, "\x1b[Hready"))
471 attached.token(t)
472 _ = stream.Close()
473 },
474 wantKind: AttachmentEventEnded,
475 wantAttached: true,
476 },
477 {
478 name: "loss_after_attachment",
479 script: func(stream *sessionTestStream, _ *workerTestTerminal, attached *sessionTestAttachments) {
480 stream.deliver(protocol.Welcome{SessionName: "alpha"})
481 stream.deliver(sessionTestOutput(1, "\x1b[Hready"))
482 attached.token(t)
483 stream.fail(lossErr)
484 },
485 wantKind: AttachmentEventLost,
486 wantAttached: true,
487 wantErr: lossErr,
488 },
489 }
490
491 for _, tt := range tests {
492 t.Run(tt.name, func(t *testing.T) {
493 stream := newSessionTestStream()
494 terminal := newWorkerTestTerminal()
495 attached := newSessionTestAttachments()
496 events := sessionTestRun(t, stream, terminal, sessionTestWorkerConfig(sessionTestRequest(true)), attached)
497 awaitHello(t, stream)
498
499 tt.script(stream, terminal, attached)
500
501 var event AttachmentEvent
502 select {
503 case event = <-events:
504 case <-time.After(5 * time.Second):
505 t.Fatal("the worker never settled")
506 }
507 require.Equal(t, AttachmentToken{Generation: 4, Attempt: 1}, event.Token)
508 require.Equal(t, tt.wantKind, event.Kind, "event error: %v", event.Err)
509 if tt.wantErr != nil {
510 require.ErrorIs(t, event.Err, tt.wantErr)
511 }
512 if tt.wantProtocolCode != 0 {
513 var protocolErr *ProtocolError
514 require.ErrorAs(t, event.Err, &protocolErr)
515 require.Equal(t, tt.wantProtocolCode, protocolErr.Code)
516 }
517 require.Equal(t, tt.wantAttached, attached.attached())
518 if !tt.wantAttached {
519 // A frame that never committed was never written or flushed; the
520 // UI transaction still observed its own boundary.
521 require.Empty(t, terminal.written(), "an uncommitted frame must not be written")
522 require.Zero(t, terminal.flushCount(), "an uncommitted frame must not be flushed")
523 }
524 })
525 }
526}
527
528func TestSessionAttachmentWorkerHelloEnvironment(t *testing.T) {
529 tests := []struct {
530 name string
531 env AttachmentEnvironment
532 want terminalcap.ColorCapabilities
533 }{
534 {name: "unset defaults to truecolor", want: terminalcap.ColorCapabilities{Mode: terminalcap.TrueColor}},
535 {name: "supplied truecolor", env: AttachmentEnvironment{TermEnv: "xterm-256color", Cwd: "/workspace", Color: terminalcap.ColorCapabilities{Mode: terminalcap.TrueColor, Source: terminalcap.SourceDeclared}}, want: terminalcap.ColorCapabilities{Mode: terminalcap.TrueColor, Source: terminalcap.SourceDeclared}},
536 {name: "supplied ansi256", env: AttachmentEnvironment{TermEnv: "xterm-256color", Cwd: "/workspace", Color: terminalcap.ColorCapabilities{Mode: terminalcap.ANSI256}}, want: terminalcap.ColorCapabilities{Mode: terminalcap.ANSI256}},
537 {name: "supplied forced ansi16", env: AttachmentEnvironment{TermEnv: "xterm-256color", Cwd: "/workspace", Color: terminalcap.ColorCapabilities{Mode: terminalcap.ANSI16, Source: terminalcap.SourceForced}}, want: terminalcap.ColorCapabilities{Mode: terminalcap.ANSI16, Source: terminalcap.SourceForced}},
538 }
539 for _, tt := range tests {
540 t.Run(tt.name, func(t *testing.T) {
541 stream := newSessionTestStream()
542 terminal := newWorkerTestTerminal()
543 attached := newSessionTestAttachments()
544 supervisor := &Supervisor{cfg: SupervisorConfig{Terminal: terminal, AttachmentEnvironment: tt.env}, clientID: [16]byte{1}}
545 request := sessionTestRequest(true)
546 worker, err := supervisor.newAttachmentWorker(request, attachmentTab{}, nil, SessionEnvironment{Provenance: SessionEnvironmentLocalCLI, Cwd: tt.env.Cwd, Env: append([]string(nil), request.Env...)})
547 require.NoError(t, err)
548 events := make(chan AttachmentEvent, 1)
549 host := newWorkerTestHost(terminal, nil, attached.record)
550 go func() {
551 event, _ := host.Run(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, worker, stream)
552 events <- event
553 }()
554
555 hello := awaitHello(t, stream)
556 require.Equal(t, tt.env.TermEnv, hello.TermEnv)
557 require.Equal(t, tt.env.Cwd, hello.Cwd)
558 require.Equal(t, tt.want, hello.Color)
559 require.NoError(t, stream.Close())
560 <-events
561 })
562 }
563}
564
565func TestSessionAttachmentWorkerHandlesProtocolErrorAfterWelcome(t *testing.T) {
566 tests := []struct {
567 name string
568 attached bool
569 }{
570 {name: "awaiting initial publication"},
571 {name: "attached pump", attached: true},
572 }
573 for _, tt := range tests {
574 t.Run(tt.name, func(t *testing.T) {
575 stream := newSessionTestStream()
576 terminal := newWorkerTestTerminal()
577 attachments := newSessionTestAttachments()
578 events := sessionTestRun(t, stream, terminal, sessionTestWorkerConfig(sessionTestRequest(true)), attachments)
579 awaitHello(t, stream)
580 stream.deliver(protocol.Welcome{SessionName: "alpha"})
581 if tt.attached {
582 stream.deliver(sessionTestOutput(1, "initial"))
583 require.Eventually(t, attachments.attached, time.Second, time.Millisecond)
584 }
585 stream.deliver(protocol.ErrorMsg{Code: 42, Text: "refused"})
586 event := <-events
587 var protocolErr *ProtocolError
588 require.ErrorAs(t, event.Err, &protocolErr)
589 require.Equal(t, uint16(42), protocolErr.Code)
590 require.Equal(t, "refused", protocolErr.Text)
591 })
592 }
593}
594
595func TestSessionAttachmentWorkerValidatesIdentityByAdmission(t *testing.T) {
596 namedRequest := sessionTestRequest(true)
597 namedRequest.Admission, namedRequest.Name, namedRequest.Target = ports.BrokerAdmissionCreateNamed, "alpha", protocol.ExactSessionTarget{}
598 ephemeralRequest := sessionTestRequest(true)
599 ephemeralRequest.Admission, ephemeralRequest.Target = ports.BrokerAdmissionCreateEphemeral, protocol.ExactSessionTarget{}
600 tests := []struct {
601 name string
602 request ports.BrokerOpenStreamRequest
603 welcome protocol.Welcome
604 output protocol.Output
605 wantErr bool
606 }{
607 {name: "exact welcome mismatch", request: sessionTestRequest(true), welcome: protocol.Welcome{CommittedIdentity: &protocol.CommittedRouteIdentity{Target: protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{2}, SessionName: "alpha"}}}, wantErr: true},
608 {name: "exact publication mismatch", request: sessionTestRequest(true), welcome: protocol.Welcome{SessionName: "alpha"}, output: func() protocol.Output {
609 o := sessionTestOutput(1, "wrong")
610 o.Context.Route.Target.SessionName = "beta"
611 return o
612 }(), wantErr: true},
613 {name: "exact match", request: sessionTestRequest(true), welcome: protocol.Welcome{CommittedIdentity: &protocol.CommittedRouteIdentity{Target: sessionTestRequest(true).Target}}, output: sessionTestOutput(1, "ok")},
614 {name: "create named welcome mismatch", request: namedRequest, welcome: protocol.Welcome{SessionName: "beta"}, wantErr: true},
615 {name: "create named publication mismatch", request: namedRequest, welcome: protocol.Welcome{SessionName: "alpha"}, output: func() protocol.Output {
616 o := sessionTestOutput(1, "wrong")
617 o.Context.Route.Target.SessionName = "beta"
618 return o
619 }(), wantErr: true},
620 {name: "create named match", request: namedRequest, welcome: protocol.Welcome{SessionName: "alpha"}, output: sessionTestOutput(1, "ok")},
621 {name: "create ephemeral accepts committed identity", request: ephemeralRequest, welcome: protocol.Welcome{CommittedIdentity: &protocol.CommittedRouteIdentity{Target: sessionTestOutput(1, "").Context.Route.Target}}, output: sessionTestOutput(1, "ok")},
622 }
623 for _, tt := range tests {
624 t.Run(tt.name, func(t *testing.T) {
625 stream := newSessionTestStream()
626 terminal := newWorkerTestTerminal()
627 attached := newSessionTestAttachments()
628 events := sessionTestRun(t, stream, terminal, sessionTestWorkerConfig(tt.request), attached)
629 awaitHello(t, stream)
630 stream.deliver(tt.welcome)
631 if tt.output.Context != nil {
632 stream.deliver(tt.output)
633 }
634 if tt.wantErr {
635 event := <-events
636 var identityErr *AttachmentIdentityError
637 require.ErrorAs(t, event.Err, &identityErr)
638 require.False(t, attached.attached())
639 require.Empty(t, terminal.written())
640 return
641 }
642 require.Eventually(t, attached.attached, time.Second, time.Millisecond)
643 require.Equal(t, "ok", terminal.written())
644 require.NoError(t, stream.Close())
645 <-events
646 })
647 }
648}
649
650// TestSessionAttachmentWorkerRemoteHello proves the same worker derives a
651// remote Hello from a remote request: the typed-preamble rules stay per
652// logical stream, and no endpoint is rewritten.
653func TestSessionAttachmentWorkerRemoteHello(t *testing.T) {
654 stream := newSessionTestStream()
655 terminal := newWorkerTestTerminal()
656 attached := newSessionTestAttachments()
657 request := sessionTestRequest(false)
658 events := sessionTestRun(t, stream, terminal, sessionTestWorkerConfig(request), attached)
659
660 hello := awaitHello(t, stream)
661 require.True(t, hello.Remote)
662 require.Equal(t, protocol.EnvironmentPolicyClientOwned, hello.EnvironmentPolicy)
663 require.Equal(t, protocol.IntentAttach, hello.Intent)
664 require.NotNil(t, hello.ExactTarget)
665 require.Equal(t, "alpha", hello.ExactTarget.SessionName)
666
667 require.NoError(t, stream.Close())
668 select {
669 case event := <-events:
670 require.Equal(t, AttachmentEventEnded, event.Kind)
671 case <-time.After(5 * time.Second):
672 t.Fatal("the worker never settled")
673 }
674}
675
676// TestSessionAttachmentWorkerNeverReadsTerminalInput proves the worker has no
677// path to terminal input: it consumes no bytes and sends no Input frame while
678// the foreground owns no input source.
679func TestSessionAttachmentWorkerNeverReadsTerminalInput(t *testing.T) {
680 stream := newSessionTestStream()
681 terminal := newWorkerTestTerminal()
682 attached := newSessionTestAttachments()
683 events := sessionTestRun(t, stream, terminal, sessionTestWorkerConfig(sessionTestRequest(true)), attached)
684
685 awaitHello(t, stream)
686 stream.deliver(protocol.Welcome{SessionName: "alpha"})
687 stream.deliver(sessionTestOutput(1, "\x1b[Hready"))
688 _ = attached.token(t)
689
690 for _, message := range stream.messages() {
691 _, isInput := message.(protocol.Input)
692 require.False(t, isInput, "the worker must never send terminal input in P5.3b")
693 }
694 require.Equal(t, "\x1b[Hready", terminal.written(), "only session output reaches the terminal")
695
696 require.NoError(t, stream.Close())
697 select {
698 case <-events:
699 case <-time.After(5 * time.Second):
700 t.Fatal("the worker never settled")
701 }
702}