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