1package broker
2
3import (
4 "context"
5 "errors"
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/stretchr/testify/mock"
12 "github.com/stretchr/testify/require"
13)
14
15// routeHosts is a mutable committed-membership view, so a test can change the
16// record between resolution and dial.
17type routeHosts struct {
18 record ports.BrokerHostRecord
19 found bool
20}
21
22func (h *routeHosts) LookupHost(ctx context.Context, endpoint string) (ports.BrokerHostRecord, bool, error) {
23 return h.record, h.found && endpoint == h.record.Registration.Endpoint, ctx.Err()
24}
25
26// testRouteAddress derives a stable address from the route contents.
27func testRouteAddress(spec ports.BrokerRouteSpec) (string, error) {
28 if err := spec.Validate(); err != nil {
29 return "", err
30 }
31 return "addr:" + string(spec.Kind) + ":" + spec.Path + spec.Target, nil
32}
33
34func testRouteRecord(t *testing.T) ports.BrokerHostRecord {
35 t.Helper()
36 registration := domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1}
37 policy := observationPolicy()
38 spec, err := ports.BrokerRouteForTransport(policy.Transport, registration.Endpoint)
39 require.NoError(t, err)
40 return ports.BrokerHostRecord{Registration: registration, Policy: policy, Route: spec, Identity: "daemon"}
41}
42
43func testRouteRequest(record ports.BrokerHostRecord) ports.BrokerOpenStreamRequest {
44 return ports.BrokerOpenStreamRequest{Endpoint: record.Registration.Endpoint, Registration: record.Registration, Policy: record.Policy, StartMode: ports.BrokerDaemonExistingOnly}
45}
46
47func TestRoutesResolveRemoteTarget(t *testing.T) {
48 record := testRouteRecord(t)
49 tests := []struct {
50 name string
51 change func(*routeHosts, *ports.BrokerOpenStreamRequest)
52 wantCode ports.BrokerErrorCode
53 }{
54 {name: "committed record resolves"},
55 {name: "unknown host", change: func(h *routeHosts, _ *ports.BrokerOpenStreamRequest) { h.found = false }, wantCode: ports.BrokerErrorUnavailable},
56 {name: "stale registration", change: func(_ *routeHosts, r *ports.BrokerOpenStreamRequest) { r.Registration.Generation++ }, wantCode: ports.BrokerErrorUnavailable},
57 {name: "conflicting policy", change: func(_ *routeHosts, r *ports.BrokerOpenStreamRequest) { r.Policy.Trust = "other" }, wantCode: ports.BrokerErrorConflictingPolicy},
58 {name: "invalid durable route", change: func(h *routeHosts, _ *ports.BrokerOpenStreamRequest) { h.record.Route = ports.BrokerRouteSpec{} }, wantCode: ports.BrokerErrorIncompatible},
59 }
60 for _, tt := range tests {
61 t.Run(tt.name, func(t *testing.T) {
62 hosts := &routeHosts{record: record, found: true}
63 request := testRouteRequest(record)
64 if tt.change != nil {
65 tt.change(hosts, &request)
66 }
67 routes := &Routes{Hosts: hosts, Address: testRouteAddress}
68 target, err := routes.ResolveDialTarget(context.Background(), request)
69 if tt.wantCode != 0 {
70 var brokerErr ports.BrokerError
71 require.True(t, errors.As(err, &brokerErr), "error %v", err)
72 require.Equal(t, tt.wantCode, brokerErr.Code)
73 return
74 }
75 require.NoError(t, err)
76 address, _ := testRouteAddress(record.Route)
77 require.Equal(t, address, target.Address)
78 require.Equal(t, record.Registration, target.Fence.Registration)
79 require.Equal(t, ports.BrokerExpectedIdentity{Identity: "daemon", Bound: true}, target.ExpectedIdentity)
80 require.Equal(t, request.StartMode, target.StartMode)
81 })
82 }
83}
84
85// A resolved remote target is re-authorized against committed membership
86// immediately before dialing: any change since resolution refuses the dial.
87func TestRoutesAuthorizeRemoteDialRejectsChangedAuthority(t *testing.T) {
88 record := testRouteRecord(t)
89 tests := []struct {
90 name string
91 change func(*routeHosts)
92 wantErr bool
93 }{
94 {name: "unchanged", change: func(*routeHosts) {}},
95 {name: "removed", change: func(h *routeHosts) { h.found = false }, wantErr: true},
96 {name: "readded", change: func(h *routeHosts) { h.record.Registration.Incarnation[0]++ }, wantErr: true},
97 {name: "policy", change: func(h *routeHosts) { h.record.Policy.Trust = "other" }, wantErr: true},
98 {name: "route", change: func(h *routeHosts) { h.record.Route.Target = "other@example.test" }, wantErr: true},
99 {name: "identity", change: func(h *routeHosts) { h.record.Identity = "other-daemon" }, wantErr: true},
100 }
101 for _, tt := range tests {
102 t.Run(tt.name, func(t *testing.T) {
103 hosts := &routeHosts{record: record, found: true}
104 routes := &Routes{Hosts: hosts, Address: testRouteAddress}
105 target, err := routes.ResolveDialTarget(context.Background(), testRouteRequest(record))
106 require.NoError(t, err)
107 tt.change(hosts)
108 spec, err := routes.AuthorizeRemoteDial(context.Background(), target)
109 if tt.wantErr {
110 require.Error(t, err)
111 return
112 }
113 require.NoError(t, err)
114 require.Equal(t, record.Route, spec)
115 })
116 }
117}
118
119func TestRoutesLocalDelegatesAndFailsClosed(t *testing.T) {
120 request := ports.BrokerOpenStreamRequest{Local: true, Policy: observationPolicy(), StartMode: ports.BrokerDaemonExistingOnly}
121 want := ports.BrokerDialTarget{Fence: ports.BrokerEndpointFence{Local: true}, Address: "local"}
122 local := portsmocks.NewMockBrokerRouteAuthority(t)
123 local.EXPECT().ResolveDialTarget(mock.Anything, request).Return(want, nil)
124
125 got, err := (&Routes{Local: local}).ResolveDialTarget(context.Background(), request)
126 require.NoError(t, err)
127 require.Equal(t, want, got)
128
129 _, err = (&Routes{}).ResolveDialTarget(context.Background(), request)
130 require.Error(t, err, "no local binding authorizes no local dial")
131 _, err = (&Routes{}).ResolveDialTarget(context.Background(), testRouteRequest(testRouteRecord(t)))
132 require.Error(t, err, "no committed membership authorizes no remote dial")
133 _, err = (&Routes{Hosts: &routeHosts{}, Address: testRouteAddress}).AuthorizeRemoteDial(context.Background(), want)
134 require.Error(t, err, "a local target never selects a remote route")
135
136 cancelled, cancel := context.WithCancel(context.Background())
137 cancel()
138 _, err = (&Routes{}).ResolveDialTarget(cancelled, request)
139 require.ErrorIs(t, err, context.Canceled)
140}