outbox_drain_test.go RawHistoryFolder

1package client
2
3import (
4	"context"
5	"errors"
6	"testing"
7	"time"
8
9	"github.com/bnema/vev/internal/protocol"
10	"github.com/stretchr/testify/require"
11)
12
13func TestOutboxDrain(t *testing.T) {
14	linkErr := errors.New("link lost")
15	for _, mode := range []string{"written", "timeout", "cancelled", "failed", "revoked"} {
16		t.Run(mode, func(t *testing.T) {
17			stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
18			box := newAttachmentOutbox(stream)
19			defer box.join()
20			require.NoError(t, box.SendClient(protocol.Detach{}))
21			clock := newSupervisorTestClock()
22			worker := &sessionAttachmentWorker{cfg: sessionAttachmentConfig{Clock: clock}}
23			host := newAttachmentHost(attachmentHostConfig{})
24			fg := host.newForeground(AttachmentToken{Generation: 1}, stream)
25			ctx, cancel := context.WithCancel(t.Context())
26			defer cancel()
27			result := make(chan error, 1)
28			go func() { result <- worker.drainOutbox(ctx, fg, box) }()
29			timer := clock.awaitTimer(t)
30			require.Equal(t, 2*time.Second, timer.delay)
31			switch mode {
32			case "written":
33				close(stream.release)
34			case "timeout":
35				timer.fire()
36			case "cancelled":
37				cancel()
38			case "failed":
39				stream.fail = linkErr
40				close(stream.release)
41			case "revoked":
42				fg.finalize()
43			}
44			select {
45			case err := <-result:
46				switch mode {
47				case "cancelled":
48					require.ErrorIs(t, err, context.Canceled)
49				case "failed":
50					require.ErrorIs(t, err, linkErr)
51				case "revoked":
52					require.ErrorIs(t, err, errAttachmentForegroundRevoked)
53				default:
54					require.NoError(t, err)
55				}
56			case <-time.After(time.Second):
57				t.Fatal("drain did not settle")
58			}
59			if mode == "written" {
60				require.Len(t, stream.messages(), 1)
61			}
62		})
63	}
64}