Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566"""Retiring an unadopted spare must not log an invalid-adoption traceback.
The daemon discards unused spares by closing their private adoption channel.E2E teardown does not expose the launcher's inherited stderr or exit status."""
import jsonimport osfrom pathlib import Pathimport socketimport subprocessimport sysimport tempfileimport unittest
PYTHON = Path(__file__).resolve().parents[2] / "priv/python"
class SpareProtocolTest(unittest.TestCase): def test_unused_spare_exits_cleanly_on_eof(self): result = subprocess.run( [sys.executable, str(PYTHON / "albedo_spare.py")], input=b"", capture_output=True, timeout=30, ) self.assertIn("ready", json.loads(result.stdout)) self.assertEqual(result.returncode, 0, result.stderr.decode()) self.assertEqual(result.stderr, b"")
def test_partial_adoption_is_still_an_error(self): result = subprocess.run( [sys.executable, str(PYTHON / "albedo_spare.py")], input=b"{", capture_output=True, timeout=30, ) self.assertNotEqual(result.returncode, 0) self.assertIn(b"invalid spare adoption frame", result.stderr)
@unittest.skipUnless(hasattr(os, "fork"), "requires fork") def test_unused_forked_spare_exits_cleanly_on_eof(self): with tempfile.TemporaryDirectory(prefix="spare-") as directory: path = str(Path(directory) / "zygote.sock") with subprocess.Popen( [sys.executable, str(PYTHON / "albedo_zygote.py"), path], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) as zygote: try: assert zygote.stdout is not None self.assertIn("ready", json.loads(zygote.stdout.readline())) with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as channel: channel.settimeout(30) channel.connect(path) with channel.makefile("rb") as reader: self.assertIn("ready", json.loads(reader.readline())) channel.shutdown(socket.SHUT_WR) self.assertEqual(reader.read(), b"") _, errors = zygote.communicate(timeout=30) self.assertEqual(zygote.returncode, 0, errors.decode()) self.assertEqual(errors, b"") finally: zygote.kill()