okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffaufrefactor(test): share inspection and stream supportcba5ff8

main
branchesmainall refs →
15.4 KiB393 linesraw
1"""CLI attachment and startup must preserve an independently owned real daemon."""
2
3import json
4import os
5import pty
6import select
7import signal
8import socket
9import stat
10import subprocess
11import sys
12import termios
13import time
14import unittest
15
16from inspect_support import Inspection
17from harness import Albedo, Provider, exclusive, operation_id, python, text
18
19
20def direct_start(daemon):
21    daemon.start_launcher()
22    daemon.await_ready()
23
24
25def runtime_limits(app):
26    expression = (
27        "Limits = [Call(erlang, system_info, [Key]) "
28        "|| Key <- [process_limit, port_limit, schedulers_online]], "
29        "io:put_chars(json:encode(Limits))"
30    )
31    return json.loads(Inspection(app).evaluate(expression, timeout=20))
32
33
34def gated_launcher(app):
35    """A pipe barrier signals the real launch attempt before letting it exec."""
36    original = app.env["ALBEDO_DAEMON"]
37    started, release = app.root / "launcher-started", app.root / "launcher-release"
38    os.mkfifo(started)
39    os.mkfifo(release)
40    attempts = app.root / "launcher-attempts"
41    wrapper = app.root / "gated-launcher"
42    wrapper.write_text(
43        f"#!{sys.executable}\n"
44        "import os\n"
45        f"with open({str(attempts)!r}, 'a') as attempts:\n"
46        "    attempts.write(str(os.getpid()) + '\\n')\n"
47        "    attempts.flush()\n"
48        f"with open({str(started)!r}, 'w') as ready:\n"
49        "    ready.write(str(os.getpid()) + '\\n')\n"
50        f"with open({str(release)!r}, 'rb', buffering=0) as gate:\n"
51        "    gate.read(1)\n"
52        f"os.execv({original!r}, [{original!r}])\n"
53    )
54    wrapper.chmod(0o700)
55    app.env["ALBEDO_DAEMON"] = str(wrapper)
56    return started, release, attempts
57
58
59class TerminalCommand:
60    def __init__(self, app, *arguments):
61        self.master, slave = pty.openpty()
62        termios.tcsetwinsize(slave, (24, 100))
63        self.process = app.daemon.start_cli(
64            *arguments, stdin=slave, stdout=slave, stderr=slave
65        )
66        os.close(slave)
67        self.output = b""
68
69    def prompt(self, timeout=10):
70        deadline = time.monotonic() + timeout
71        while time.monotonic() < deadline:
72            readable, _, _ = select.select(
73                [self.master], [], [], max(0, deadline - time.monotonic())
74            )
75            if readable:
76                try:
77                    chunk = os.read(self.master, 65536)
78                except OSError as error:
79                    raise AssertionError(
80                        "CLI exited before restart prompt: "
81                        + self.output.decode(errors="replace")
82                    ) from error
83                self.output += chunk
84                if b"\x1b[6n" in chunk:
85                    os.write(self.master, b"\x1b[1;1R")
86                if b"\x1b[c" in chunk:
87                    os.write(self.master, b"\x1b[?1;2c")
88                if b"[y/N]" in self.output or b"[Y/n]" in self.output:
89                    return
90        raise AssertionError(
91            "CLI did not offer a restart: " + self.output.decode(errors="replace")
92        )
93
94    def answer(self, value):
95        os.write(self.master, (value + "\r").encode())
96        result = self.process.wait(timeout=20)
97        os.close(self.master)
98        return result
99
100
101# exclusive: owns daemon startup and restart behavior
102@exclusive
103class LifecycleTest(unittest.TestCase):
104    def test_standalone_default_home_authentication_and_vm_defaults_match_cli(self):
105        def prepare(app):
106            app.home = app.daemon.home = app.root / "user-home" / ".albedo"
107            app.home.mkdir(mode=0o755)
108            app.env.pop("ALBEDO_HOME")
109            app.env.pop("ALBEDO_TOKEN", None)
110            app.env["ALBEDO_INSPECT"] = "1"
111            app.daemon.boot_action = direct_start
112
113        with Albedo(prepare=prepare) as app:
114            direct = dict(app.connection)
115            self.assertGreaterEqual(len(direct["token"]), 32)
116            self.assertEqual(stat.S_IMODE(app.home.stat().st_mode), 0o700)
117            self.assertEqual(
118                stat.S_IMODE((app.home / "daemon.json").stat().st_mode), 0o600
119            )
120            connection = socket.create_connection(
121                ("127.0.0.1", direct["port"]), timeout=5
122            )
123            with connection:
124                connection.sendall(
125                    f"GET /server HTTP/1.1\r\nHost: localhost:{direct['port']}\r\nConnection: close\r\n\r\n".encode()
126                )
127                self.assertIn(b"401", connection.recv(1024).split(b"\r\n", 1)[0])
128            direct_limits = runtime_limits(app)
129            session = app.session()
130            app.prompt(session, "independent daemon is usable").close()
131            app.idle(session)
132            self.assertIn(
133                "ok",
134                [
135                    part["text"]
136                    for entry in app.history(session)["items"]
137                    for part in entry["content"]
138                    if part["kind"] == "text"
139                ],
140            )
141            self.assertEqual(app.connection, direct)
142            app.restart()
143            self.assertNotEqual(app.connection["pid"], direct["pid"])
144            self.assertNotEqual(app.connection["token"], direct["token"])
145            self.assertEqual(runtime_limits(app), direct_limits)
146            app.restart(
147                prepare=lambda fixture: fixture.env.update(
148                    ERL_FLAGS=fixture.env["ERL_FLAGS"] + " +P 131072 +Q 32768",
149                    ALBEDO_TOKEN="operator-chosen-authentication-token",
150                )
151            )
152            self.assertEqual(runtime_limits(app)[:2], [131072, 32768])
153            self.assertEqual(
154                app.connection["token"], "operator-chosen-authentication-token"
155            )
156            self.assertEqual(app.api("/server").status, 200)
157
158    def test_cancelling_cli_startup_wait_leaves_the_started_daemon_owned(self):
159        captured = {}
160
161        def prepare(app):
162            started, release, attempts = gated_launcher(app)
163
164            def start(daemon):
165                command = daemon.start_cli(
166                    "daemon", stdout=subprocess.PIPE, stderr=subprocess.PIPE
167                )
168                with started.open() as ready:
169                    captured["child"] = int(ready.readline())
170                adopted = False
171                try:
172                    command.send_signal(signal.SIGINT)
173                    stdout, stderr = command.communicate(timeout=5)
174                    captured["cancelled"] = command.returncode
175                    captured["output"] = stdout + stderr
176                    with release.open("wb", buffering=0) as gate:
177                        gate.write(b"x")
178                    daemon.await_ready()
179                    adopted = True
180                finally:
181                    if not adopted:
182                        try:
183                            os.kill(captured["child"], signal.SIGKILL)
184                        except ProcessLookupError:
185                            pass
186                captured["attempts"] = attempts.read_text().splitlines()
187
188            app.daemon.boot_action = start
189
190        with Albedo(prepare=prepare) as app:
191            self.assertNotEqual(captured["cancelled"], 0, captured["output"])
192            self.assertEqual(app.connection["pid"], captured["child"])
193            self.assertEqual(len(captured["attempts"]), 1)
194            session = app.session()
195            app.prompt(session, "daemon survived waiter cancellation").close()
196            app.idle(session)
197            self.assertIn(
198                "ok",
199                [
200                    part["text"]
201                    for entry in app.history(session)["items"]
202                    for part in entry["content"]
203                    if part["kind"] == "text"
204                ],
205            )
206
207    def test_simultaneous_cli_starters_converge_on_one_real_execution(self):
208        captured = {}
209
210        def prepare(app):
211            started, release, attempts = gated_launcher(app)
212
213            def start(daemon):
214                first = daemon.start_cli(
215                    "daemon", stdout=subprocess.PIPE, stderr=subprocess.PIPE
216                )
217                second = daemon.start_cli(
218                    "daemon", stdout=subprocess.PIPE, stderr=subprocess.PIPE
219                )
220                with started.open() as ready:
221                    captured["owner"] = int(ready.readline())
222                adopted = False
223                try:
224                    with release.open("wb", buffering=0) as gate:
225                        gate.write(b"x")
226                    captured["outputs"] = [
227                        command.communicate(timeout=20) for command in (first, second)
228                    ]
229                    captured["codes"] = [first.returncode, second.returncode]
230                    daemon.await_ready()
231                    adopted = True
232                finally:
233                    if not adopted:
234                        try:
235                            os.kill(captured["owner"], signal.SIGKILL)
236                        except ProcessLookupError:
237                            pass
238                captured["attempts"] = attempts.read_text().splitlines()
239
240            app.daemon.boot_action = start
241
242        with Albedo(prepare=prepare) as app:
243            self.assertEqual(captured["codes"], [0, 0], captured["outputs"])
244            self.assertEqual(captured["attempts"], [str(app.connection["pid"])])
245            self.assertEqual(app.connection["pid"], captured["owner"])
246            self.assertEqual(app.api("/sessions").status, 200)
247
248    def test_compatible_other_build_is_kept_without_a_terminal_and_explicit_yes_restarts(
249        self,
250    ):
251        def prepare(app):
252            app.env["ALBEDO_BUILD"] = "independent-compatible-build"
253            app.daemon.boot_action = direct_start
254
255        with Albedo(prepare=prepare) as app:
256            previous = dict(app.connection)
257            self.assertEqual(previous["build"], "independent-compatible-build")
258            # The daemon records the content digest of the code it runs, so a
259            # client can prove a build difference without a label.
260            self.assertRegex(previous["digest"], r"^[0-9a-f]{64}$")
261            app.env.pop("ALBEDO_BUILD")
262            app.cli("daemon")
263            self.assertEqual(
264                json.loads((app.home / "daemon.json").read_text()), previous
265            )
266            terminal = TerminalCommand(app, "daemon")
267            terminal.prompt()
268            self.assertEqual(terminal.answer(""), 0)
269            self.assertEqual(
270                json.loads((app.home / "daemon.json").read_text()), previous
271            )
272            for cancel in ("\x03", "\x04"):
273                terminal = TerminalCommand(app, "daemon")
274                terminal.prompt()
275                self.assertNotEqual(terminal.answer(cancel), 0)
276                self.assertEqual(
277                    json.loads((app.home / "daemon.json").read_text()), previous
278                )
279            terminal = TerminalCommand(app, "daemon")
280            terminal.prompt()
281            self.assertEqual(terminal.answer("y"), 0)
282            app.daemon._refresh()
283            self.assertNotEqual(app.connection["pid"], previous["pid"])
284            self.assertNotEqual(app.connection["token"], previous["token"])
285
286    def test_restart_confirmation_cannot_replace_a_newer_daemon(self):
287        with Albedo() as app:
288            terminal = TerminalCommand(app, "daemon")
289            terminal.prompt()
290            app.restart()
291            replacement = dict(app.connection)
292            self.assertNotEqual(terminal.answer("y"), 0)
293            self.assertEqual(
294                json.loads((app.home / "daemon.json").read_text()), replacement
295            )
296            self.assertEqual(app.api("/server").status, 200)
297
298    def test_explicit_upgrade_recovers_history_receipt_and_uncertain_tool_without_repeating_effect(
299        self,
300    ):
301        fixture = {}
302
303        def script(request):
304            messages = request["messages"]
305            if any(
306                "daemon restart" in str(message.get("content", ""))
307                for message in messages
308            ):
309                return text("recovered after upgrade")
310            if messages[-1]["role"] == "tool":
311                return text("tool finished")
312            return python(
313                "from pathlib import Path\nimport threading\n"
314                f"with Path({str(fixture['effect'])!r}).open('a') as effect:\n"
315                "    effect.write('effect applied\\n')\n"
316                "    effect.flush()\n"
317                "threading.Event().wait(60)"
318            )
319
320        provider = Provider(script)
321        self.addCleanup(provider.close)
322        with Albedo(provider) as app:
323            fixture["effect"] = app.workspace / "effect"
324            session = app.session()
325            operation = operation_id()
326            app.api(
327                f"/sessions/{session}/inputs/{operation}",
328                {"kind": "message", "text": "apply an effect then wait"},
329                method="PUT",
330            ).close()
331            deadline = time.monotonic() + 20
332            while (
333                not fixture["effect"].exists()
334                or fixture["effect"].read_text() != "effect applied\n"
335            ):
336                if time.monotonic() > deadline:
337                    app._fail("tool never published its external effect")
338                time.sleep(0.02)
339            self.assertTrue(
340                json.load(app.api(f"/sessions/{session}?tail=0"))["status"]["phase"]
341                == "running"
342            )
343            previous = dict(app.connection)
344            terminal = TerminalCommand(app, "daemon")
345            terminal.prompt()
346            self.assertEqual(terminal.answer("y"), 0)
347            app.daemon._refresh()
348            self.assertNotEqual(app.connection["pid"], previous["pid"])
349            app.idle(session)
350            receipt = json.load(app.api(f"/sessions/{session}/inputs/{operation}"))
351            self.assertEqual(receipt["delivery"], "committed")
352            self.assertEqual(receipt["turn"]["state"], "abandoned")
353            entries = app.history(session)["items"]
354            self.assertEqual(
355                sum(entry["input_id"] == operation for entry in entries), 1
356            )
357            self.assertTrue(
358                any(
359                    entry["kind"] == "note"
360                    and any(
361                        part["kind"] == "text" and "restart" in part["text"]
362                        for part in entry["content"]
363                    )
364                    for entry in entries
365                )
366            )
367            self.assertTrue(
368                any(
369                    part["kind"] == "text" and part["text"] == "recovered after upgrade"
370                    for entry in entries
371                    for part in entry["content"]
372                )
373            )
374            self.assertEqual(fixture["effect"].read_text(), "effect applied\n")
375
376    def test_failed_discovery_publication_exits_and_releases_home_for_retry(self):
377        captured = {}
378
379        def prepare(app):
380            def start(daemon):
381                blocked = app.home / "daemon.json"
382                blocked.mkdir()
383                process = daemon.start_launcher()
384                captured["exit"] = process.wait(timeout=20)
385                captured["pid"] = process.pid
386                blocked.rmdir()
387
388            app.daemon.boot_action = start
389
390        with Albedo(prepare=prepare) as app:
391            self.assertNotEqual(captured["exit"], 0)
392            self.assertNotEqual(app.connection["pid"], captured["pid"])
393            self.assertEqual(app.api("/server").status, 200)