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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305"""Workspace directory and preview resources preserve real filesystem and VCS facts.
What the picker shows comes from real directories, git and jj run by thedaemon with its own environment, so the checks are end to end: a plain tree,a git repository (branch, changes, detached HEAD), a colocated jj repository(jj wins, bookmarks, and browsing never writes a jj operation), linguistshares, and path validation. The jj cases skip when jj is not installed."""
from datetime import datetimeimport jsonimport osfrom pathlib import Pathimport shutilimport subprocessimport tempfileimport unittestimport urllib.errorimport urllib.parse
from harness import Albedo, exclusive
GIT_ENV = { "GIT_AUTHOR_NAME": "fixture", "GIT_AUTHOR_EMAIL": "fixture@example.com", "GIT_COMMITTER_NAME": "fixture", "GIT_COMMITTER_EMAIL": "fixture@example.com", "JJ_USER": "fixture", "JJ_EMAIL": "fixture@example.com",}
def sh(cwd, *args): return subprocess.run( args, cwd=cwd, env=dict(os.environ, **GIT_ENV), check=True, capture_output=True, text=True, ).stdout
def write(path, text): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text)
class FoldersTest(unittest.TestCase): def setUp(self): self.app = Albedo().__enter__() self.addCleanup(self.app.__exit__, None, None, None) directory = ( Path("/var/tmp") if self._testMethodName == "test_plain_tree_lists_directories_and_previews_two_levels" else self.app.root ) self.root = Path(tempfile.mkdtemp(prefix="albedo-folders-", dir=directory)) self.addCleanup(shutil.rmtree, self.root, True)
def get(self, location, *, preview=False): query = {"location": str(location)} if preview: query["include"] = "preview" with self.app.api("/workspaces?" + urllib.parse.urlencode(query)) as response: return json.load(response)
def failure(self, location, *, preview=False): with self.assertRaises(urllib.error.HTTPError) as caught: self.get(location, preview=preview) return caught.exception.code, json.load(caught.exception)
@exclusive def test_plain_tree_lists_directories_and_previews_two_levels(self): for name in ("beta", "Alpha", ".hidden", "gamma/one", "gamma/two"): (self.root / name).mkdir(parents=True) for name in "abcdef": (self.root / "gamma" / f"{name}.py").write_text("x = 1\n") (self.root / "notes.txt").write_text("not a directory\n") (self.root / "zeta").symlink_to(self.root / "beta")
listed = self.get(f"{self.root}/./gamma/../") self.assertEqual(listed["directory"], str(self.root)) self.assertIsNone(listed["next"]) self.assertEqual( [(e["name"], e["hidden"], e["vcs"]) for e in listed["items"]], [ (".hidden", True, None), ("Alpha", False, None), ("beta", False, None), ("gamma", False, None), ("zeta", False, None), ], ) modified = datetime.fromisoformat( listed["items"][1]["modified_at"].replace("Z", "+00:00") ) self.assertAlmostEqual( modified.timestamp(), (self.root / "Alpha").stat().st_mtime, delta=1 )
preview = self.get(self.root, preview=True)["preview"] self.assertIsNone(preview["repository"]) self.assertEqual(preview["languages"], []) self.assertEqual(preview["more"], 0) self.assertEqual( [(e["name"], e["kind"]) for e in preview["tree"]], [ ("Alpha", "directory"), ("beta", "directory"), ("gamma", "directory"), ("zeta", "symlink"), ("notes.txt", "file"), ], ) gamma = preview["tree"][2] self.assertEqual( [c["name"] for c in gamma["children"]], ["one", "two", "a.py", "b.py"] ) self.assertEqual(gamma["more"], 4) self.assertNotIn("children", gamma["children"][0]) self.assertEqual(gamma["children"][2]["language"], "Python") self.assertEqual(preview["tree"][4]["language"], "Text")
def test_git_repository_reports_branch_changes_and_detached_head(self): repo = self.root / "project" write(repo / "src" / "main.go", "package main\n" * 30) write(repo / "src" / "util.go", "package main\n" * 30) write(repo / "script.py", "print(1)\n" * 10) write(repo / "data.json", "{}\n" * 500) write(repo / ".gitignore", "build/\n") sh(repo, "git", "init", "-q", "-b", "trunk") sh(repo, "git", "add", ".") sh(repo, "git", "commit", "-q", "-m", "one") write(repo / "src" / "main.go", "package main // edited\n") write(repo / "src" / "new.go", "package main\n") write(repo / "build" / "out.o", "ignored\n") commit = sh(repo, "git", "rev-parse", "--short", "HEAD").strip()
self.assertEqual(self.get(self.root)["items"][0]["vcs"], "git") found = self.get(repo / "src", preview=True)["preview"]["repository"] self.assertEqual( { key: found[key] for key in ("kind", "root", "branch", "revision", "changed") }, { "kind": "git", "root": str(repo), "branch": "trunk", "revision": commit, "changed": 2, }, )
preview = self.get(repo, preview=True)["preview"] self.assertEqual(preview["repository"], found) # JSON is data, so only Go and Python count, largest first. self.assertEqual( [language["name"] for language in preview["languages"]], ["Go", "Python"] ) self.assertAlmostEqual( sum(language["share"] for language in preview["languages"]), 1.0 ) tree = {e["name"]: e for e in preview["tree"]} # Inside a repository only tracked or changed paths show: the # ignored build/ stays out, the untracked src/new.go is there. self.assertEqual(list(tree), ["src", "data.json", "script.py"]) self.assertEqual(tree["src"]["changed"], 2) self.assertEqual( {c["name"]: c["changed"] for c in tree["src"]["children"]}, {"main.go": 1, "new.go": 1, "util.go": 0}, ) self.assertEqual(tree["script.py"]["changed"], 0)
sh(repo, "git", "checkout", "-q", "--detach") self.assertIsNone( self.get(repo, preview=True)["preview"]["repository"]["branch"] )
def test_unborn_git_repository_has_no_commit(self): repo = self.root / "fresh" repo.mkdir() sh(repo, "git", "init", "-q", "-b", "main") (repo / "draft.md").write_text("# draft\n") found = self.get(repo, preview=True)["preview"]["repository"] self.assertEqual( { key: found[key] for key in ( "kind", "root", "branch", "revision", "changed", "touched_at", ) }, { "kind": "git", "root": str(repo), "branch": "main", "revision": None, "changed": 1, "touched_at": None, }, )
@unittest.skipUnless(shutil.which("jj"), "jj is not installed") def test_colocated_jj_repository_wins_without_writing_an_operation(self): repo = self.root / "colocated" repo.mkdir() sh(repo, "jj", "git", "init", "--colocate") write(repo / "lib" / "core.gleam", "pub fn main() { Nil }\n") sh(repo, "jj", "commit", "-m", "one") sh(repo, "jj", "bookmark", "create", "main", "-r", "@-") write(repo / "lib" / "more.gleam", "pub fn more() { Nil }\n") sh(repo, "jj", "commit", "-m", "two") write(repo / "lib" / "core.gleam", "pub fn main() { 1 }\n") sh(repo, "jj", "status") change = sh( repo, "jj", "log", "-r", "@", "--no-graph", "-T", "change_id.shortest(4)" ).strip() # Not yet snapshotted, so jj does not know it and the tree leaves it # out; a browse that snapshotted would record an operation. write(repo / "lib" / "later.gleam", "pub fn later() { Nil }\n") operations = self.operations(repo)
self.assertEqual(self.get(self.root)["items"][0]["vcs"], "jj") found = self.get(repo / "lib", preview=True)["preview"]["repository"] self.assertEqual(found["kind"], "jj") self.assertEqual(found["root"], str(repo)) self.assertEqual(found["change_id"], change) self.assertEqual(found["bookmark"], {"name": "main", "ahead": 2}) self.assertEqual(found["changed"], 1) self.assertIsInstance(found["touched_at"], str) datetime.fromisoformat(found["touched_at"].replace("Z", "+00:00"))
preview = self.get(repo, preview=True)["preview"] self.assertEqual(preview["repository"], found) self.assertEqual( [ (language["name"], language["share"]) for language in preview["languages"] ], [("Gleam", 1.0)], ) lib = preview["tree"][0] self.assertEqual((lib["name"], lib["changed"]), ("lib", 1)) self.assertEqual( {c["name"]: c["changed"] for c in lib["children"]}, {"core.gleam": 1, "more.gleam": 0}, )
self.assertEqual(self.operations(repo), operations)
# A bookmark that moved on without @ still names @'s line of work, # rather than an older backup behind @. def log(rev): return sh( repo, "jj", "log", "-r", rev, "--no-graph", "-T", "change_id" ).strip()
two, one = log("@-"), log("@--") sh(repo, "jj", "bookmark", "create", "backup/old", "-r", one) sh(repo, "jj", "new", two, "-m", "side") sh(repo, "jj", "bookmark", "set", "main", "-r", "@") sh(repo, "jj", "edit", change) self.assertEqual( self.get(repo, preview=True)["preview"]["repository"]["bookmark"], {"name": "main", "ahead": 1}, )
@staticmethod def operations(repo): return sh( repo, "jj", "op", "log", "--no-graph", "--ignore-working-copy", "-T", 'id ++ "\\n"', )
def test_paths_expand_home_and_refuse_the_rest(self): home = self.app.root / "user-home" listed = self.get("~") self.assertEqual((listed["directory"], listed["home"]), (str(home), str(home))) self.assertEqual(self.get("~/")["directory"], str(home))
(self.root / "file.txt").write_text("x\n") for preview in (False, True): self.assertEqual(self.failure("relative/dir", preview=preview)[0], 400) self.assertEqual(self.failure("~someone", preview=preview)[0], 400) self.assertEqual(self.failure("", preview=preview)[0], 400) code, body = self.failure(self.root / "missing", preview=preview) self.assertEqual(code, 404) self.assertEqual(body["status"], 404) self.assertEqual( self.failure(self.root / "file.txt", preview=preview)[0], 404 )