1package ports
2
3import (
4 "reflect"
5 "strings"
6 "testing"
7 "time"
8 "unicode/utf8"
9
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/protocol"
12 "github.com/bnema/vev/internal/protocol/catalogue"
13 "github.com/stretchr/testify/require"
14)
15
16func testBrokerRegistration(endpoint string, incarnation byte, generation domain.RemoteGeneration) domain.RemoteRegistration {
17 return domain.RemoteRegistration{
18 Endpoint: endpoint,
19 Incarnation: [16]byte{incarnation, 7, 9},
20 Generation: generation,
21 }
22}
23
24func testBrokerLifecycle() domain.SessionLifecycleID {
25 return domain.SessionLifecycleID{1, 2, 3}
26}
27
28func testBrokerTarget() protocol.ExactSessionTarget {
29 return protocol.ExactSessionTarget{LifecycleID: testBrokerLifecycle(), SessionName: "work"}
30}
31
32func testBrokerPolicy() BrokerPolicy {
33 return BrokerPolicy{
34 ProtocolVersion: protocol.Version,
35 CatalogSchemaVersion: catalogue.RemoteCatalogSchemaVersion,
36 EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned,
37 Transport: "quic",
38 Trust: "trust", Launch: "launch", Isolation: "user",
39 }
40}
41
42func testBrokerObservation(endpoint string) BrokerDaemonObservation {
43 return BrokerDaemonObservation{
44 Endpoint: endpoint,
45 DisplayOrigin: endpoint,
46 Registration: testBrokerRegistration(endpoint, 1, 1),
47 Policy: testBrokerPolicy(),
48 Availability: domain.RemoteAvailabilityReachable,
49 }
50}
51
52func testLocalBrokerObservation() BrokerDaemonObservation {
53 return BrokerDaemonObservation{
54 Local: true,
55 DisplayOrigin: "local",
56 Policy: testBrokerPolicy(),
57 Availability: domain.RemoteAvailabilityReachable,
58 }
59}
60
61func validBrokerSnapshot() BrokerSnapshot {
62 return BrokerSnapshot{
63 Epoch: 3,
64 Revision: 9,
65 Daemons: []BrokerDaemonObservation{testBrokerObservation("user@arch")},
66 }
67}
68
69// TestBrokerHostStoreIsASnapshotStore pins the interface guard: the exclusive
70// owner of durable membership also owns the advisory snapshot, so a
71// BrokerHostStore must stay usable wherever a BrokerSnapshotStore is required.
72func TestBrokerHostStoreIsASnapshotStore(t *testing.T) {
73 host := reflect.TypeOf((*BrokerHostStore)(nil)).Elem()
74 snapshot := reflect.TypeOf((*BrokerSnapshotStore)(nil)).Elem()
75 if !host.Implements(snapshot) {
76 t.Fatal("BrokerHostStore must implement BrokerSnapshotStore")
77 }
78}
79
80// TestValidateDurableHostProjection matches the durable rules the offline store
81// enforces and the registry re-checks before adopting an observation.
82func TestValidateDurableHostProjection(t *testing.T) {
83 session := catalogue.RemoteCatalogSession{
84 LifecycleID: testBrokerLifecycle(),
85 Name: "work",
86 State: catalogue.RemoteCatalogSessionUp,
87 Tabs: []catalogue.RemoteCatalogTab{},
88 }
89 tests := []struct {
90 name string
91 mutate func(*BrokerDaemonObservation)
92 wantErr bool
93 }{
94 {name: "empty inventory", mutate: func(*BrokerDaemonObservation) {}},
95 {name: "valid inventory", mutate: func(o *BrokerDaemonObservation) {
96 o.InventoryKnown = true
97 o.Sessions = []catalogue.RemoteCatalogSession{session}
98 }},
99 {name: "local observation", mutate: func(o *BrokerDaemonObservation) {
100 *o = testLocalBrokerObservation()
101 }, wantErr: true},
102 {name: "zero availability", mutate: func(o *BrokerDaemonObservation) { o.Availability = 0 }, wantErr: true},
103 {name: "unknown availability", mutate: func(o *BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityUnknown }},
104 {name: "availability past the closed range", mutate: func(o *BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityNoDaemon + 1 }, wantErr: true},
105 {name: "failure kind past the closed range", mutate: func(o *BrokerDaemonObservation) { o.LastFailure.Kind = domain.RemoteFailureInvalidResponse + 1 }, wantErr: true},
106 {name: "missing lifecycle identity", mutate: func(o *BrokerDaemonObservation) {
107 o.InventoryKnown = true
108 o.Sessions = []catalogue.RemoteCatalogSession{{Name: "work", State: catalogue.RemoteCatalogSessionUp, Tabs: []catalogue.RemoteCatalogTab{}}}
109 }, wantErr: true},
110 {name: "absent tab list", mutate: func(o *BrokerDaemonObservation) {
111 broken := session
112 broken.Tabs = nil
113 o.InventoryKnown = true
114 o.Sessions = []catalogue.RemoteCatalogSession{broken}
115 }, wantErr: true},
116 {name: "sessions past the catalogue bound", mutate: func(o *BrokerDaemonObservation) {
117 o.InventoryKnown = true
118 o.Sessions = make([]catalogue.RemoteCatalogSession, catalogue.RemoteCatalogMaxSessions+1)
119 for i := range o.Sessions {
120 o.Sessions[i] = session
121 }
122 }, wantErr: true},
123 }
124 for _, tc := range tests {
125 t.Run(tc.name, func(t *testing.T) {
126 obs := testBrokerObservation("user@arch")
127 obs.LastSuccess = time.Unix(50, 0)
128 tc.mutate(&obs)
129 if gotErr := ValidateDurableHostProjection(obs); (gotErr != nil) != tc.wantErr {
130 t.Fatalf("ValidateDurableHostProjection() = %v, wantErr %t", gotErr, tc.wantErr)
131 }
132 })
133 }
134}
135
136func TestBrokerSnapshotValidateBounds(t *testing.T) {
137 tests := []struct {
138 name string
139 mutate func(*BrokerSnapshot)
140 wantErr bool
141 }{
142 {name: "valid", mutate: func(*BrokerSnapshot) {}, wantErr: false},
143 {name: "zero epoch", mutate: func(s *BrokerSnapshot) { s.Epoch = 0 }, wantErr: true},
144 {name: "zero revision", mutate: func(s *BrokerSnapshot) { s.Revision = 0 }, wantErr: true},
145 {name: "too many daemons", mutate: func(s *BrokerSnapshot) {
146 s.Daemons = make([]BrokerDaemonObservation, BrokerMaxDaemonsPerSnapshot+1)
147 for i := range s.Daemons {
148 s.Daemons[i] = testBrokerObservation("user@arch")
149 s.Daemons[i].Registration = testBrokerRegistration("user@arch", 1, 1)
150 }
151 }, wantErr: true},
152 {name: "local daemon plus remote hosts is valid", mutate: func(s *BrokerSnapshot) {
153 s.Daemons = append([]BrokerDaemonObservation{testLocalBrokerObservation()}, s.Daemons...)
154 }},
155 {name: "two local daemons", mutate: func(s *BrokerSnapshot) {
156 s.Daemons = append(s.Daemons, testLocalBrokerObservation(), testLocalBrokerObservation())
157 }, wantErr: true},
158 {name: "duplicate host", mutate: func(s *BrokerSnapshot) {
159 s.Daemons = append(s.Daemons, testBrokerObservation("user@arch"))
160 }, wantErr: true},
161 {name: "endpoint registration mismatch", mutate: func(s *BrokerSnapshot) {
162 s.Daemons[0].Registration = testBrokerRegistration("user@other", 1, 1)
163 }, wantErr: true},
164 {name: "zero registration incarnation", mutate: func(s *BrokerSnapshot) {
165 s.Daemons[0].Registration = domain.RemoteRegistration{Endpoint: "user@arch", Generation: 1}
166 }, wantErr: true},
167 {name: "live host as tombstone", mutate: func(s *BrokerSnapshot) {
168 s.Removed = []BrokerHostTombstone{{
169 Endpoint: "user@arch",
170 Registration: testBrokerRegistration("user@arch", 1, 1),
171 RetiredRevision: 9,
172 }}
173 }, wantErr: true},
174 {name: "too many tombstones", mutate: func(s *BrokerSnapshot) {
175 s.Removed = make([]BrokerHostTombstone, BrokerMaxTombstones+1)
176 for i := range s.Removed {
177 s.Removed[i] = BrokerHostTombstone{
178 Endpoint: "user@gone",
179 Registration: testBrokerRegistration("user@gone", 1, 1),
180 RetiredRevision: 9,
181 }
182 }
183 }, wantErr: true},
184 }
185 for _, tt := range tests {
186 t.Run(tt.name, func(t *testing.T) {
187 snapshot := validBrokerSnapshot()
188 tt.mutate(&snapshot)
189 if err := snapshot.Validate(); (err != nil) != tt.wantErr {
190 t.Fatalf("Validate() = %v, wantErr %t", err, tt.wantErr)
191 }
192 })
193 }
194}
195
196func TestBrokerSnapshotEpochRevisionOrdering(t *testing.T) {
197 tests := []struct {
198 name string
199 current BrokerSnapshot
200 candidate BrokerSnapshot
201 wantNewer bool
202 description string
203 }{
204 {
205 name: "higher revision wins in epoch",
206 current: BrokerSnapshot{Epoch: 3, Revision: 9},
207 candidate: BrokerSnapshot{Epoch: 3, Revision: 10},
208 wantNewer: true,
209 description: "monotonic revision within an epoch",
210 },
211 {
212 name: "lower revision loses in epoch",
213 current: BrokerSnapshot{Epoch: 3, Revision: 10},
214 candidate: BrokerSnapshot{Epoch: 3, Revision: 9},
215 wantNewer: false,
216 description: "stale revision never applies",
217 },
218 {
219 name: "equal revision loses",
220 current: BrokerSnapshot{Epoch: 3, Revision: 9},
221 candidate: BrokerSnapshot{Epoch: 3, Revision: 9},
222 wantNewer: false,
223 description: "replay of the same revision is not newer",
224 },
225 {
226 name: "epoch change forces resync",
227 current: BrokerSnapshot{Epoch: 3, Revision: 99},
228 candidate: BrokerSnapshot{Epoch: 4, Revision: 1},
229 wantNewer: true,
230 description: "any valid epoch change supersedes regardless of revision",
231 },
232 {
233 name: "lower epoch still forces resync",
234 current: BrokerSnapshot{Epoch: 9, Revision: 1},
235 candidate: BrokerSnapshot{Epoch: 4, Revision: 50},
236 wantNewer: true,
237 description: "epochs are fresh per process, never numerically ordered",
238 },
239 {
240 name: "zero epoch never supersedes",
241 current: BrokerSnapshot{Epoch: 3, Revision: 9},
242 candidate: BrokerSnapshot{},
243 wantNewer: false,
244 description: "invalid snapshots never apply",
245 },
246 }
247 for _, tt := range tests {
248 t.Run(tt.name, func(t *testing.T) {
249 if got := tt.candidate.Supersedes(tt.current); got != tt.wantNewer {
250 t.Fatalf("Supersedes() = %t, want %t (%s)", got, tt.wantNewer, tt.description)
251 }
252 })
253 }
254}
255
256func TestBrokerSnapshotCloneAndFind(t *testing.T) {
257 original := validBrokerSnapshot()
258 original.Daemons = append([]BrokerDaemonObservation{testLocalBrokerObservation()}, original.Daemons...)
259 original.Daemons[1].InventoryKnown = true
260 original.Daemons[1].Sessions = []catalogue.RemoteCatalogSession{{
261 LifecycleID: testBrokerLifecycle(),
262 Name: "work",
263 State: catalogue.RemoteCatalogSessionUp,
264 Tabs: []catalogue.RemoteCatalogTab{{ID: "tab-1", Index: 0, Name: "shell"}},
265 }}
266 cloned := original.Clone()
267 if found, ok := cloned.Find("user@arch"); !ok || found.Endpoint != "user@arch" {
268 t.Fatalf("Find() = %+v, %t, want the host", found, ok)
269 }
270 if _, ok := cloned.Find("user@mule"); ok {
271 t.Fatal("Find() must miss unknown endpoints")
272 }
273 local := 0
274 for _, daemon := range cloned.Daemons {
275 if daemon.Local {
276 local++
277 }
278 }
279 if local != 1 {
280 t.Fatalf("Clone() lost the local daemon entry: %d local entries", local)
281 }
282 cloned.Daemons[1].Sessions[0].Name = "mutated"
283 cloned.Daemons[1].Sessions[0].Tabs[0].ID = "mutated"
284 if original.Daemons[1].Sessions[0].Name != "work" || original.Daemons[1].Sessions[0].Tabs[0].ID != "tab-1" {
285 t.Fatal("Clone() shares memory with the original")
286 }
287}
288
289func TestBrokerTombstoneFencing(t *testing.T) {
290 registration := testBrokerRegistration("user@arch", 4, 2)
291 tombstone := BrokerHostTombstone{
292 Endpoint: "user@arch",
293 Registration: registration,
294 RetiredRevision: 12,
295 }
296 if err := tombstone.Validate(); err != nil {
297 t.Fatalf("Validate() = %v", err)
298 }
299 tests := []struct {
300 name string
301 candidate domain.RemoteRegistration
302 wantFence bool
303 }{
304 {name: "same registration fenced", candidate: registration, wantFence: true},
305 {name: "older generation fenced", candidate: testBrokerRegistration("user@arch", 4, 1), wantFence: true},
306 {name: "fresh incarnation not fenced", candidate: testBrokerRegistration("user@arch", 5, 1), wantFence: false},
307 {name: "other endpoint not fenced", candidate: testBrokerRegistration("user@mule", 4, 2), wantFence: false},
308 }
309 for _, tt := range tests {
310 t.Run(tt.name, func(t *testing.T) {
311 if got := tombstone.Fences(tt.candidate); got != tt.wantFence {
312 t.Fatalf("Fences() = %t, want %t", got, tt.wantFence)
313 }
314 })
315 }
316
317 invalid := []BrokerHostTombstone{
318 {Registration: registration, RetiredRevision: 1},
319 {Endpoint: "user@arch", Registration: testBrokerRegistration("user@other", 4, 2), RetiredRevision: 1},
320 {Endpoint: "user@arch", Registration: registration},
321 }
322 for i, tomb := range invalid {
323 if err := tomb.Validate(); err == nil {
324 t.Fatalf("invalid tombstone %d accepted", i)
325 }
326 }
327}
328
329func TestBrokerPolicyCompatibility(t *testing.T) {
330 base := testBrokerPolicy()
331 if err := base.Validate(); err != nil {
332 t.Fatalf("Validate() = %v", err)
333 }
334 if !base.Compatible(base) {
335 t.Fatal("identical policy must be compatible")
336 }
337 conflicts := map[string]BrokerPolicy{
338 "protocol version": withBrokerPolicy(base, func(p *BrokerPolicy) { p.ProtocolVersion++ }),
339 "catalog schema": withBrokerPolicy(base, func(p *BrokerPolicy) { p.CatalogSchemaVersion++ }),
340 "environment": withBrokerPolicy(base, func(p *BrokerPolicy) { p.EnvironmentPolicy = protocol.EnvironmentPolicyClientOwned }),
341 "transport": withBrokerPolicy(base, func(p *BrokerPolicy) { p.Transport = "stdio" }),
342 "trust": withBrokerPolicy(base, func(p *BrokerPolicy) { p.Trust = "other" }),
343 "launch": withBrokerPolicy(base, func(p *BrokerPolicy) { p.Launch = "other" }),
344 "isolation": withBrokerPolicy(base, func(p *BrokerPolicy) { p.Isolation = "other" }),
345 }
346 for name, conflict := range conflicts {
347 if base.Compatible(conflict) {
348 t.Fatalf("conflicting policy %q must be rejected, never merged", name)
349 }
350 }
351 invalid := []BrokerPolicy{
352 withBrokerPolicy(base, func(p *BrokerPolicy) { p.ProtocolVersion = 0 }),
353 withBrokerPolicy(base, func(p *BrokerPolicy) { p.CatalogSchemaVersion = 0 }),
354 withBrokerPolicy(base, func(p *BrokerPolicy) { p.EnvironmentPolicy = protocol.EnvironmentPolicy(9) }),
355 withBrokerPolicy(base, func(p *BrokerPolicy) { p.Transport = "" }),
356 }
357 for i, policy := range invalid {
358 if err := policy.Validate(); err == nil {
359 t.Fatalf("invalid policy %d accepted", i)
360 }
361 }
362}
363
364func TestBrokerOpenStreamRequestValidate(t *testing.T) {
365 valid := func() BrokerOpenStreamRequest {
366 return BrokerOpenStreamRequest{
367 Epoch: 1, Purpose: BrokerStreamAttachment,
368 Admission: BrokerAdmissionExact,
369 Connection: BrokerConnectionID{1},
370 Stream: 2,
371 Endpoint: "user@arch",
372 Registration: testBrokerRegistration("user@arch", 4, 2),
373 Target: testBrokerTarget(),
374 Env: []string{"TERM=xterm-256color"},
375 Policy: testBrokerPolicy(),
376 StartMode: BrokerDaemonStartIfNeeded,
377 }
378 }
379 if err := valid().Validate(); err != nil {
380 t.Fatalf("Validate() = %v", err)
381 }
382 tests := []struct {
383 name string
384 mutate func(*BrokerOpenStreamRequest)
385 wantErr string
386 }{
387 {name: "zero connection", mutate: func(r *BrokerOpenStreamRequest) { r.Connection = BrokerConnectionID{} }},
388 {name: "zero stream", mutate: func(r *BrokerOpenStreamRequest) { r.Stream = 0 }},
389 {name: "empty endpoint", mutate: func(r *BrokerOpenStreamRequest) { r.Endpoint = "" }},
390 {name: "endpoint registration mismatch", mutate: func(r *BrokerOpenStreamRequest) { r.Registration = testBrokerRegistration("user@other", 4, 2) }},
391 {name: "zero target lifecycle", mutate: func(r *BrokerOpenStreamRequest) {
392 r.Target = protocol.ExactSessionTarget{SessionName: "work"}
393 }},
394 {name: "bad session name", mutate: func(r *BrokerOpenStreamRequest) {
395 r.Target = protocol.ExactSessionTarget{LifecycleID: testBrokerLifecycle(), SessionName: "bad name!"}
396 }},
397 {name: "exact admission carries a creation name", mutate: func(r *BrokerOpenStreamRequest) {
398 r.Name = "work"
399 }},
400 {name: "named creation carries an exact target", mutate: func(r *BrokerOpenStreamRequest) {
401 r.Admission = BrokerAdmissionCreateNamed
402 r.Name = "work"
403 }},
404 {name: "named creation with invalid name", mutate: func(r *BrokerOpenStreamRequest) {
405 r.Admission = BrokerAdmissionCreateNamed
406 r.Name = "bad name!"
407 r.Target = protocol.ExactSessionTarget{}
408 }},
409 {name: "ephemeral creation carries a name", mutate: func(r *BrokerOpenStreamRequest) {
410 r.Admission = BrokerAdmissionCreateEphemeral
411 r.Target = protocol.ExactSessionTarget{}
412 r.Name = "work"
413 }},
414 {name: "ephemeral creation carries a target", mutate: func(r *BrokerOpenStreamRequest) {
415 r.Admission = BrokerAdmissionCreateEphemeral
416 }},
417 {name: "too many env entries", mutate: func(r *BrokerOpenStreamRequest) {
418 r.Env = make([]string, BrokerMaxEnvEntries+1)
419 }},
420 {name: "oversize env entry", mutate: func(r *BrokerOpenStreamRequest) {
421 r.Env = []string{strings.Repeat("A", BrokerMaxEnvEntryBytes+1)}
422 }},
423 {name: "invalid env encoding", mutate: func(r *BrokerOpenStreamRequest) {
424 r.Env = []string{"TERM=\xff"}
425 }},
426 {name: "invalid policy", mutate: func(r *BrokerOpenStreamRequest) { r.Policy = BrokerPolicy{} }},
427 {name: "zero start mode", mutate: func(r *BrokerOpenStreamRequest) { r.StartMode = 0 }},
428 {name: "unknown start mode", mutate: func(r *BrokerOpenStreamRequest) { r.StartMode = BrokerDaemonStartMode(9) }},
429 {name: "attachment existing only", mutate: func(r *BrokerOpenStreamRequest) { r.StartMode = BrokerDaemonExistingOnly }},
430 }
431 for _, tt := range tests {
432 t.Run(tt.name, func(t *testing.T) {
433 request := valid()
434 tt.mutate(&request)
435 if err := request.Validate(); err == nil {
436 t.Fatal("invalid open-stream request accepted")
437 }
438 })
439 }
440}
441
442func TestBrokerMutationOutcomeUncertainty(t *testing.T) {
443 for _, outcome := range []BrokerMutationOutcome{BrokerOutcomeOK, BrokerOutcomeFailed, BrokerOutcomeUnknown} {
444 if err := outcome.Validate(); err != nil {
445 t.Fatalf("Validate(%v) = %v", outcome, err)
446 }
447 }
448 if err := BrokerMutationOutcome(0).Validate(); err == nil {
449 t.Fatal("zero mutation outcome accepted")
450 }
451 if BrokerOutcomeOK.String() != "ok" || BrokerOutcomeUnknown.String() != "outcome_unknown" {
452 t.Fatal("mutation outcome tokens changed")
453 }
454
455 valid := BrokerOperationResult{Operation: BrokerOperationID{1}, Outcome: BrokerOutcomeFailed, Text: "dial refused"}
456 if err := valid.Validate(); err != nil {
457 t.Fatalf("Validate() = %v", err)
458 }
459 unknown := BrokerOperationResult{Operation: BrokerOperationID{1}, Outcome: BrokerOutcomeUnknown, Text: "lost after commit"}
460 if err := unknown.Validate(); err != nil {
461 t.Fatalf("outcome-unknown result must stay valid: %v", err)
462 }
463 okWithText := BrokerOperationResult{Operation: BrokerOperationID{1}, Outcome: BrokerOutcomeOK, Text: "extra"}
464 if err := okWithText.Validate(); err == nil {
465 t.Fatal("successful result must not carry text")
466 }
467 oversize := BrokerOperationResult{Operation: BrokerOperationID{1}, Outcome: BrokerOutcomeFailed, Text: strings.Repeat("e", BrokerMaxErrorBytes+1)}
468 if err := oversize.Validate(); err == nil {
469 t.Fatal("oversize result text accepted")
470 }
471 unsafe := BrokerOperationResult{Operation: BrokerOperationID{1}, Outcome: BrokerOutcomeFailed, Text: "dial\x1b[31m refused"}
472 if err := unsafe.Validate(); err == nil {
473 t.Fatal("terminal control sequence in operation result accepted")
474 }
475}
476
477func TestBrokerErrorTerminalOutcomes(t *testing.T) {
478 tests := []struct {
479 code BrokerErrorCode
480 wantTerminal bool
481 }{
482 {code: BrokerErrorUnavailable, wantTerminal: false},
483 {code: BrokerErrorIncompatible, wantTerminal: false},
484 {code: BrokerErrorTimeout, wantTerminal: false},
485 {code: BrokerErrorCancelled, wantTerminal: false},
486 {code: BrokerErrorOutcomeUnknown, wantTerminal: false},
487 {code: BrokerErrorAttachmentLost, wantTerminal: false},
488 {code: BrokerErrorStaleEpoch, wantTerminal: false},
489 {code: BrokerErrorConflictingPolicy, wantTerminal: false},
490 {code: BrokerErrorExplicitExit, wantTerminal: true},
491 {code: BrokerErrorFatalTerminal, wantTerminal: true},
492 {code: BrokerErrorHostConflict, wantTerminal: false},
493 {code: BrokerErrorMembershipImmutable, wantTerminal: false},
494 }
495 for _, tt := range tests {
496 t.Run(tt.code.String(), func(t *testing.T) {
497 failure := BrokerError{Code: tt.code, Text: "visible typed error"}
498 if err := failure.Validate(); err != nil {
499 t.Fatalf("Validate() = %v", err)
500 }
501 if got := failure.Terminal(); got != tt.wantTerminal {
502 t.Fatalf("Terminal() = %t, want %t", got, tt.wantTerminal)
503 }
504 if failure.Error() == "" {
505 t.Fatal("Error() must render a message")
506 }
507 })
508 }
509 if err := (BrokerError{Code: BrokerErrorCode(0)}).Validate(); err == nil {
510 t.Fatal("zero broker error code accepted")
511 }
512 oversize := BrokerError{Code: BrokerErrorTimeout, Text: strings.Repeat("e", BrokerMaxErrorBytes+1)}
513 if err := oversize.Validate(); err == nil {
514 t.Fatal("oversize broker error text accepted")
515 }
516 unsafe := BrokerError{Code: BrokerErrorTimeout, Text: "timeout\u202erorrE"}
517 if err := unsafe.Validate(); err == nil {
518 t.Fatal("bidirectional control in broker error accepted")
519 }
520}
521
522func TestBrokerIdentityScopesStayDistinct(t *testing.T) {
523 // Every identity level validates independently and zero values never
524 // identify anything: connection, stream, operation, daemon incarnation,
525 // registration, session lifecycle, and broker epoch are distinct types.
526 if err := (BrokerConnectionID{}).Validate(); err == nil {
527 t.Fatal("zero connection ID accepted")
528 }
529 if err := BrokerStreamID(0).Validate(); err == nil {
530 t.Fatal("zero stream ID accepted")
531 }
532 if err := (BrokerOperationID{}).Validate(); err == nil {
533 t.Fatal("zero operation ID accepted")
534 }
535 if err := (BrokerDaemonIncarnation{}).Validate(); err == nil {
536 t.Fatal("zero daemon incarnation accepted")
537 }
538 if err := (domain.RemoteRegistration{}).Validate(); err == nil {
539 t.Fatal("zero host registration accepted")
540 }
541 if err := (protocol.ExactSessionTarget{}).Validate(); err == nil {
542 t.Fatal("zero session lifecycle target accepted")
543 }
544 if err := BrokerDaemonIdentity("").Validate(); err == nil {
545 t.Fatal("empty daemon identity accepted")
546 }
547 if err := (BrokerStreamLost{}).Validate(); err == nil {
548 t.Fatal("zero stream loss accepted")
549 }
550 loss := BrokerStreamLost{
551 Connection: BrokerConnectionID{1},
552 Stream: 2,
553 Epoch: 3,
554 Cause: domain.RemoteFailureTimeout,
555 }
556 if err := loss.Validate(); err != nil {
557 t.Fatalf("Validate() = %v", err)
558 }
559}
560
561func withBrokerPolicy(base BrokerPolicy, mutate func(*BrokerPolicy)) BrokerPolicy {
562 mutate(&base)
563 return base
564}
565
566func TestBrokerInterfacesArePorts(t *testing.T) {
567 // Broker, client, and daemon cross-layer contracts are ports: the
568 // seams must stay interfaces so Mockery can generate mocks and
569 // adapters can implement them without importing use cases.
570 for name, iface := range map[string]reflect.Type{
571 "BrokerService": reflect.TypeOf((*BrokerService)(nil)).Elem(),
572 "BrokerRouteAuthority": reflect.TypeOf((*BrokerRouteAuthority)(nil)).Elem(),
573 "BrokerIdentityBinder": reflect.TypeOf((*BrokerIdentityBinder)(nil)).Elem(),
574 "BrokerLogicalConnection": reflect.TypeOf((*BrokerLogicalConnection)(nil)).Elem(),
575 "BrokerSnapshotStore": reflect.TypeOf((*BrokerSnapshotStore)(nil)).Elem(),
576 "BrokerHostProbe": reflect.TypeOf((*BrokerHostProbe)(nil)).Elem(),
577 "BrokerPhysicalConnection": reflect.TypeOf((*BrokerPhysicalConnection)(nil)).Elem(),
578 "BrokerEndpointConnector": reflect.TypeOf((*BrokerEndpointConnector)(nil)).Elem(),
579 "BrokerConnector": reflect.TypeOf((*BrokerConnector)(nil)).Elem(),
580 "BrokerAuthority": reflect.TypeOf((*BrokerAuthority)(nil)).Elem(),
581 "BrokerListener": reflect.TypeOf((*BrokerListener)(nil)).Elem(),
582 "BrokerSubscription": reflect.TypeOf((*BrokerSubscription)(nil)).Elem(),
583 } {
584 if iface.Kind() != reflect.Interface {
585 t.Fatalf("%s must stay an interface, got %v", name, iface.Kind())
586 }
587 }
588}
589
590// TestSanitizeBrokerDisplayText pins the sanitizer to the display validator:
591// every result is valid UTF-8 and carries no rune validateBrokerDisplayText
592// refuses, so a producer can sanitize a derived presentation hint and never
593// publish a value its own contract rejects.
594func TestSanitizeBrokerDisplayText(t *testing.T) {
595 tests := []struct {
596 name string
597 input string
598 want string
599 }{
600 {name: "ordinary text is returned unchanged", input: "user@arch", want: "user@arch"},
601 {name: "login-prefixed origin is returned unchanged", input: "user@host.example:2222", want: "user@host.example:2222"},
602 {name: "accented and emoji text is kept intact", input: "caf\u00e9 \U0001f5a5", want: "caf\u00e9 \U0001f5a5"},
603 {name: "right-to-left override is dropped", input: "sh\u202eell", want: "shell"},
604 {name: "left-to-right embedding is dropped", input: "a\u202ab", want: "ab"},
605 {name: "C0 control is dropped", input: "a\x00b\x1bc", want: "abc"},
606 {name: "line separator is dropped", input: "a\u2028b", want: "ab"},
607 {name: "paragraph separator is dropped", input: "a\u2029b", want: "ab"},
608 {name: "invalid utf-8 is replaced", input: "ok\xff\xfe", want: "ok\ufffd\ufffd"},
609 {name: "every rune disallowed yields empty", input: "\u202e\u2028\x00", want: ""},
610 }
611 for _, tc := range tests {
612 t.Run(tc.name, func(t *testing.T) {
613 got := SanitizeBrokerDisplayText(tc.input)
614 require.Equal(t, tc.want, got)
615 require.True(t, utf8.ValidString(got), "sanitized text must always be valid UTF-8")
616 require.Equal(t, -1, strings.IndexFunc(got, func(r rune) bool { return !brokerDisplayRuneAllowed(r) }), "sanitized text must carry no disallowed rune")
617 require.NoError(t, validateBrokerDisplayText(got, BrokerMaxDisplayOriginBytes, "display origin"))
618 })
619 }
620}
621
622// TestBrokerSnapshotValidateLocalDaemonPlacement pins the wire placement rule:
623// the local daemon observation is carried at index zero only, a local
624// observation anywhere else is refused, and a local daemon is still refused
625// when it appears twice.
626func TestBrokerSnapshotValidateLocalDaemonPlacement(t *testing.T) {
627 tests := []struct {
628 name string
629 daemons []BrokerDaemonObservation
630 wantErrContains string
631 }{
632 {
633 name: "local daemon first",
634 daemons: []BrokerDaemonObservation{testLocalBrokerObservation(), testBrokerObservation("user@arch")},
635 },
636 {
637 name: "local daemon at index one is refused",
638 daemons: []BrokerDaemonObservation{testBrokerObservation("user@arch"), testLocalBrokerObservation()},
639 wantErrContains: "local daemon at index 1",
640 },
641 {
642 name: "local daemon after two remotes is refused",
643 daemons: []BrokerDaemonObservation{testBrokerObservation("user@arch"), testBrokerObservation("user@other"), testLocalBrokerObservation()},
644 wantErrContains: "local daemon at index 2",
645 },
646 {
647 name: "two local daemons are refused",
648 daemons: []BrokerDaemonObservation{testLocalBrokerObservation(), testLocalBrokerObservation()},
649 wantErrContains: "more than one local daemon",
650 },
651 }
652 for _, tc := range tests {
653 t.Run(tc.name, func(t *testing.T) {
654 snapshot := validBrokerSnapshot()
655 snapshot.Daemons = tc.daemons
656 err := snapshot.Validate()
657 if tc.wantErrContains != "" {
658 require.ErrorContains(t, err, tc.wantErrContains)
659 return
660 }
661 require.NoError(t, err)
662 })
663 }
664}
665
666func TestBrokerStreamPurposeShapes(t *testing.T) {
667 for _, tc := range []struct {
668 name string
669 purpose BrokerStreamPurpose
670 admission BrokerStreamAdmission
671 createName string
672 target bool
673 env bool
674 startMode BrokerDaemonStartMode
675 valid bool
676 }{
677 {"attachment exact", BrokerStreamAttachment, BrokerAdmissionExact, "", true, true, BrokerDaemonStartIfNeeded, true},
678 {"attachment missing target", BrokerStreamAttachment, BrokerAdmissionExact, "", false, false, BrokerDaemonStartIfNeeded, false},
679 {"attachment missing admission", BrokerStreamAttachment, 0, "", true, false, BrokerDaemonStartIfNeeded, false},
680 {"attachment named creation", BrokerStreamAttachment, BrokerAdmissionCreateNamed, "work", false, true, BrokerDaemonStartIfNeeded, true},
681 {"attachment ephemeral creation", BrokerStreamAttachment, BrokerAdmissionCreateEphemeral, "", false, true, BrokerDaemonStartIfNeeded, true},
682 {"attachment existing only", BrokerStreamAttachment, BrokerAdmissionCreateEphemeral, "", false, true, BrokerDaemonExistingOnly, false},
683 {"attachment zero start mode", BrokerStreamAttachment, BrokerAdmissionCreateEphemeral, "", false, true, 0, false},
684 {"control", BrokerStreamControl, 0, "", false, false, BrokerDaemonStartIfNeeded, true},
685 {"control existing only", BrokerStreamControl, 0, "", false, false, BrokerDaemonExistingOnly, true},
686 {"observation", BrokerStreamObservation, 0, "", false, false, BrokerDaemonExistingOnly, true},
687 {"observation start if needed", BrokerStreamObservation, 0, "", false, false, BrokerDaemonStartIfNeeded, false},
688 {"observation zero start mode", BrokerStreamObservation, 0, "", false, false, 0, false},
689 {"control target", BrokerStreamControl, 0, "", true, false, BrokerDaemonStartIfNeeded, false},
690 {"control admission", BrokerStreamControl, BrokerAdmissionExact, "", false, false, BrokerDaemonStartIfNeeded, false},
691 {"observation env", BrokerStreamObservation, 0, "", false, true, BrokerDaemonExistingOnly, false},
692 {"observation admission", BrokerStreamObservation, BrokerAdmissionCreateEphemeral, "", false, false, BrokerDaemonExistingOnly, false},
693 {"unknown", 0, 0, "", false, false, BrokerDaemonStartIfNeeded, false},
694 } {
695 t.Run(tc.name, func(t *testing.T) {
696 r := BrokerOpenStreamRequest{Epoch: 1, Connection: BrokerConnectionID{1}, Stream: 1, Local: true, Purpose: tc.purpose, Admission: tc.admission, Name: tc.createName, Policy: testBrokerPolicy(), StartMode: tc.startMode}
697 if tc.target {
698 r.Target = testBrokerTarget()
699 }
700 if tc.env {
701 r.Env = []string{"TERM=xterm"}
702 }
703 if (r.Validate() == nil) != tc.valid {
704 t.Fatalf("Validate()=%v", r.Validate())
705 }
706 })
707 }
708}