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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134"""Sessions at remote locations: `[user@]host:/abs/path` workspaces.
A remote location is a stored key the daemon parses and labels withoutconnecting to the host. These cover what a client sees: the canonicalspelling and parsed `location`, rejected forms, and moving between local andremote. What needsthe host itself (kernels, `~`, folders, project files) isremote_kernels_test.py, over a loopback ssh."""
import jsonimport unittestimport urllib.errorimport urllib.parse
from harness import Albedo, exclusive, operation_id
HOST = "albedo-e2e-nowhere"
class LocationsTest(unittest.TestCase): def setUp(self): self.app = Albedo().__enter__() self.addCleanup(self.app.__exit__, None, None, None)
def api(self, path, body=None, **options): with self.app.api(path, body, **options) as response: return json.load(response)
def failure(self, path, body=None, **options): with self.assertRaises(urllib.error.HTTPError) as caught: self.api(path, body, **options) return caught.exception.code, json.load(caught.exception)["detail"]
def create(self, workspace): return self.api( f"/sessions/{operation_id()}", { "kind": "new", "workspace": workspace, "provider_profile": self.app.profile, }, method="PUT", headers={"If-None-Match": "*"}, )
def move(self, session, workspace): snapshot = self.api(f"/sessions/{session}?tail=0") resource = snapshot["configuration_resource"] return self.api( resource["url"], {"workspace": workspace, "family_revision": snapshot["family_revision"]}, method="PATCH", headers={"If-Match": resource["etag"]}, )["session"]
def test_remote_workspace_is_stored_canonically_with_its_location(self): created = self.create(f"mayer@{HOST}:/home/mayer//proj/./albedo/") canonical = f"mayer@{HOST}:/home/mayer/proj/albedo" self.assertEqual(created["workspace"], canonical) location = created["location"] self.assertEqual( (location["host"], location["user"], location["path"]), (HOST, "mayer", "/home/mayer/proj/albedo"), ) query = urllib.parse.urlencode({"workspace": canonical}) listed = next( s for s in self.api("/sessions?" + query)["items"] if s["id"] == created["id"] ) self.assertEqual(listed["workspace"], canonical)
local = self.create(str(self.app.workspace)) self.assertEqual(local["workspace"], str(self.app.workspace)) self.assertEqual( local["location"], { "host": None, "user": None, "path": str(self.app.workspace), "label": None, }, )
# `albedo new` takes the active profile, which other tests on the shared # daemon rewrite; a daemon of its own keeps that profile this fixture's. # exclusive: CLI new depends on the active provider profile other fixtures rewrite @exclusive def test_the_cli_starts_a_remote_session_without_resolving_it_locally(self): session = json.loads(self.app.cli("new", f"{HOST}:/srv/cli"))["session"] listed = next(s for s in self.api("/sessions")["items"] if s["id"] == session) self.assertEqual(listed["workspace"], f"{HOST}:/srv/cli")
def test_unusable_locations_are_rejected(self): for workspace, said in ( (f"{HOST}:proj", "must be absolute"), (f"{HOST}:", "must be absolute"), ("-oProxyCommand=touch:/tmp", "not a valid host"), (f"-l@{HOST}:/srv", "not a valid user"), ("relative/path", "absolute path or host:/absolute/path"), ("dir/with:colon", "absolute path or host:/absolute/path"), ): with self.subTest(workspace=workspace): status, message = self.failure( f"/sessions/{operation_id()}", { "kind": "new", "workspace": workspace, "provider_profile": self.app.profile, }, method="PUT", headers={"If-None-Match": "*"}, ) self.assertEqual(status, 400) self.assertIn(said, message)
def test_a_session_moves_to_a_remote_location_and_back(self): session = self.app.session() moved = self.move(session, f"{HOST}:/srv//app/") self.assertEqual(moved["workspace"], f"{HOST}:/srv/app") self.assertEqual(moved["location"]["host"], HOST) with self.assertRaises(urllib.error.HTTPError) as invalid: self.move(session, f"{HOST}:app") self.assertEqual(invalid.exception.code, 400) self.assertIn("must be absolute", json.load(invalid.exception)["detail"]) back = self.move(session, str(self.app.workspace)) self.assertEqual(back["workspace"], str(self.app.workspace)) self.assertIsNone(back["location"]["host"])
if __name__ == "__main__": unittest.main()