1package app
2
3// Real-subprocess coverage for the hidden `_broker-ready` probe.
4//
5// These tests re-execute the test binary through the production dispatch path
6// (TestMain intercepts the hidden broker commands when brokerHelperEnv is set),
7// so argument parsing, endpoint classification, the real broker IPC setup, the
8// readiness evaluation, the JSON document, and the process exit code are all
9// exercised exactly as an operator would observe them. A real production broker serve
10// composes the broker over isolated XDG directories, and the probe reads it
11// over a real AF_UNIX carriage.
12
13import (
14 "encoding/json"
15 "errors"
16 "net"
17 "os"
18 "os/exec"
19 "path/filepath"
20 "strings"
21 "syscall"
22 "testing"
23 "time"
24
25 "github.com/stretchr/testify/require"
26
27 "github.com/bnema/vev/internal/adapters/brokerconfig"
28 "github.com/bnema/vev/internal/adapters/brokeripc"
29)
30
31// readySubprocessSandbox isolates real production XDG directories and records every
32// spawned broker helper, exactly like the status subprocess sandbox.
33type readySubprocessSandbox struct {
34 t *testing.T
35 layout brokerconfig.Layout
36 recordDir string
37 broker *exec.Cmd
38}
39
40// newReadySubprocessSandbox provisions one empty production config and arranges
41// cleanup of every recorded broker process.
42func newReadySubprocessSandbox(t *testing.T) *readySubprocessSandbox {
43 t.Helper()
44 layout := emptyProductionBrokerLayout(t, "3s")
45 recordDir := filepath.Join(shortTempDir(t, "vevw"), "r")
46 require.NoError(t, os.MkdirAll(recordDir, 0o700))
47 t.Setenv(brokerHelperEnv, "1")
48 t.Setenv(brokerHelperRecordDirEnv, recordDir)
49 sandbox := &readySubprocessSandbox{t: t, layout: layout, recordDir: recordDir}
50 t.Cleanup(sandbox.cleanup)
51 return sandbox
52}
53
54// socketPath is the broker endpoint inside the sandbox runtime directory.
55func (s *readySubprocessSandbox) socketPath() string {
56 return brokeripc.SocketPath(s.layout.Runtime)
57}
58
59// ready runs `_broker-ready` for this sandbox with the supplied extra flags and
60// returns its stdout, stderr, and exit code.
61func (s *readySubprocessSandbox) ready(timeout string, extra ...string) (string, string, int) {
62 s.t.Helper()
63 args := append([]string{brokerReadyCommand, "--timeout", timeout}, extra...)
64 return runBrokerCommand(s.t, args...)
65}
66
67// startBroker runs one foreground `_broker-serve` helper in the background.
68func (s *readySubprocessSandbox) startBroker() {
69 s.t.Helper()
70 command := exec.Command(os.Args[0], productionBrokerServeCommand)
71 command.Env = os.Environ()
72 command.Stdout = os.Stdout
73 command.Stderr = os.Stderr
74 require.NoError(s.t, command.Start())
75 s.broker = command
76}
77
78// waitForBrokerSocket waits until the broker has bound its endpoint.
79func (s *readySubprocessSandbox) waitForBrokerSocket(timeout time.Duration) bool {
80 s.t.Helper()
81 deadline := time.Now().Add(timeout)
82 for time.Now().Before(deadline) {
83 if info, err := os.Lstat(s.socketPath()); err == nil && info.Mode()&os.ModeSocket != 0 {
84 return true
85 }
86 time.Sleep(10 * time.Millisecond)
87 }
88 return false
89}
90
91// cleanup terminates the background broker and every recorded serve process so
92// no test leaks an orphan.
93func (s *readySubprocessSandbox) cleanup() {
94 if s.broker != nil && s.broker.Process != nil {
95 _ = s.broker.Process.Signal(syscall.SIGTERM)
96 done := make(chan struct{})
97 go func() { _ = s.broker.Wait(); close(done) }()
98 select {
99 case <-done:
100 case <-time.After(10 * time.Second):
101 _ = s.broker.Process.Kill()
102 <-done
103 }
104 }
105 for _, pid := range recordedPIDsIn(s.t, s.recordDir, productionBrokerServeCommand) {
106 if err := syscall.Kill(pid, syscall.SIGKILL); err != nil && !errors.Is(err, syscall.ESRCH) {
107 continue
108 }
109 _ = waitForProcessExit(pid, 10*time.Second)
110 }
111}
112
113// readyDoc is the bounded readiness report the probe prints.
114type readyDoc struct {
115 Schema string `json:"schema"`
116 Status string `json:"status"`
117 Require string `json:"require"`
118 Reason string `json:"reason"`
119 Pending []string `json:"pending"`
120 Epoch string `json:"epoch"`
121}
122
123// decodeReadyReport decodes exactly one bounded JSON report from stdout.
124func decodeReadyReport(t *testing.T, raw string) readyDoc {
125 t.Helper()
126 trimmed := strings.TrimSpace(raw)
127 require.Equal(t, 1, strings.Count(trimmed, "\n")+1, "the probe prints exactly one report line")
128 var report readyDoc
129 require.NoError(t, json.Unmarshal([]byte(trimmed), &report))
130 require.Equal(t, "vev.broker-ready/v1", report.Schema)
131 return report
132}
133
134// TestBrokerReadySubprocessReadsARealBrokerAuthority proves the positive path
135// end to end: a real `_broker-serve` over an isolated root publishes its local
136// daemon entry, `--require local-authority` is satisfied with exit zero and the
137// broker epoch, and `--require local-catalogue` stays pending because the
138// provisioned local route is absent, so it is never falsely reported ready.
139func TestBrokerReadySubprocessReadsARealBrokerAuthority(t *testing.T) {
140 sandbox := newReadySubprocessSandbox(t)
141 // One provisioned local binding whose daemon is not started: the broker
142 // publishes its local entry as derived authority, so `local-authority` is
143 // observable while the catalogue is not.
144
145 sandbox.startBroker()
146 require.True(t, sandbox.waitForBrokerSocket(15*time.Second), "the broker must bind its endpoint")
147
148 stdout, stderr, code := sandbox.ready("5s", "--require", "local-authority")
149 require.Equal(t, 0, code, "stderr: %s", stderr)
150 authority := decodeReadyReport(t, stdout)
151 require.Equal(t, "ready", authority.Status)
152 require.Equal(t, "authority", authority.Reason)
153 require.Equal(t, []string{}, authority.Pending)
154 require.Equal(t, "local-authority", authority.Require)
155 require.NotEqual(t, "0", authority.Epoch, "a live broker reports its own epoch")
156 require.NotEmpty(t, strings.TrimSpace(stdout))
157
158 stdout, stderr, code = sandbox.ready("5s", "--require", "local-catalogue")
159 require.Equal(t, 3, code, "stderr: %s", stderr)
160 catalogue := decodeReadyReport(t, stdout)
161 require.Equal(t, "timeout", catalogue.Status)
162 require.Equal(t, []string{"local-catalogue"}, catalogue.Pending)
163 require.NotEqual(t, "ready", catalogue.Status, "an unobserved catalogue is never reported ready")
164
165}
166
167// TestBrokerReadySubprocessStaleSocketIsNeverReady proves a socket file left by
168// a dead owner is retried into the deadline: the probe reports timeout, never a
169// false ready, and never leaves a second socket behind.
170func TestBrokerReadySubprocessStaleSocketIsNeverReady(t *testing.T) {
171 sandbox := newReadySubprocessSandbox(t)
172 require.NoError(t, os.MkdirAll(sandbox.layout.Runtime, 0o700))
173 bindStaleUnixSocket(t, sandbox.socketPath())
174
175 start := time.Now()
176 stdout, _, code := sandbox.ready("500ms", "--require", "local-authority")
177 require.Equal(t, 3, code)
178 report := decodeReadyReport(t, stdout)
179 require.Equal(t, "timeout", report.Status)
180 require.Equal(t, "deadline", report.Reason)
181 require.Less(t, time.Since(start), 10*time.Second, "the probe is bounded by its own timeout")
182
183 // A stale socket is recoverable absence, so nothing was started over it: the
184 // path is still the same dead socket file and no helper process ran.
185 info, err := os.Lstat(sandbox.socketPath())
186 require.NoError(t, err)
187 require.NotZero(t, info.Mode()&os.ModeSocket)
188 require.Empty(t, recordedPIDsIn(t, sandbox.recordDir, productionBrokerServeCommand))
189}
190
191// TestBrokerReadySubprocessBlockedHandshakeIsNeverReady proves a live endpoint
192// that never answers the broker preamble is never reported ready: the probe
193// stays inside its own bound and reports timeout.
194func TestBrokerReadySubprocessBlockedHandshakeIsNeverReady(t *testing.T) {
195 sandbox := newReadySubprocessSandbox(t)
196 require.NoError(t, os.MkdirAll(sandbox.layout.Runtime, 0o700))
197 // A listener that accepts and never answers: the broker preamble stalls.
198 _ = bindUnixSocket(t, sandbox.socketPath())
199
200 start := time.Now()
201 stdout, _, code := sandbox.ready("700ms", "--require", "local-authority")
202 require.Equal(t, 3, code)
203 report := decodeReadyReport(t, stdout)
204 require.Equal(t, "timeout", report.Status)
205 require.NotEqual(t, "ready", report.Status, "a stalled handshake is never a false ready")
206 require.Less(t, time.Since(start), 10*time.Second)
207
208 // A stalled-but-live endpoint is never respawned over.
209 require.Empty(t, recordedPIDsIn(t, sandbox.recordDir, productionBrokerServeCommand))
210}
211
212// TestBrokerReadySubprocessNonSocketPathIsTerminal proves a path that exists but
213// is not a Unix socket is a terminal security failure (exit 4) rather than a
214// retryable absence, for both a regular file and a directory.
215func TestBrokerReadySubprocessNonSocketPathIsTerminal(t *testing.T) {
216 tests := []struct {
217 name string
218 place func(t *testing.T, socketPath string)
219 }{
220 {
221 name: "regular file",
222 place: func(t *testing.T, socketPath string) {
223 require.NoError(t, os.WriteFile(socketPath, []byte("not a socket"), 0o600))
224 },
225 },
226 {
227 name: "directory",
228 place: func(t *testing.T, socketPath string) {
229 require.NoError(t, os.MkdirAll(socketPath, 0o700))
230 },
231 },
232 }
233 for _, tt := range tests {
234 t.Run(tt.name, func(t *testing.T) {
235 sandbox := newReadySubprocessSandbox(t)
236 require.NoError(t, os.MkdirAll(sandbox.layout.Runtime, 0o700))
237 tt.place(t, sandbox.socketPath())
238
239 stdout, stderr, code := sandbox.ready("500ms", "--require", "local-authority")
240 require.Equal(t, 4, code, "stderr: %s", stderr)
241 report := decodeReadyReport(t, stdout)
242 require.Equal(t, "terminal", report.Status)
243 require.Equal(t, "security", report.Reason)
244 require.Empty(t, recordedPIDsIn(t, sandbox.recordDir, productionBrokerServeCommand))
245 })
246 }
247}
248
249// TestBrokerReadySubprocessPermissionRefusalIsTerminal proves a path this
250// process may not inspect is a terminal security failure with exit 4, and never
251// a timeout or an absence that would invite a spawn.
252func TestBrokerReadySubprocessPermissionRefusalIsTerminal(t *testing.T) {
253 if os.Geteuid() == 0 {
254 t.Skip("an unreadable directory does not deny root")
255 }
256 sandbox := newReadySubprocessSandbox(t)
257 require.NoError(t, os.MkdirAll(sandbox.layout.Runtime, 0o700))
258 require.NoError(t, os.Chmod(sandbox.layout.Runtime, 0o000))
259 t.Cleanup(func() { _ = os.Chmod(sandbox.layout.Runtime, 0o700) })
260
261 stdout, stderr, code := sandbox.ready("500ms", "--require", "local-authority")
262 require.Equal(t, 4, code, "stderr: %s", stderr)
263 report := decodeReadyReport(t, stdout)
264 require.Equal(t, "terminal", report.Status)
265 require.Equal(t, "security", report.Reason)
266 require.Empty(t, recordedPIDsIn(t, sandbox.recordDir, productionBrokerServeCommand))
267}
268
269// TestBrokerReadySubprocessAbsentSocketTimesOut proves a missing endpoint is
270// retried into the deadline and reported as a timeout with no broker spawned.
271func TestBrokerReadySubprocessAbsentSocketTimesOut(t *testing.T) {
272 sandbox := newReadySubprocessSandbox(t)
273 _, statErr := os.Lstat(sandbox.socketPath())
274 require.ErrorIs(t, statErr, os.ErrNotExist)
275
276 stdout, _, code := sandbox.ready("400ms", "--require", "local-authority")
277 require.Equal(t, 3, code)
278 report := decodeReadyReport(t, stdout)
279 require.Equal(t, "timeout", report.Status)
280 require.Equal(t, "deadline", report.Reason)
281 require.Empty(t, recordedPIDsIn(t, sandbox.recordDir, productionBrokerServeCommand), "the readiness probe never ensures or spawns")
282}
283
284// TestBrokerReadySubprocessNonBrokerEndpointIsTerminal proves a live endpoint
285// that answers the broker setup with foreign framing is a terminal protocol
286// failure (exit 4) rather than a retryable absence, so a socket that is not the
287// expected broker is never waited out as if it were merely slow.
288func TestBrokerReadySubprocessNonBrokerEndpointIsTerminal(t *testing.T) {
289 sandbox := newReadySubprocessSandbox(t)
290 require.NoError(t, os.MkdirAll(sandbox.layout.Runtime, 0o700))
291 listener := bindUnixSocket(t, sandbox.socketPath())
292 go answerWithForeignBytes(listener)
293
294 stdout, stderr, code := sandbox.ready("700ms", "--require", "local-authority")
295 require.Equal(t, 4, code, "stderr: %s", stderr)
296 report := decodeReadyReport(t, stdout)
297 require.Equal(t, "terminal", report.Status)
298 require.Equal(t, "protocol", report.Reason)
299 require.Empty(t, recordedPIDsIn(t, sandbox.recordDir, productionBrokerServeCommand))
300}
301
302// answerWithForeignBytes serves one connection per accept with a bounded
303// non-framed answer, standing in for a Unix socket that is not a broker.
304func answerWithForeignBytes(listener net.Listener) {
305 for {
306 conn, err := listener.Accept()
307 if err != nil {
308 return
309 }
310 go func() {
311 defer func() { _ = conn.Close() }()
312 buf := make([]byte, 4096)
313 _ = conn.SetReadDeadline(time.Now().Add(3 * time.Second))
314 _, _ = conn.Read(buf)
315 _, _ = conn.Write([]byte("this endpoint is not a vev broker at all........."))
316 }()
317 }
318}
319
320// TestBrokerReadySubprocessArgumentRefusalIsUsage proves the production dispatch
321// path refuses an invalid invocation with the usage exit code and prints no
322// report at all.
323func TestBrokerReadySubprocessArgumentRefusalIsUsage(t *testing.T) {
324 sandbox := newReadySubprocessSandbox(t)
325
326 stdout, stderr, code := sandbox.ready("1s", "--require", "remote-catalogue")
327 require.Equal(t, 2, code, "stderr: %s", stderr)
328 require.Empty(t, strings.TrimSpace(stdout), "an argument refusal prints no report")
329 require.NotEmpty(t, strings.TrimSpace(stderr))
330}