detached 3b243bf

test(app): drive the broker through production entry points

commit
3b243bfa2de23b4307f0f305d633165d8cda42d1
author
brice · · 2026-09-23 14:41 UTC
parent
7f208e1
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
2 files · +177 -1
Unified
UnifiedSplit
A internal/app/broker_composition_test.go +176 -0
@@ -0,0 +1,176 @@
1+package app
2+
3+import (
4+ "context"
5+ "os"
6+ "os/exec"
7+ "path/filepath"
8+ "strings"
9+ "syscall"
10+ "testing"
11+ "time"
12+
13+ "github.com/bnema/vev/internal/adapters/daemonidentity"
14+ "github.com/bnema/vev/internal/adapters/daemonmux"
15+ "github.com/bnema/vev/internal/adapters/ipc"
16+ "github.com/bnema/vev/internal/domain"
17+ "github.com/bnema/vev/internal/ports"
18+ "github.com/bnema/vev/internal/protocol"
19+ "github.com/stretchr/testify/require"
20+)
21+
22+// startProductionBrokerFixture uses the same private XDG layout and real daemon
23+// entry point as a deployed broker. In particular no test provisions a local
24+// identity or socket in broker.json: the daemon publishes its own authority.
25+func startProductionBrokerFixture(t *testing.T) (ports.BrokerService, context.Context) {
26+ t.Helper()
27+ t.Setenv("VEV_ENV", "")
28+ t.Setenv("VEV_ENV_ROOT", "")
29+ t.Setenv("XDG_RUNTIME_DIR", shortTempDir(t, "vb"))
30+ t.Setenv("XDG_STATE_HOME", shortTempDir(t, "vb"))
31+ t.Setenv("XDG_CONFIG_HOME", shortTempDir(t, "vb"))
32+ t.Setenv(brokerHelperEnv, "1")
33+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
34+ t.Cleanup(cancel)
35+ daemon := exec.Command(os.Args[0], "--daemon")
36+ daemon.Env = os.Environ()
37+ daemon.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
38+ require.NoError(t, daemon.Start())
39+ daemonDone := make(chan error, 1)
40+ go func() { daemonDone <- daemon.Wait() }()
41+ t.Cleanup(func() {
42+ _ = daemon.Process.Signal(syscall.SIGTERM)
43+ select {
44+ case <-daemonDone:
45+ case <-time.After(5 * time.Second):
46+ _ = daemon.Process.Kill()
47+ <-daemonDone
48+ }
49+ })
50+ muxPath := daemonmux.SocketPath(ipc.SocketDir())
51+ require.Eventually(t, func() bool {
52+ info, err := os.Stat(muxPath)
53+ return err == nil && info.Mode()&os.ModeSocket != 0
54+ }, 10*time.Second, 10*time.Millisecond, "daemon never published its mux socket at %s", muxPath)
55+ identity, err := daemonidentity.Load(filepath.Join(os.Getenv("XDG_STATE_HOME"), "vev"))
56+ require.NoError(t, err)
57+ require.NotEmpty(t, identity)
58+ broker := exec.Command(os.Args[0], productionBrokerServeCommand)
59+ broker.Env = os.Environ()
60+ broker.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
61+ require.NoError(t, broker.Start())
62+ brokerDone := make(chan error, 1)
63+ go func() { brokerDone <- broker.Wait() }()
64+ t.Cleanup(func() {
65+ _ = broker.Process.Signal(syscall.SIGTERM)
66+ select {
67+ case <-brokerDone:
68+ case <-time.After(5 * time.Second):
69+ _ = broker.Process.Kill()
70+ <-brokerDone
71+ }
72+ })
73+ var service ports.BrokerService
74+ require.Eventually(t, func() bool {
75+ var dialErr error
76+ service, dialErr = connectExistingBroker(ctx)
77+ return dialErr == nil
78+ }, 10*time.Second, 10*time.Millisecond)
79+ t.Cleanup(func() { require.NoError(t, service.Close()) })
80+ return service, ctx
81+}
82+
83+func TestProductionBrokerCompositionLocalAndMutableRemote(t *testing.T) {
84+ service, ctx := startProductionBrokerFixture(t)
85+ sub, err := service.Subscribe()
86+ require.NoError(t, err)
87+ defer sub.Close()
88+ require.NotEmpty(t, service.Snapshot().Daemons)
89+ require.True(t, service.Snapshot().Daemons[0].Local)
90+
91+ streamID, err := service.NextStreamID()
92+ require.NoError(t, err)
93+ stream, err := service.OpenStream(ctx, ports.BrokerOpenStreamRequest{
94+ Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionCreateNamed,
95+ Name: "production-composition", Local: true, Stream: streamID,
96+ Policy: localDaemonPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded,
97+ })
98+ require.NoError(t, err)
99+ defer stream.Close()
100+ require.NoError(t, stream.SendClient(protocol.Hello{
101+ Version: protocol.Version, Intent: protocol.IntentNew, Name: "production-composition",
102+ Size: domain.Size{Cols: 80, Rows: 24}, EnvironmentPolicy: protocol.EnvironmentPolicyClientOwned,
103+ }))
104+ welcome := awaitWelcome(t, stream)
105+ require.NotNil(t, welcome.CommittedIdentity)
106+ require.Eventually(t, func() bool {
107+ snapshot := service.Snapshot()
108+ return len(snapshot.Daemons) > 0 && snapshot.Daemons[0].Identity != ""
109+ }, 10*time.Second, 10*time.Millisecond, "local observation: %+v", service.Snapshot())
110+
111+ registration, err := service.AddHost(ctx, "harness@127.0.0.1", remoteBrokerPolicy("stdio"))
112+ require.NoError(t, err)
113+ require.Equal(t, "harness@127.0.0.1", registration.Endpoint)
114+ require.Eventually(t, func() bool {
115+ _, ok := service.Snapshot().Find(registration.Endpoint)
116+ return ok
117+ }, 10*time.Second, 10*time.Millisecond)
118+ _, err = service.RemoveHost(ctx, registration)
119+ require.NoError(t, err)
120+}
121+
122+// Both production remote transports are reached through mutable AddHost,
123+// never through a broker.json fixture route. SSH is replaced by an owned test
124+// shim; the remote mux helper and daemon are real subprocesses.
125+func TestProductionBrokerRemoteMuxTransports(t *testing.T) {
126+ for _, tc := range []struct {
127+ name, helper string
128+ }{
129+ {"stdio", brokerMuxStdioCommand},
130+ {"quic", brokerMuxQUICBootstrapCommand},
131+ } {
132+ t.Run(tc.name, func(t *testing.T) {
133+ binDir := shortTempDir(t, "vb")
134+ argsFile := filepath.Join(binDir, "ssh-args")
135+ script := "#!/bin/sh\nprintf '%s\\n' \"$@\" > \"$VEV_TEST_SSH_ARGS\"\nexec \"$VEV_TEST_SSH_BIN\" \"$VEV_TEST_SSH_MODE\" --production --daemon-start existing-only\n"
136+ require.NoError(t, os.WriteFile(filepath.Join(binDir, "ssh"), []byte(script), 0o755))
137+ t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH"))
138+ t.Setenv(fakeSSHArgsEnv, argsFile)
139+ t.Setenv(fakeSSHBinEnv, os.Args[0])
140+ t.Setenv(fakeSSHModeEnv, tc.helper)
141+ t.Setenv(brokerHelperRecordDirEnv, binDir)
142+ t.Cleanup(func() {
143+ for _, pid := range recordedPIDsIn(t, binDir, brokerMuxQUICProxyCommand) {
144+ _ = syscall.Kill(pid, syscall.SIGTERM)
145+ _ = waitForProcessExit(pid, 5*time.Second)
146+ }
147+ })
148+ service, ctx := startProductionBrokerFixture(t)
149+ registration, err := service.AddHost(ctx, "harness@127.0.0.1", remoteBrokerPolicy(tc.name))
150+ require.NoError(t, err)
151+ streamID, err := service.NextStreamID()
152+ require.NoError(t, err)
153+ stream, err := service.OpenStream(ctx, ports.BrokerOpenStreamRequest{
154+ Purpose: ports.BrokerStreamObservation, Stream: streamID,
155+ Endpoint: registration.Endpoint, Registration: registration,
156+ Policy: remoteBrokerPolicy(tc.name), StartMode: ports.BrokerDaemonExistingOnly,
157+ })
158+ require.NoError(t, err)
159+ defer stream.Close()
160+ require.NoError(t, stream.SendClient(protocol.CommandRequest{RequestID: 1, Version: protocol.Version, Slug: "remote-catalog", JSON: true}))
161+ message, err := stream.ReceiveServer()
162+ require.NoError(t, err)
163+ result, ok := message.(protocol.CommandResult)
164+ require.True(t, ok, "daemon observation reply: %T %v", message, message)
165+ require.Equal(t, protocol.CommandSucceeded, result.Outcome)
166+ argv, err := os.ReadFile(argsFile)
167+ require.NoError(t, err)
168+ require.Contains(t, string(argv), tc.helper)
169+ require.Contains(t, string(argv), "-T")
170+ require.Contains(t, string(argv), "--")
171+ require.NotContains(t, strings.ToLower(string(argv)), "stricthostkeychecking=no")
172+ _, err = service.RemoveHost(ctx, registration)
173+ require.NoError(t, err)
174+ })
175+ }
176+}
@@ -0,0 +1,176 @@
1+package app
2+
3+import (
4+ "context"
5+ "os"
6+ "os/exec"
7+ "path/filepath"
8+ "strings"
9+ "syscall"
10+ "testing"
11+ "time"
12+
13+ "github.com/bnema/vev/internal/adapters/daemonidentity"
14+ "github.com/bnema/vev/internal/adapters/daemonmux"
15+ "github.com/bnema/vev/internal/adapters/ipc"
16+ "github.com/bnema/vev/internal/domain"
17+ "github.com/bnema/vev/internal/ports"
18+ "github.com/bnema/vev/internal/protocol"
19+ "github.com/stretchr/testify/require"
20+)
21+
22+// startProductionBrokerFixture uses the same private XDG layout and real daemon
23+// entry point as a deployed broker. In particular no test provisions a local
24+// identity or socket in broker.json: the daemon publishes its own authority.
25+func startProductionBrokerFixture(t *testing.T) (ports.BrokerService, context.Context) {
26+ t.Helper()
27+ t.Setenv("VEV_ENV", "")
28+ t.Setenv("VEV_ENV_ROOT", "")
29+ t.Setenv("XDG_RUNTIME_DIR", shortTempDir(t, "vb"))
30+ t.Setenv("XDG_STATE_HOME", shortTempDir(t, "vb"))
31+ t.Setenv("XDG_CONFIG_HOME", shortTempDir(t, "vb"))
32+ t.Setenv(brokerHelperEnv, "1")
33+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
34+ t.Cleanup(cancel)
35+ daemon := exec.Command(os.Args[0], "--daemon")
36+ daemon.Env = os.Environ()
37+ daemon.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
38+ require.NoError(t, daemon.Start())
39+ daemonDone := make(chan error, 1)
40+ go func() { daemonDone <- daemon.Wait() }()
41+ t.Cleanup(func() {
42+ _ = daemon.Process.Signal(syscall.SIGTERM)
43+ select {
44+ case <-daemonDone:
45+ case <-time.After(5 * time.Second):
46+ _ = daemon.Process.Kill()
47+ <-daemonDone
48+ }
49+ })
50+ muxPath := daemonmux.SocketPath(ipc.SocketDir())
51+ require.Eventually(t, func() bool {
52+ info, err := os.Stat(muxPath)
53+ return err == nil && info.Mode()&os.ModeSocket != 0
54+ }, 10*time.Second, 10*time.Millisecond, "daemon never published its mux socket at %s", muxPath)
55+ identity, err := daemonidentity.Load(filepath.Join(os.Getenv("XDG_STATE_HOME"), "vev"))
56+ require.NoError(t, err)
57+ require.NotEmpty(t, identity)
58+ broker := exec.Command(os.Args[0], productionBrokerServeCommand)
59+ broker.Env = os.Environ()
60+ broker.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
61+ require.NoError(t, broker.Start())
62+ brokerDone := make(chan error, 1)
63+ go func() { brokerDone <- broker.Wait() }()
64+ t.Cleanup(func() {
65+ _ = broker.Process.Signal(syscall.SIGTERM)
66+ select {
67+ case <-brokerDone:
68+ case <-time.After(5 * time.Second):
69+ _ = broker.Process.Kill()
70+ <-brokerDone
71+ }
72+ })
73+ var service ports.BrokerService
74+ require.Eventually(t, func() bool {
75+ var dialErr error
76+ service, dialErr = connectExistingBroker(ctx)
77+ return dialErr == nil
78+ }, 10*time.Second, 10*time.Millisecond)
79+ t.Cleanup(func() { require.NoError(t, service.Close()) })
80+ return service, ctx
81+}
82+
83+func TestProductionBrokerCompositionLocalAndMutableRemote(t *testing.T) {
84+ service, ctx := startProductionBrokerFixture(t)
85+ sub, err := service.Subscribe()
86+ require.NoError(t, err)
87+ defer sub.Close()
88+ require.NotEmpty(t, service.Snapshot().Daemons)
89+ require.True(t, service.Snapshot().Daemons[0].Local)
90+
91+ streamID, err := service.NextStreamID()
92+ require.NoError(t, err)
93+ stream, err := service.OpenStream(ctx, ports.BrokerOpenStreamRequest{
94+ Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionCreateNamed,
95+ Name: "production-composition", Local: true, Stream: streamID,
96+ Policy: localDaemonPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded,
97+ })
98+ require.NoError(t, err)
99+ defer stream.Close()
100+ require.NoError(t, stream.SendClient(protocol.Hello{
101+ Version: protocol.Version, Intent: protocol.IntentNew, Name: "production-composition",
102+ Size: domain.Size{Cols: 80, Rows: 24}, EnvironmentPolicy: protocol.EnvironmentPolicyClientOwned,
103+ }))
104+ welcome := awaitWelcome(t, stream)
105+ require.NotNil(t, welcome.CommittedIdentity)
106+ require.Eventually(t, func() bool {
107+ snapshot := service.Snapshot()
108+ return len(snapshot.Daemons) > 0 && snapshot.Daemons[0].Identity != ""
109+ }, 10*time.Second, 10*time.Millisecond, "local observation: %+v", service.Snapshot())
110+
111+ registration, err := service.AddHost(ctx, "harness@127.0.0.1", remoteBrokerPolicy("stdio"))
112+ require.NoError(t, err)
113+ require.Equal(t, "harness@127.0.0.1", registration.Endpoint)
114+ require.Eventually(t, func() bool {
115+ _, ok := service.Snapshot().Find(registration.Endpoint)
116+ return ok
117+ }, 10*time.Second, 10*time.Millisecond)
118+ _, err = service.RemoveHost(ctx, registration)
119+ require.NoError(t, err)
120+}
121+
122+// Both production remote transports are reached through mutable AddHost,
123+// never through a broker.json fixture route. SSH is replaced by an owned test
124+// shim; the remote mux helper and daemon are real subprocesses.
125+func TestProductionBrokerRemoteMuxTransports(t *testing.T) {
126+ for _, tc := range []struct {
127+ name, helper string
128+ }{
129+ {"stdio", brokerMuxStdioCommand},
130+ {"quic", brokerMuxQUICBootstrapCommand},
131+ } {
132+ t.Run(tc.name, func(t *testing.T) {
133+ binDir := shortTempDir(t, "vb")
134+ argsFile := filepath.Join(binDir, "ssh-args")
135+ script := "#!/bin/sh\nprintf '%s\\n' \"$@\" > \"$VEV_TEST_SSH_ARGS\"\nexec \"$VEV_TEST_SSH_BIN\" \"$VEV_TEST_SSH_MODE\" --production --daemon-start existing-only\n"
136+ require.NoError(t, os.WriteFile(filepath.Join(binDir, "ssh"), []byte(script), 0o755))
137+ t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH"))
138+ t.Setenv(fakeSSHArgsEnv, argsFile)
139+ t.Setenv(fakeSSHBinEnv, os.Args[0])
140+ t.Setenv(fakeSSHModeEnv, tc.helper)
141+ t.Setenv(brokerHelperRecordDirEnv, binDir)
142+ t.Cleanup(func() {
143+ for _, pid := range recordedPIDsIn(t, binDir, brokerMuxQUICProxyCommand) {
144+ _ = syscall.Kill(pid, syscall.SIGTERM)
145+ _ = waitForProcessExit(pid, 5*time.Second)
146+ }
147+ })
148+ service, ctx := startProductionBrokerFixture(t)
149+ registration, err := service.AddHost(ctx, "harness@127.0.0.1", remoteBrokerPolicy(tc.name))
150+ require.NoError(t, err)
151+ streamID, err := service.NextStreamID()
152+ require.NoError(t, err)
153+ stream, err := service.OpenStream(ctx, ports.BrokerOpenStreamRequest{
154+ Purpose: ports.BrokerStreamObservation, Stream: streamID,
155+ Endpoint: registration.Endpoint, Registration: registration,
156+ Policy: remoteBrokerPolicy(tc.name), StartMode: ports.BrokerDaemonExistingOnly,
157+ })
158+ require.NoError(t, err)
159+ defer stream.Close()
160+ require.NoError(t, stream.SendClient(protocol.CommandRequest{RequestID: 1, Version: protocol.Version, Slug: "remote-catalog", JSON: true}))
161+ message, err := stream.ReceiveServer()
162+ require.NoError(t, err)
163+ result, ok := message.(protocol.CommandResult)
164+ require.True(t, ok, "daemon observation reply: %T %v", message, message)
165+ require.Equal(t, protocol.CommandSucceeded, result.Outcome)
166+ argv, err := os.ReadFile(argsFile)
167+ require.NoError(t, err)
168+ require.Contains(t, string(argv), tc.helper)
169+ require.Contains(t, string(argv), "-T")
170+ require.Contains(t, string(argv), "--")
171+ require.NotContains(t, strings.ToLower(string(argv)), "stricthostkeychecking=no")
172+ _, err = service.RemoveHost(ctx, registration)
173+ require.NoError(t, err)
174+ })
175+ }
176+}
M internal/app/spawn_test.go +1 -1
@@ -33,7 +33,7 @@ func TestMain(m *testing.M) {
3333 case brokerServeCommand, brokerLauncherCommand, brokerStatusCommand,
3434 brokerMuxStdioCommand, brokerMuxQUICBootstrapCommand, brokerMuxQUICProxyCommand,
3535 brokerReadyCommand,
36- productionBrokerServeCommand, productionBrokerLauncherCommand:
36+ productionBrokerServeCommand, productionBrokerLauncherCommand, "--daemon":
3737 recordBrokerHelperProcess(os.Args[1])
3838 if os.Args[1] == brokerLauncherCommand {
3939 if record := os.Getenv(brokerLauncherBlockRecordEnv); record != "" {
@@ -33,7 +33,7 @@ func TestMain(m *testing.M) {
33 case brokerServeCommand, brokerLauncherCommand, brokerStatusCommand,33 case brokerServeCommand, brokerLauncherCommand, brokerStatusCommand,
34 brokerMuxStdioCommand, brokerMuxQUICBootstrapCommand, brokerMuxQUICProxyCommand,34 brokerMuxStdioCommand, brokerMuxQUICBootstrapCommand, brokerMuxQUICProxyCommand,
35 brokerReadyCommand,35 brokerReadyCommand,
36- productionBrokerServeCommand, productionBrokerLauncherCommand:36+ productionBrokerServeCommand, productionBrokerLauncherCommand, "--daemon":
37 recordBrokerHelperProcess(os.Args[1])37 recordBrokerHelperProcess(os.Args[1])
38 if os.Args[1] == brokerLauncherCommand {38 if os.Args[1] == brokerLauncherCommand {
39 if record := os.Getenv(brokerLauncherBlockRecordEnv); record != "" {39 if record := os.Getenv(brokerLauncherBlockRecordEnv); record != "" {

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close