1package brokeripc
2
3import (
4 "context"
5 "testing"
6 "time"
7
8 "github.com/stretchr/testify/require"
9
10 "github.com/bnema/vev/internal/ports"
11)
12
13// awaitDone waits for one broker service to become terminal.
14func awaitDone(t *testing.T, service ports.BrokerService) {
15 t.Helper()
16 select {
17 case <-service.Done():
18 case <-time.After(5 * time.Second):
19 t.Fatal("broker service did not become terminal")
20 }
21}
22
23// TestConnectorConnectReturnsIndependentService proves the ports.BrokerConnector
24// implementation dials one connection per Connect and hands ownership to the
25// caller.
26func TestConnectorConnectReturnsIndependentService(t *testing.T) {
27 e := startEndpoint(t, Config{})
28 connector := NewConnector(e.path, Config{})
29 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
30 defer cancel()
31
32 first, err := connector.Connect(ctx)
33 require.NoError(t, err)
34 require.NotNil(t, first)
35
36 second, err := connector.Connect(ctx)
37 require.NoError(t, err)
38 require.NotNil(t, second)
39 require.NotEqual(t, first.ConnectionID(), second.ConnectionID(), "each Connect admits an independent connection")
40
41 // An orderly local Close reports no terminal cause and is stable.
42 require.NoError(t, first.Close())
43 awaitDone(t, first)
44 require.NoError(t, first.Err())
45 require.NoError(t, first.Err(), "Err is stable after Done")
46 require.False(t, func() bool {
47 select {
48 case <-second.Done():
49 return true
50 default:
51 return false
52 }
53 }(), "closing one connection must not close another")
54
55 require.NoError(t, second.Close())
56 awaitDone(t, second)
57}
58
59// TestConnectorReportsCarriageLoss proves the client service surfaces a broker
60// loss through Done and Err while it holds no logical stream.
61func TestConnectorReportsCarriageLoss(t *testing.T) {
62 e := startEndpoint(t, Config{})
63 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
64 defer cancel()
65
66 service, err := NewConnector(e.path, Config{}).Connect(ctx)
67 require.NoError(t, err)
68
69 // The broker side is torn down: the client observes a non-nil terminal
70 // cause instead of a silent close.
71 require.NoError(t, e.accept().Close())
72 awaitDone(t, service)
73 require.Error(t, service.Err())
74
75 require.NoError(t, service.Close())
76 require.Error(t, service.Err(), "the first terminal cause is never overwritten by a local Close")
77}
78
79func TestConnectorRefusesInvalidConfiguration(t *testing.T) {
80 var absent *Connector
81 _, err := absent.Connect(context.Background())
82 require.ErrorIs(t, err, ErrConfig)
83
84 _, err = NewConnector("", Config{}).Connect(context.Background())
85 require.ErrorIs(t, err, ErrConfig)
86}
87
88// TestConnectorCancellationInterruptsStalledRegistration proves the
89// ports.BrokerConnector seam honors cancellation while a peer stalls in the
90// registration exchange: Connect returns the caller's cancellation instead of
91// waiting out the handshake budget. The supervisor's cancellation lifetime
92// depends on exactly this contract.
93func TestConnectorCancellationInterruptsStalledRegistration(t *testing.T) {
94 path := testSocketPath(t)
95 peer := startSilentBrokerPeer(t, path)
96
97 connector := NewConnector(path, Config{HandshakeTimeout: 30 * time.Second})
98 ctx, cancel := context.WithCancel(context.Background())
99 defer cancel()
100 result := make(chan error, 1)
101 go func() {
102 _, err := connector.Connect(ctx)
103 result <- err
104 }()
105 peer.awaitPreamble(t)
106 cancel()
107
108 select {
109 case err := <-result:
110 require.ErrorIs(t, err, context.Canceled, "Connect must return the caller's cancellation, not wait out the handshake budget")
111 case <-time.After(5 * time.Second):
112 t.Fatal("Connect must return on cancellation instead of waiting out HandshakeTimeout")
113 }
114 peer.stop()
115}
116
117// TestConnectorServiceOutlivesSetupContext pins the ports.BrokerConnector
118// contract that a successful service is independent of the setup context:
119// cancelling that context after Connect returns must not close or disturb the
120// established connection.
121func TestConnectorServiceOutlivesSetupContext(t *testing.T) {
122 e := startEndpoint(t, Config{})
123 ctx, cancel := context.WithCancel(context.Background())
124 service, err := NewConnector(e.path, Config{}).Connect(ctx)
125 require.NoError(t, err)
126 cancel()
127
128 select {
129 case <-service.Done():
130 t.Fatal("cancelling the setup context must not close the established service")
131 case <-time.After(100 * time.Millisecond):
132 }
133
134 // The detached connection still serves its own scope and can be closed by
135 // its owner, which reports an orderly end.
136 require.NotEqual(t, ports.BrokerConnectionID{}, service.ConnectionID())
137 require.NoError(t, service.Close())
138 awaitDone(t, service)
139 require.NoError(t, service.Err())
140}