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"""Bridge startup reports a dead kernel at once, with its status and log.
E2E only sees a remote session degrade to ssh command mode; it cannot tell akernel that died from one the bridge gave up on, which is the bug that made aslow but live remote host fail at a fixed 10 s. Nor can it order a kernel'sexit before the bridge hands it the token, which once crashed the bridge witha broken pipe instead of the kernel's reason."""
import contextlibimport ioimport subprocessimport tempfileimport timeimport unittestfrom unittest import mockfrom pathlib import Path
import sys
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "priv" / "python"))import albedo_bridge # noqa: E402
class BridgeStartupTest(unittest.TestCase): def start_dying_kernel(self): with tempfile.TemporaryDirectory() as directory: kernel = Path(directory) / "kernel.py" kernel.write_text("raise SystemExit(17)\n") run_dir = Path(directory) / "run" old_kernel, old_timeout = albedo_bridge.KERNEL, albedo_bridge.START_TIMEOUT albedo_bridge.KERNEL = str(kernel) albedo_bridge.START_TIMEOUT = 120.0 error = io.StringIO() began = time.monotonic() try: with contextlib.redirect_stderr(error): with self.assertRaises(SystemExit) as raised: albedo_bridge.start(str(run_dir), "[]", "token") finally: albedo_bridge.KERNEL, albedo_bridge.START_TIMEOUT = ( old_kernel, old_timeout, ) self.assertEqual(raised.exception.code, albedo_bridge.GONE) self.assertLess(time.monotonic() - began, 2.0) self.assertIn("status 17", error.getvalue())
def test_kernel_exits_before_binding_fails_with_its_reason(self): self.start_dying_kernel()
def test_kernel_gone_before_its_token_fails_with_its_reason(self): popen = subprocess.Popen
def exited(*args, **kwargs): kernel = popen(*args, **kwargs) kernel.wait() return kernel
with mock.patch.object(albedo_bridge.subprocess, "Popen", exited): self.start_dying_kernel()
if __name__ == "__main__": unittest.main()