Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222"""The remote plugin: loopback kernel, references, relay, and degradation."""import asyncioimport contextlibimport ioimport jsonimport osimport sysimport tempfileimport unittestfrom pathlib import Pathfrom unittest.mock import patch
ROOT = Path(__file__).resolve().parents[2]sys.path.insert(0, str(ROOT / "priv" / "python"))from albedo_api import PythonApifrom albedo_plugins import remote
class FakeHostError(Exception): def __init__(self, code, message): super().__init__(message) self.code, self.message = code, message
class FakeCapture: """The retained channel a background job writes into."""
def __init__(self, id): self.id, self.seen, self.tail_data = id, 0, bytearray() self._data = bytearray()
def write(self, text): data = text.encode() self._data.extend(data) self.seen += len(data) self.tail_data.extend(data) del self.tail_data[:-65536]
def fake_ssh_path() -> str: """A loopback ssh: option pairs, target, then run the command locally.""" script = Path(tempfile.mkdtemp(prefix="albedo-fake-ssh-")) / "ssh" script.write_text( "#!/bin/sh\n" "while [ $# -gt 0 ]; do\n" " case \"$1\" in\n" " -*) shift 2 ;;\n" " *) break ;;\n" " esac\n" "done\n" "shift\n" "exec sh -c \"$*\"\n") script.chmod(0o755) return str(script)
FAKE_SSH = fake_ssh_path()
async def daemon_host(method, args): if method == "work.list": return [{"id": 1, "title": "from-the-daemon", "notes": "", "status": "open", "parent": None, "session": None, "run": None, "revision": 1}] if method == "work.get": raise FakeHostError("missing", f"no work item {args['id']}") return {}
class RemotePluginTest(unittest.IsolatedAsyncioTestCase): def setUp(self): remote.configured.clear() remote.connections.clear() self.home = tempfile.mkdtemp(prefix="albedo-remote-home-") self.workspace = tempfile.mkdtemp(prefix="albedo-remote-ws-") os.chdir(self.workspace) self.environment = patch.dict(os.environ, { "HOME": self.home, "ALBEDO_HOME": os.path.join(self.home, ".albedo"), "ALBEDO_SSH": "loopback"}) self.environment.start() self.addCleanup(self.environment.stop) self.addCleanup(os.chdir, ROOT) self.fake_ssh = patch.object(remote, "ssh_base", lambda: [FAKE_SSH]) self.fake_ssh.start() self.addCleanup(self.fake_ssh.stop)
async def asyncSetUp(self): self.loop = asyncio.get_running_loop() self.api = PythonApi(self.loop, daemon_host, FakeHostError, FakeCapture, 64 * 1024, lambda event: None, lambda close: None, lambda cls: None, 2, ["bash", "files", "work", "skills", "remote"]) self.namespace = {} self.remote = remote.setup(self.api)["remote"]
async def connect(self, **kwargs): return await self.remote.connect(**kwargs)
def test_setup_binds_the_namespace_object(self): self.assertIs(self.namespace.get("remote") or self.remote, self.remote)
def test_resolve_names_every_source_when_no_target_exists(self): with patch.dict(os.environ, {"ALBEDO_SSH": ""}): os.environ.pop("ALBEDO_SSH", None) with self.assertRaises(remote.RemoteError) as raised: remote.resolve() self.assertIn("$ALBEDO_SSH", str(raised.exception))
async def test_connect_boots_the_kernel_and_answers_every_tool(self): rem = await self.connect() self.addCleanup(rem.close) self.assertIn("kernel", repr(rem)) self.assertIn("bash", (await rem.tools())["names"])
text = await rem.files.read("/etc/hosts") self.assertIsInstance(text, str)
# The call is synchronous like local bash; the reference settles on await. job = rem.bash("echo loopback-echo; sleep 0.2") self.assertIn("pending", repr(job)) self.assertIsNone(job.poll()) self.assertEqual(job.tail(), "") awaited = await job self.assertIs(awaited, job) self.assertEqual(job.poll(), 0) self.assertEqual(job.exit_code, 0) self.assertEqual(job.returncode, 0) # subprocess's spelling answers too self.assertFalse(job.timed_out) self.assertIsNotNone(job.duration) self.assertGreaterEqual(job.duration, 0.2) self.assertEqual(job.tail(), "loopback-echo\n") self.assertIsNotNone(job.id) self.assertIn("tail", dir(job)) # dir() mixes local and mirrored names ending = await job.stop # awaiting an uncalled method runs it self.assertTrue(ending.gone)
def test_control_sockets_always_live_under_a_short_temp_path(self): self.fake_ssh.stop() try: argv = remote.ssh_base() finally: self.fake_ssh.start() control = [arg for arg in argv if arg.startswith("ControlPath=")][0].split("=", 1)[1] self.assertLessEqual(len(control) + 42, 104) self.assertIn("albedo-ssh-cm", control)
async def test_typed_results_rebuild_as_real_local_objects(self): rem = await self.connect() self.addCleanup(rem.close) Path(self.workspace, "note.txt").write_text("needle one\nplain\nneedle two\n") rows = await rem.files.find("needle", ".") self.assertEqual(type(rows).__name__, "RemoteRef") self.assertEqual(type(rows._value).__name__, "Rows") self.assertEqual(type(rows[0]).__name__, "Match") self.assertTrue(repr(rows).startswith("./note.txt:1:")) self.assertEqual(rows[0].to_dict()["line"], 1) # local method, no round trip
async def test_session_tools_relay_to_this_daemon(self): rem = await self.connect() self.addCleanup(rem.close) items = await rem.work.list() self.assertEqual(items[0]["title"], "from-the-daemon") with self.assertRaises(remote.RemoteExecutionError) as raised: await rem.work.get(99) self.assertEqual(raised.exception.ename, "WorkError")
async def test_ssh_conveniences_work_over_the_control_connection(self): rem = await self.connect() self.addCleanup(rem.close) await rem.write("greeting.txt", "hola") self.assertEqual(await rem.read("greeting.txt"), "hola")
async def test_references_passed_back_stay_references(self): rem = await self.connect() self.addCleanup(rem.close) job = await rem.bash("echo roundtrip") encoded = remote._encode_arg(job) self.assertEqual(encoded, {"__ref__": job._handle}) # a call still in flight passes its pending identity, not a copy racing = rem.bash("echo still-pending") self.assertEqual(remote._encode_arg(racing), {"pending": racing._call_id}) await racing
async def test_a_kernel_that_cannot_boot_degrades_with_a_warning(self): warning = io.StringIO() with contextlib.redirect_stdout(warning): rem = await self.connect(modules=["definitely-not-a-module"]) self.addCleanup(rem.close) self.assertIn("degraded", repr(rem)) self.assertIn("only rem.bash() is available", warning.getvalue()) job = rem.bash("echo degraded-echo") await job self.assertEqual(job.poll(), 0) self.assertEqual(job.returncode, 0) # subprocess's spelling answers too self.assertIsNotNone(job.duration) self.assertFalse(job.timed_out) self.assertEqual(job.tail(), "degraded-echo\n") with self.assertRaises(AttributeError) as raised: rem.files self.assertIn("degraded to ssh command mode", str(raised.exception)) tools = await rem.tools() self.assertIn("degraded", tools)
async def test_an_unreachable_host_fails_connect_without_degrading(self): async def unreachable(target, script, *, timeout, stdin=None): return 255, b"", "ssh: connect to host nosuchhost: no route to host"
with patch.object(remote, "ssh_run", unreachable): with self.assertRaises(remote.RemoteError) as raised: await self.connect(host="nosuchhost") self.assertIn("unreachable", str(raised.exception))
async def test_close_all_ends_every_connection(self): rem = await self.connect() await remote.Remote.close_all() self.assertTrue(rem.closed) self.assertEqual(remote.connections, [])
if __name__ == "__main__": unittest.main()