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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279"""The workspace picker's folder browser: `/fs/list`, `/fs/repo`, `/fs/preview`.
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."""
import jsonimport osfrom pathlib import Pathimport shutilimport subprocessimport tempfileimport unittestimport urllib.errorimport urllib.parse
from harness import Albedo
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) self.root = Path(tempfile.mkdtemp(prefix="folders-", dir=self.app.root)) self.addCleanup(shutil.rmtree, self.root, True)
def get(self, route, path): query = urllib.parse.urlencode({"path": str(path)}) with self.app.api(f"/fs/{route}?{query}") as response: return json.load(response)
def failure(self, route, path): with self.assertRaises(urllib.error.HTTPError) as caught: self.get(route, path) return caught.exception.code, json.load(caught.exception)
def test_health_lists_the_browser(self): with self.app.api("/health") as response: self.assertIn("workspace_browser", json.load(response)["capabilities"])
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("list", f"{self.root}/./gamma/../") self.assertEqual(listed["path"], str(self.root)) self.assertFalse(listed["truncated"]) self.assertEqual( [(e["name"], e["hidden"], e["vcs"]) for e in listed["entries"]], [ (".hidden", True, None), ("Alpha", False, None), ("beta", False, None), ("gamma", False, None), ("zeta", False, None), ], ) self.assertEqual( listed["entries"][1]["modified"], int((self.root / "Alpha").stat().st_mtime) )
preview = self.get("preview", self.root) self.assertIsNone(preview["repo"]) self.assertEqual(preview["languages"], []) self.assertEqual(preview["more"], 0) self.assertEqual( [(e["name"], e["dir"]) for e in preview["tree"]], [ ("Alpha", True), ("beta", True), ("gamma", True), ("zeta", True), ("notes.txt", False), ], ) 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() touched = int(sh(repo, "git", "log", "-1", "--format=%ct"))
self.assertEqual(self.get("list", self.root)["entries"][0]["vcs"], "git") found = self.get("repo", repo / "src")["repo"] self.assertEqual( found, { "kind": "git", "root": str(repo), "branch": "trunk", "commit": commit, "changed": 2, "touched": touched, }, )
preview = self.get("preview", repo) self.assertEqual(preview["repo"], found) # JSON is data, so only Go and Python count, largest first. self.assertEqual( [language["name"] for language in preview["languages"]], ["Go", "Python"] ) self.assertEqual(preview["languages"][0]["color"], "#00ADD8") 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", repo)["repo"]["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") self.assertEqual( self.get("repo", repo)["repo"], { "kind": "git", "root": str(repo), "branch": "main", "commit": None, "changed": 1, "touched": 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("list", self.root)["entries"][0]["vcs"], "jj") found = self.get("repo", repo / "lib")["repo"] self.assertEqual(found["kind"], "jj") self.assertEqual(found["root"], str(repo)) self.assertEqual(found["change"], change) self.assertEqual(found["bookmark"], {"name": "main", "ahead": 2}) self.assertEqual(found["changed"], 1) self.assertIsInstance(found["touched"], int)
preview = self.get("preview", repo) self.assertEqual(preview["repo"], 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", repo)["repo"]["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("list", "~") self.assertEqual((listed["path"], listed["home"]), (str(home), str(home))) self.assertEqual(self.get("list", "~/")["path"], str(home))
(self.root / "file.txt").write_text("x\n") for route in ("list", "repo", "preview"): self.assertEqual(self.failure(route, "relative/dir")[0], 400) self.assertEqual(self.failure(route, "~someone")[0], 400) self.assertEqual(self.failure(route, "")[0], 400) code, body = self.failure(route, self.root / "missing") self.assertEqual(code, 404) self.assertIn("error", body) self.assertEqual(self.failure(route, self.root / "file.txt")[0], 404)