1package client
2
3import (
4 "context"
5 "testing"
6 "time"
7
8 "github.com/bnema/vev/internal/protocol"
9 "github.com/stretchr/testify/require"
10)
11
12func TestOutboxFullCancellation(t *testing.T) {
13 for _, foreground := range []bool{false, true} {
14 t.Run(map[bool]string{false: "context", true: "foreground"}[foreground], func(t *testing.T) {
15 stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
16 box := newAttachmentOutbox(stream)
17 defer box.join()
18 for range outboxMaxPending {
19 require.NoError(t, box.SendClient(protocol.Ack{}))
20 }
21 ctx, cancel := context.WithCancel(t.Context())
22 defer cancel()
23 revoked := make(chan struct{})
24 result := make(chan error, 1)
25 go func() { result <- box.enqueue(ctx, revoked, protocol.Ack{}) }()
26 if foreground {
27 close(revoked)
28 } else {
29 cancel()
30 }
31 select {
32 case err := <-result:
33 if foreground {
34 require.ErrorIs(t, err, errAttachmentForegroundRevoked)
35 } else {
36 require.ErrorIs(t, err, context.Canceled)
37 }
38 case <-time.After(time.Second):
39 t.Fatal("full outbox ignored cancellation")
40 }
41 })
42 }
43}
44
45// TestOutboxFullAdmitsAfterProgress checks that a caller blocked on a full
46// outbox is admitted once the sender writes and frees space.
47func TestOutboxFullAdmitsAfterProgress(t *testing.T) {
48 stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
49 box := newAttachmentOutbox(stream)
50 defer box.join()
51 for range outboxMaxPending {
52 require.NoError(t, box.SendClient(protocol.Ack{}))
53 }
54 result := make(chan error, 1)
55 go func() { result <- box.enqueue(t.Context(), nil, protocol.Detach{}) }()
56 select {
57 case err := <-result:
58 t.Fatalf("full outbox admitted before progress: %v", err)
59 case <-time.After(50 * time.Millisecond):
60 }
61 close(stream.release)
62 select {
63 case err := <-result:
64 require.NoError(t, err)
65 case <-time.After(time.Second):
66 t.Fatal("blocked caller was not admitted after progress")
67 }
68 require.Eventually(t, func() bool { return len(stream.messages()) == outboxMaxPending+1 }, time.Second, time.Millisecond)
69}
70
71func TestOutboxMixedUnsentInput(t *testing.T) {
72 stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
73 box := newAttachmentOutbox(stream)
74 defer box.join()
75 for _, message := range []protocol.ClientMessage{protocol.Ack{}, protocol.Input{Data: []byte("a")}, protocol.TerminalFocus{}, protocol.Input{Data: []byte("b")}} {
76 require.NoError(t, box.SendClient(message))
77 }
78 require.Equal(t, []byte("ab"), box.unsentInput(0, box.accepted()))
79 require.Equal(t, []byte("b"), box.unsentInput(2, box.accepted()))
80}