"""CLI attachment and startup must preserve an independently owned real daemon."""
import json
import os
import pty
import select
import signal
import socket
import stat
import subprocess
import sys
import termios
import time
import unittest
from harness import Albedo, Provider, exclusive, operation_id, python, text
def direct_start(daemon):
daemon.start_launcher()
daemon.await_ready()
def runtime_limits(app):
cookie = (app.home / "inspect.cookie").read_text().strip()
node = f"albedo_{app.daemon._pid}@{socket.gethostname().split('.')[0]}"
expression = (
f"Node = '{node}', "
"Limits = [rpc:call(Node, erlang, system_info, [Key]) "
"|| Key <- [process_limit, port_limit, schedulers_online]], "
"io:put_chars(json:encode(Limits)), halt()."
)
result = subprocess.run(
[
"erl",
"+S",
"2:2",
"-sname",
f"lifecycle_probe_{app.daemon._pid}",
"-setcookie",
cookie,
"-noshell",
"-eval",
expression,
],
capture_output=True,
text=True,
timeout=20,
check=True,
)
return json.loads(result.stdout)
def gated_launcher(app):
"""A pipe barrier signals the real launch attempt before letting it exec."""
original = app.env["ALBEDO_DAEMON"]
started, release = app.root / "launcher-started", app.root / "launcher-release"
os.mkfifo(started)
os.mkfifo(release)
attempts = app.root / "launcher-attempts"
wrapper = app.root / "gated-launcher"
wrapper.write_text(
f"#!{sys.executable}\n"
"import os\n"
f"with open({str(attempts)!r}, 'a') as attempts:\n"
" attempts.write(str(os.getpid()) + '\\n')\n"
" attempts.flush()\n"
f"with open({str(started)!r}, 'w') as ready:\n"
" ready.write(str(os.getpid()) + '\\n')\n"
f"with open({str(release)!r}, 'rb', buffering=0) as gate:\n"
" gate.read(1)\n"
f"os.execv({original!r}, [{original!r}])\n"
)
wrapper.chmod(0o700)
app.env["ALBEDO_DAEMON"] = str(wrapper)
return started, release, attempts
class TerminalCommand:
def __init__(self, app, *arguments):
self.master, slave = pty.openpty()
termios.tcsetwinsize(slave, (24, 100))
self.process = app.daemon.start_cli(
*arguments, stdin=slave, stdout=slave, stderr=slave
)
os.close(slave)
self.output = b""
def prompt(self, timeout=10):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
readable, _, _ = select.select(
[self.master], [], [], max(0, deadline - time.monotonic())
)
if readable:
try:
chunk = os.read(self.master, 65536)
except OSError as error:
raise AssertionError(
"CLI exited before restart prompt: "
+ self.output.decode(errors="replace")
) from error
self.output += chunk
if b"\x1b[6n" in chunk:
os.write(self.master, b"\x1b[1;1R")
if b"\x1b[c" in chunk:
os.write(self.master, b"\x1b[?1;2c")
if b"[y/N]" in self.output or b"[Y/n]" in self.output:
return
raise AssertionError(
"CLI did not offer a restart: " + self.output.decode(errors="replace")
)
def answer(self, value):
os.write(self.master, (value + "\r").encode())
result = self.process.wait(timeout=20)
os.close(self.master)
return result
@exclusive
class LifecycleTest(unittest.TestCase):
def test_standalone_default_home_authentication_and_vm_defaults_match_cli(self):
def prepare(app):
app.home = app.daemon.home = app.root / "user-home" / ".albedo"
app.home.mkdir(mode=0o755)
app.env.pop("ALBEDO_HOME")
app.env.pop("ALBEDO_TOKEN", None)
app.env["ALBEDO_INSPECT"] = "1"
app.daemon.boot_action = direct_start
with Albedo(prepare=prepare) as app:
direct = dict(app.connection)
self.assertGreaterEqual(len(direct["token"]), 32)
self.assertEqual(stat.S_IMODE(app.home.stat().st_mode), 0o700)
self.assertEqual(
stat.S_IMODE((app.home / "daemon.json").stat().st_mode), 0o600
)
connection = socket.create_connection(
("127.0.0.1", direct["port"]), timeout=5
)
with connection:
connection.sendall(
b"GET /health HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"
)
self.assertIn(b"403", connection.recv(1024).split(b"\r\n", 1)[0])
direct_limits = runtime_limits(app)
session = app.session()
app.prompt(session, "independent daemon is usable").close()
app.idle(session)
self.assertIn("ok", [event.get("text") for event in app.events(session)])
self.assertEqual(app.connection, direct)
app.restart()
self.assertNotEqual(app.connection["pid"], direct["pid"])
self.assertNotEqual(app.connection["token"], direct["token"])
self.assertEqual(runtime_limits(app), direct_limits)
app.restart(
prepare=lambda fixture: fixture.env.update(
ERL_FLAGS=fixture.env["ERL_FLAGS"] + " +P 131072 +Q 32768",
ALBEDO_TOKEN="operator-chosen-authentication-token",
)
)
self.assertEqual(runtime_limits(app)[:2], [131072, 32768])
self.assertEqual(
app.connection["token"], "operator-chosen-authentication-token"
)
self.assertEqual(app.api("/health").status, 200)
def test_cancelling_cli_startup_wait_leaves_the_started_daemon_owned(self):
captured = {}
def prepare(app):
started, release, attempts = gated_launcher(app)
def start(daemon):
command = daemon.start_cli(
"daemon", stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
with started.open() as ready:
captured["child"] = int(ready.readline())
adopted = False
try:
command.send_signal(signal.SIGINT)
stdout, stderr = command.communicate(timeout=5)
captured["cancelled"] = command.returncode
captured["output"] = stdout + stderr
with release.open("wb", buffering=0) as gate:
gate.write(b"x")
daemon.await_ready()
adopted = True
finally:
if not adopted:
try:
os.kill(captured["child"], signal.SIGKILL)
except ProcessLookupError:
pass
captured["attempts"] = attempts.read_text().splitlines()
app.daemon.boot_action = start
with Albedo(prepare=prepare) as app:
self.assertNotEqual(captured["cancelled"], 0, captured["output"])
self.assertEqual(app.connection["pid"], captured["child"])
self.assertEqual(len(captured["attempts"]), 1)
session = app.session()
app.prompt(session, "daemon survived waiter cancellation").close()
app.idle(session)
self.assertIn("ok", [event.get("text") for event in app.events(session)])
def test_simultaneous_cli_starters_converge_on_one_real_execution(self):
captured = {}
def prepare(app):
started, release, attempts = gated_launcher(app)
def start(daemon):
first = daemon.start_cli(
"daemon", stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
second = daemon.start_cli(
"daemon", stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
with started.open() as ready:
captured["owner"] = int(ready.readline())
adopted = False
try:
with release.open("wb", buffering=0) as gate:
gate.write(b"x")
captured["outputs"] = [
command.communicate(timeout=20) for command in (first, second)
]
captured["codes"] = [first.returncode, second.returncode]
daemon.await_ready()
adopted = True
finally:
if not adopted:
try:
os.kill(captured["owner"], signal.SIGKILL)
except ProcessLookupError:
pass
captured["attempts"] = attempts.read_text().splitlines()
app.daemon.boot_action = start
with Albedo(prepare=prepare) as app:
self.assertEqual(captured["codes"], [0, 0], captured["outputs"])
self.assertEqual(captured["attempts"], [str(app.connection["pid"])])
self.assertEqual(app.connection["pid"], captured["owner"])
self.assertEqual(app.api("/sessions").status, 200)
def test_compatible_other_build_is_kept_without_a_terminal_and_explicit_yes_restarts(
self,
):
def prepare(app):
app.env["ALBEDO_BUILD"] = "independent-compatible-build"
app.daemon.boot_action = direct_start
with Albedo(prepare=prepare) as app:
previous = dict(app.connection)
self.assertEqual(previous["build"], "independent-compatible-build")
app.env.pop("ALBEDO_BUILD")
app.cli("daemon")
self.assertEqual(
json.loads((app.home / "daemon.json").read_text()), previous
)
terminal = TerminalCommand(app, "daemon")
terminal.prompt()
self.assertEqual(terminal.answer(""), 0)
self.assertEqual(
json.loads((app.home / "daemon.json").read_text()), previous
)
for cancel in ("\x03", "\x04"):
terminal = TerminalCommand(app, "daemon")
terminal.prompt()
self.assertNotEqual(terminal.answer(cancel), 0)
self.assertEqual(
json.loads((app.home / "daemon.json").read_text()), previous
)
terminal = TerminalCommand(app, "daemon")
terminal.prompt()
self.assertEqual(terminal.answer("y"), 0)
app.daemon._refresh()
self.assertNotEqual(app.connection["pid"], previous["pid"])
self.assertNotEqual(app.connection["token"], previous["token"])
def test_restart_confirmation_cannot_replace_a_newer_daemon(self):
with Albedo() as app:
terminal = TerminalCommand(app, "daemon")
terminal.prompt()
app.restart()
replacement = dict(app.connection)
self.assertNotEqual(terminal.answer("y"), 0)
self.assertEqual(
json.loads((app.home / "daemon.json").read_text()), replacement
)
self.assertEqual(app.api("/health").status, 200)
def test_explicit_upgrade_recovers_history_receipt_and_uncertain_tool_without_repeating_effect(
self,
):
fixture = {}
def script(request):
messages = request["messages"]
if any(
"daemon restart" in str(message.get("content", ""))
for message in messages
):
return text("recovered after upgrade")
if messages[-1]["role"] == "tool":
return text("tool finished")
return python(
"from pathlib import Path\nimport time\n"
f"with Path({str(fixture['effect'])!r}).open('a') as effect:\n"
" effect.write('effect applied\\n')\n"
" effect.flush()\n"
"time.sleep(60)"
)
provider = Provider(script)
self.addCleanup(provider.close)
with Albedo(provider) as app:
fixture["effect"] = app.workspace / "effect"
session = app.session()
operation = operation_id()
app.api(
f"/sessions/{session}/events",
{"operationId": operation, "content": "apply an effect then wait"},
).close()
deadline = time.monotonic() + 20
while (
not fixture["effect"].exists()
or fixture["effect"].read_text() != "effect applied\n"
):
if time.monotonic() > deadline:
app._fail("tool never published its external effect")
time.sleep(0.02)
self.assertTrue(
json.load(app.api(f"/sessions/{session}/status"))["running"]
)
previous = dict(app.connection)
terminal = TerminalCommand(app, "daemon")
terminal.prompt()
self.assertEqual(terminal.answer("y"), 0)
app.daemon._refresh()
self.assertNotEqual(app.connection["pid"], previous["pid"])
app.idle(session)
receipt = json.load(app.api(f"/operations/{operation}"))
self.assertEqual(receipt["deliveryStatus"], "committed")
events = app.events(session)
self.assertEqual(
len(
[event for event in events if event.get("operationId") == operation]
),
1,
)
self.assertTrue(
any(event.get("source") == "daemon restart" for event in events)
)
self.assertTrue(
any(event.get("text") == "recovered after upgrade" for event in events)
)
self.assertEqual(fixture["effect"].read_text(), "effect applied\n")
def test_failed_discovery_publication_exits_and_releases_home_for_retry(self):
captured = {}
def prepare(app):
def start(daemon):
blocked = app.home / "daemon.json"
blocked.mkdir()
process = daemon.start_launcher()
captured["exit"] = process.wait(timeout=20)
captured["pid"] = process.pid
blocked.rmdir()
app.daemon.boot_action = start
with Albedo(prepare=prepare) as app:
self.assertNotEqual(captured["exit"], 0)
self.assertNotEqual(app.connection["pid"], captured["pid"])
self.assertEqual(app.api("/health").status, 200)