1package broker
2
3import (
4 "context"
5 "encoding/json"
6 "testing"
7
8 "github.com/bnema/vev/internal/domain"
9 "github.com/bnema/vev/internal/ports"
10 portsmocks "github.com/bnema/vev/internal/ports/mocks"
11 "github.com/bnema/vev/internal/protocol"
12 "github.com/bnema/vev/internal/protocol/catalogue"
13 "github.com/stretchr/testify/mock"
14 "github.com/stretchr/testify/require"
15)
16
17func observationPolicy() ports.BrokerPolicy {
18 return ports.BrokerPolicy{
19 ProtocolVersion: protocol.Version, CatalogSchemaVersion: catalogue.RemoteCatalogSchemaVersion,
20 Transport: "ssh-quic", Trust: "ssh", Launch: "explicit", Isolation: "user",
21 EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned,
22 }
23}
24
25func observationCatalogue(t *testing.T) string {
26 t.Helper()
27 encoded, err := json.Marshal(catalogue.RemoteCatalog{ProtocolVersion: protocol.Version, SchemaVersion: catalogue.RemoteCatalogSchemaVersion, Sessions: []catalogue.RemoteCatalogSession{}})
28 require.NoError(t, err)
29 return string(encoded) + "\n"
30}
31
32// observationHosts answers LookupHost for exactly one configured record.
33func observationHosts(t *testing.T, record ports.BrokerHostRecord) ports.BrokerHostAuthorityReader {
34 hosts := portsmocks.NewMockBrokerHostAuthorityReader(t)
35 hosts.EXPECT().LookupHost(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, endpoint string) (ports.BrokerHostRecord, bool, error) {
36 return record, endpoint == record.Registration.Endpoint, ctx.Err()
37 }).Maybe()
38 return hosts
39}
40
41func TestObservationRequestIsObservationOnly(t *testing.T) {
42 registration := domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1}
43 policy := observationPolicy()
44 tests := []struct {
45 name string
46 build func(t *testing.T) (ports.BrokerOpenStreamRequest, error)
47 check func(t *testing.T, request ports.BrokerOpenStreamRequest)
48 }{
49 {
50 name: "remote",
51 build: func(t *testing.T) (ports.BrokerOpenStreamRequest, error) {
52 probe := &RemoteProbe{Epoch: 7, Hosts: observationHosts(t, ports.BrokerHostRecord{Registration: registration, Policy: policy})}
53 return probe.request(context.Background(), registration)
54 },
55 check: func(t *testing.T, request ports.BrokerOpenStreamRequest) {
56 require.False(t, request.Local)
57 require.Equal(t, registration, request.Registration)
58 require.Equal(t, registration.Endpoint, request.Endpoint)
59 },
60 },
61 {
62 name: "local",
63 build: func(t *testing.T) (ports.BrokerOpenStreamRequest, error) {
64 probe, err := NewLocalDaemonProbe(7, "local-address", policy, portsmocks.NewMockBrokerEndpointConnector(t), portsmocks.NewMockSessionCodec(t), func() (ports.BrokerDaemonIdentity, error) { return "daemon", nil })
65 require.NoError(t, err)
66 return probe.request()
67 },
68 check: func(t *testing.T, request ports.BrokerOpenStreamRequest) {
69 require.True(t, request.Local)
70 require.Empty(t, request.Endpoint)
71 require.Equal(t, domain.RemoteRegistration{}, request.Registration)
72 },
73 },
74 }
75 for _, tt := range tests {
76 t.Run(tt.name, func(t *testing.T) {
77 request, err := tt.build(t)
78 require.NoError(t, err)
79 require.NoError(t, request.Validate())
80 require.Equal(t, ports.BrokerEpoch(7), request.Epoch)
81 require.Equal(t, ports.BrokerStreamObservation, request.Purpose)
82 require.Zero(t, request.Admission)
83 require.Empty(t, request.Name)
84 require.Empty(t, request.Env)
85 require.Equal(t, protocol.ExactSessionTarget{}, request.Target)
86 require.Equal(t, policy, request.Policy)
87 require.NotZero(t, request.Connection)
88 require.NotZero(t, request.Stream)
89 tt.check(t, request)
90 // Observation must never start the target: a spawn-capable request
91 // is refused rather than silently narrowed.
92 require.Equal(t, ports.BrokerDaemonExistingOnly, request.StartMode)
93 startable := request
94 startable.StartMode = ports.BrokerDaemonStartIfNeeded
95 require.Error(t, startable.Validate())
96 })
97 }
98}
99
100func TestRemoteProbeRejectsStaleRegistrationBeforeDial(t *testing.T) {
101 registration := domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1}
102 tests := []struct {
103 name string
104 probe *RemoteProbe
105 wantErr string
106 }{
107 {name: "incomplete composition", probe: &RemoteProbe{Epoch: 7}, wantErr: "incomplete broker remote probe"},
108 {name: "unknown registration", probe: &RemoteProbe{
109 Epoch: 7, Routes: portsmocks.NewMockBrokerRouteAuthority(t), Binder: portsmocks.NewMockBrokerIdentityBinder(t),
110 Connector: portsmocks.NewMockBrokerEndpointConnector(t), Hosts: observationHosts(t, ports.BrokerHostRecord{}),
111 }, wantErr: "unknown or stale probe registration"},
112 }
113 for _, tt := range tests {
114 t.Run(tt.name, func(t *testing.T) {
115 _, err := tt.probe.Probe(context.Background(), registration)
116 require.ErrorContains(t, err, tt.wantErr)
117 })
118 }
119}
120
121func TestNewLocalDaemonProbeRejectsIncompleteComposition(t *testing.T) {
122 policy := observationPolicy()
123 loader := func() (ports.BrokerDaemonIdentity, error) { return "daemon", nil }
124 connector := portsmocks.NewMockBrokerEndpointConnector(t)
125 codec := portsmocks.NewMockSessionCodec(t)
126 tests := []struct {
127 name string
128 epoch ports.BrokerEpoch
129 address string
130 connector ports.BrokerEndpointConnector
131 codec ports.SessionCodec
132 loader LocalIdentityLoader
133 wantErr string
134 }{
135 {name: "epoch", address: "a", connector: connector, codec: codec, loader: loader, wantErr: "epoch"},
136 {name: "connector", epoch: 1, address: "a", codec: codec, loader: loader, wantErr: "connector"},
137 {name: "codec", epoch: 1, address: "a", connector: connector, loader: loader, wantErr: "codec"},
138 {name: "loader", epoch: 1, address: "a", connector: connector, codec: codec, wantErr: "identity loader"},
139 {name: "address", epoch: 1, connector: connector, codec: codec, loader: loader, wantErr: "address"},
140 }
141 for _, tt := range tests {
142 t.Run(tt.name, func(t *testing.T) {
143 _, err := NewLocalDaemonProbe(tt.epoch, tt.address, policy, tt.connector, tt.codec, tt.loader)
144 require.ErrorContains(t, err, tt.wantErr)
145 })
146 }
147}
148
149func TestLocalDaemonProbeWithoutIdentityDoesNotDial(t *testing.T) {
150 // No Connect expectation: any dial fails the mock.
151 probe, err := NewLocalDaemonProbe(7, "local-address", observationPolicy(), portsmocks.NewMockBrokerEndpointConnector(t), portsmocks.NewMockSessionCodec(t), func() (ports.BrokerDaemonIdentity, error) { return "", nil })
152 require.NoError(t, err)
153 observation, err := probe.ProbeLocal(context.Background())
154 require.NoError(t, err)
155 require.Equal(t, domain.RemoteAvailabilityUnknown, observation.Availability)
156 require.Zero(t, observation.Identity)
157}
158
159func TestLocalDaemonProbeNilReceiverFailsClosed(t *testing.T) {
160 var probe *LocalDaemonProbe
161 observation, err := probe.ProbeLocal(context.Background())
162 require.Error(t, err)
163 require.Equal(t, domain.RemoteAvailabilityUnreachable, observation.Availability)
164}