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.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667"""Kernel and stop-helper names remain visible without changing launch behavior."""
import astimport subprocessimport unittest
from harness import Albedo, Provider, exclusive, python, textfrom kernel_support import ModelCells, process_name
@exclusiveclass ProcessNameTests(ModelCells, unittest.TestCase): def boot(self, prepare): self.cells = []
def script(request): if request["input"][-1].get("role") == "user" and self.cells: return python(self.cells.pop(0)) return text("ok")
self.provider = Provider(script) self.addCleanup(self.provider.close) self.app = Albedo(self.provider, protocol="responses", prepare=prepare) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.session = self.app.session()
def kernel_pid(self): result, _ = self.cell("hold = run('sleep', '120')\nos.getpid()") self.assertEqual(result["status"], "ok", result) return ast.literal_eval(result["value"])
def test_kernel_and_signal_helper_have_role_names(self): def prepare(app): directory = app.home / "bin" directory.mkdir() for role in ("kernel", "signal"): (directory / ("albedo-" + role)).symlink_to("/missing-python")
self.boot(prepare) pid = self.kernel_pid() self.assertEqual(process_name(pid), "albedo-kernel") print("kernel comm:", process_name(pid)) resource = f"/sessions/{self.session}?view=configuration" with self.app.api(resource) as response: validator = response.getheader("ETag") response.read() self.app.api(resource, method="DELETE", headers={"If-Match": validator}).close() listing = subprocess.check_output(["ps", "-axo", "pid=,command="], text=True) helpers = [ int(line.split(None, 1)[0]) for line in listing.splitlines() if str(self.app.home / "bin" / "albedo-signal") in line and "albedo_signal.py" in line ] self.assertEqual(len(helpers), 1, listing) self.assertEqual(process_name(helpers[0]), "albedo-signal") print("signal comm:", process_name(helpers[0])) self.assertTrue((self.app.home / "bin" / "albedo-kernel").is_symlink())
def test_unwritable_alias_directory_keeps_kernel_launch_working(self): def prepare(app): (app.home / "bin").write_text("not a directory")
self.boot(prepare) self.assertGreater(self.kernel_pid(), 1)