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