1package daemonmux
2
3import (
4 "testing"
5
6 "github.com/stretchr/testify/require"
7
8 "github.com/bnema/vev/internal/ports"
9)
10
11// TestStreamEngineRestrictAdmissions proves the accepted physical policy seam:
12// it is configured once before any admission, refuses a mismatched fresh Open
13// before that Open records anything, leaves a replayed identity classified as a
14// duplicate rather than a policy refusal, and admits a matching Open.
15func TestStreamEngineRestrictAdmissions(t *testing.T) {
16 engine := NewStreamEngine()
17 require.NoError(t, engine.RestrictAdmissions(testPolicy()))
18 require.ErrorIs(t, engine.RestrictAdmissions(testPolicy()), ErrStreamState, "the policy is fixed once")
19
20 mismatched := openFor(1)
21 mismatched.Policy = alternatePolicy(testPolicy())
22 require.ErrorIs(t, engine.Open(mismatched), ErrOpenPolicyRefused)
23 _, tracked := engine.Status(1)
24 require.False(t, tracked, "a policy refusal records no stream")
25 require.Zero(t, engine.Live())
26
27 matching := openFor(1)
28 matching.Policy = testPolicy()
29 require.NoError(t, engine.Open(matching))
30 require.Equal(t, 1, engine.Live())
31
32 // A duplicate identity is still fenced by reference/identity first, so a
33 // mismatched replay never disturbs the live stream it names.
34 replay := openFor(1)
35 replay.Policy = alternatePolicy(testPolicy())
36 require.ErrorIs(t, engine.Open(replay), ErrStreamIDReused)
37 require.Equal(t, 1, engine.Live())
38}
39
40// TestStreamEngineRestrictAdmissionsGuards proves the seam refuses an invalid
41// policy, a terminal engine, and a nil receiver, and that a policy gate is
42// optional: an unrestricted engine admits any validly encoded Open.
43func TestStreamEngineRestrictAdmissionsGuards(t *testing.T) {
44 var nilEngine *StreamEngine
45 require.ErrorIs(t, nilEngine.RestrictAdmissions(testPolicy()), ErrPhysicalClosed)
46
47 engine := NewStreamEngine()
48 require.Error(t, engine.RestrictAdmissions(ports.BrokerPolicy{}), "an invalid policy is refused")
49
50 unrestricted := NewStreamEngine()
51 require.NoError(t, unrestricted.Open(openFor(1)))
52
53 terminated := NewStreamEngine()
54 terminated.Terminate()
55 require.ErrorIs(t, terminated.RestrictAdmissions(testPolicy()), ErrPhysicalClosed)
56}