"""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 the
daemon 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), linguist
shares, and path validation. The jj cases skip when jj is not installed.
"""
import json
import os
from pathlib import Path
import shutil
import subprocess
import tempfile
import unittest
import urllib.error
import 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)