"""A session's checkout changes and file listings come from real git and jj. The diff client reviews everything a checkout changes, so these run git and jj in throwaway repositories through the daemon: every kind of change against HEAD and against the upstream fork point, the size limits, the ETag a polling client revalidates with, the listing with files and ignored marks, and that none of it writes to the index or jj's operation log. """ import json from pathlib import Path import shutil import tempfile import unittest import urllib.error import urllib.parse from harness import Albedo, Provider, operation_id, text from vcs_support import sh KIB = 1024 def lines(count, width=99): return "".join(f"{index:0{width}d}\n" for index in range(count)) class ChangesTest(unittest.TestCase): def setUp(self): self.provider = Provider(lambda request: text("ok")) self.addCleanup(self.provider.close) self.app = Albedo(self.provider).__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.root = Path(tempfile.mkdtemp(prefix="albedo-changes-", dir=self.app.root)) self.addCleanup(shutil.rmtree, self.root, True) def session(self, workspace): session = operation_id() with self.app.api( f"/sessions/{session}", { "kind": "new", "workspace": str(workspace), "provider_profile": self.app.profile, }, method="PUT", headers={"If-None-Match": "*"}, ) as response: json.load(response) return session def changes(self, session, base=None, etag=None): query = "" if base is None else "?base=" + base headers = {} if etag is None else {"If-None-Match": etag} with self.app.api( f"/sessions/{session}/changes{query}", headers=headers ) as response: body = response.read() return response.status, response.headers["ETag"], body def answer(self, session, base=None): status, _, body = self.changes(session, base) self.assertEqual(status, 200) return json.loads(body) def refused(self, session, base=None): with self.assertRaises(urllib.error.HTTPError) as caught: self.changes(session, base) return caught.exception.code, json.load(caught.exception)["code"] def listing(self, location, include=None): query = {"location": str(location)} if include: query["include"] = include with self.app.api("/workspaces?" + urllib.parse.urlencode(query)) as response: return json.load(response) def git_project(self): """A clone of a bare remote, one local commit ahead of origin/main.""" remote = self.root / "remote.git" sh(self.root, "git", "init", "-q", "--bare", "-b", "main", str(remote)) repo = self.root / "project" sh(self.root, "git", "clone", "-q", str(remote), str(repo)) sh(repo, "git", "checkout", "-q", "-b", "main") (repo / "mod.txt").write_text("a\nb\nc\n") (repo / "del.txt").write_text("gone\n") (repo / "ren.txt").write_text(lines(20)) (repo / "bin.dat").write_bytes(b"\0\1\2" * 10) (repo / "big.txt").write_text("small\n") (repo / ".gitignore").write_text("build/\n*.log\nlinked\n") sh(repo, "git", "add", ".") sh(repo, "git", "commit", "-q", "-m", "one") sh(repo, "git", "push", "-q", "-u", "origin", "main") (repo / "committed.txt").write_text("local\n") sh(repo, "git", "add", "committed.txt") sh(repo, "git", "commit", "-q", "-m", "two") # Every kind of change, staged and not. (repo / "mod.txt").write_text("a\nB\nc\nd\n") (repo / "del.txt").unlink() sh(repo, "git", "mv", "ren.txt", "renamed.txt") (repo / "staged.txt").write_text("new\n") sh(repo, "git", "add", "staged.txt") (repo / "bin.dat").write_bytes(b"\0\3\2" * 10) (repo / "big.txt").write_text(lines(3000)) (repo / "new.txt").write_text("u1\nu2") (repo / "blob.bin").write_bytes(b"\0" * 100) (repo / "build").mkdir() (repo / "build" / "out.o").write_text("ignored\n") (repo / "run.log").write_text("ignored\n") return repo def test_git_changes_against_head_and_upstream(self): repo = self.git_project() index = (repo / ".git" / "index").read_bytes() session = self.session(repo) head = sh(repo, "git", "rev-parse", "--short=7", "HEAD").strip() pushed = sh(repo, "git", "rev-parse", "--short=7", "origin/main").strip() found = self.answer(session) self.assertEqual( found["checkout"], {"kind": "git", "root": str(repo), "head": head, "branch": "main"}, ) self.assertEqual( found["base"], {"spec": "uncommitted", "label": "HEAD", "commit": head} ) self.assertFalse(found["truncated"]) files = {file["path"]: file for file in found["files"]} self.assertEqual( [file["path"] for file in found["files"]], sorted(files), ) self.assertEqual( {path: file["status"] for path, file in files.items()}, { "bin.dat": "modified", "big.txt": "modified", "blob.bin": "untracked", "del.txt": "deleted", "mod.txt": "modified", "new.txt": "untracked", "renamed.txt": "renamed", "staged.txt": "added", }, ) mod = files["mod.txt"] self.assertEqual((mod["added"], mod["removed"], mod["binary"]), (2, 1, False)) self.assertIsNone(mod["omitted"]) self.assertTrue(mod["diff"].startswith("diff --git a/mod.txt b/mod.txt\n")) self.assertIn("--- a/mod.txt\n+++ b/mod.txt\n", mod["diff"]) self.assertIn("\n-b\n+B\n", mod["diff"]) self.assertEqual(files["renamed.txt"]["old_path"], "ren.txt") self.assertEqual(files["del.txt"]["removed"], 1) self.assertEqual(files["staged.txt"]["added"], 1) for path in ("bin.dat", "blob.bin"): binary = files[path] self.assertEqual( (binary["binary"], binary["omitted"], binary["diff"]), (True, "binary", None), ) self.assertEqual((binary["added"], binary["removed"]), (0, 0)) big = files["big.txt"] self.assertEqual((big["omitted"], big["diff"]), ("too_large", None)) self.assertEqual((big["added"], big["removed"]), (3000, 1)) new = files["new.txt"] self.assertEqual((new["added"], new["removed"]), (2, 0)) self.assertEqual( new["diff"], "diff --git a/new.txt b/new.txt\nnew file mode 100644\n" "--- /dev/null\n+++ b/new.txt\n@@ -0,0 +1,2 @@\n+u1\n+u2\n" "\\ No newline at end of file", ) upstream = self.answer(session, "upstream") self.assertEqual( upstream["base"], {"spec": "upstream", "label": "origin/main", "commit": pushed}, ) listed = {file["path"]: file["status"] for file in upstream["files"]} self.assertEqual(listed["committed.txt"], "added") self.assertEqual(listed["new.txt"], "untracked") self.assertNotIn("committed.txt", files) # Revalidating costs nothing while nothing changed. status, etag, body = self.changes(session) self.assertEqual(status, 200) status, same, body = self.changes(session, etag=etag) self.assertEqual((status, same, body), (304, etag, b"")) (repo / "mod.txt").write_text("changed again\n") status, newer, _ = self.changes(session, etag=etag) self.assertEqual(status, 200) self.assertNotEqual(newer, etag) # Reading never wrote the index. self.assertEqual((repo / ".git" / "index").read_bytes(), index) def test_diffs_past_the_budget_are_left_out_in_order(self): repo = self.git_project() for name in "abcde": (repo / f"page-{name}.txt").write_text(lines(2000)) found = self.answer(self.session(repo)) omitted = { file["path"]: file["omitted"] for file in found["files"] if file["path"].startswith("page-") } self.assertEqual( omitted, { "page-a.txt": None, "page-b.txt": None, "page-c.txt": None, "page-d.txt": "budget", "page-e.txt": "budget", }, ) # Once one is over budget, every later diff is too. after = [ file for file in found["files"] if file["path"] > "page-d.txt" and file["omitted"] is None ] self.assertEqual(after, []) self.assertEqual( { file["added"] for file in found["files"] if file["path"].startswith("page-") }, {2000}, ) def test_refusals(self): plain = self.root / "plain" plain.mkdir() self.assertEqual(self.refused(self.session(plain)), (409, "not_a_repository")) lone = self.root / "lone" lone.mkdir() sh(lone, "git", "init", "-q", "-b", "main") (lone / "a.txt").write_text("a\n") sh(lone, "git", "add", ".") sh(lone, "git", "commit", "-q", "-m", "one") session = self.session(lone) self.assertEqual(self.refused(session, "upstream"), (409, "no_upstream")) self.assertEqual(self.refused(session, "elsewhere"), (400, "invalid_request")) self.assertEqual(self.refused(operation_id())[0], 404) # An unborn HEAD compares with nothing. fresh = self.root / "fresh" fresh.mkdir() sh(fresh, "git", "init", "-q", "-b", "main") (fresh / "draft.md").write_text("# draft\n") found = self.answer(self.session(fresh)) self.assertEqual( found["base"], {"spec": "uncommitted", "label": "HEAD", "commit": None} ) self.assertEqual(found["checkout"]["head"], None) self.assertEqual( [(file["path"], file["status"]) for file in found["files"]], [("draft.md", "untracked")], ) def test_listing_with_files_marks_what_git_ignores(self): repo = self.git_project() (repo / "linked").symlink_to(repo / "build") plain = self.listing(repo) self.assertEqual( [item["name"] for item in plain["items"]], [".git", "build", "linked"] ) self.assertNotIn("kind", plain["items"][0]) listed = self.listing(repo, "files") items = {item["name"]: item for item in listed["items"]} self.assertEqual( [item["name"] for item in listed["items"]], [ ".git", "build", "linked", ".gitignore", "big.txt", "bin.dat", "blob.bin", "committed.txt", "mod.txt", "new.txt", "renamed.txt", "run.log", "staged.txt", ], ) self.assertEqual( (items["build"]["kind"], items["build"]["size"], items["build"]["ignored"]), ("directory", None, True), ) self.assertEqual( ( items["mod.txt"]["kind"], items["mod.txt"]["size"], items["mod.txt"]["ignored"], ), ("file", 8, False), ) self.assertTrue(items["run.log"]["ignored"]) self.assertEqual( (items["linked"]["kind"], items["linked"]["ignored"]), ("directory", True) ) self.assertFalse(items["new.txt"]["ignored"]) self.assertEqual(items["mod.txt"]["location"], str(repo / "mod.txt")) self.assertTrue(items[".gitignore"]["hidden"]) both = self.listing(repo, "preview,files") self.assertEqual(both["items"], listed["items"]) self.assertEqual(both["preview"]["repository"]["kind"], "git") outside = self.root / "outside" outside.mkdir() (outside / "notes.txt").write_text("x\n") self.assertEqual( [ (i["name"], i["kind"], i["ignored"]) for i in self.listing(outside, "files")["items"] ], [("notes.txt", "file", None)], ) @unittest.skipUnless(shutil.which("jj"), "jj is not installed") def test_jj_changes_without_writing_an_operation(self): colocated = self.root / "colocated" colocated.mkdir() sh(colocated, "jj", "git", "init", "--colocate") (colocated / "a.txt").write_text("one\n") (colocated / "b.txt").write_text("bee\n") sh(colocated, "jj", "commit", "-m", "one") sh(colocated, "jj", "bookmark", "create", "main", "-r", "@-") (colocated / "a.txt").write_text("two\n") (colocated / "b.txt").unlink() sh(colocated, "jj", "status") parent = sh( colocated, "jj", "log", "-r", "@-", "--no-graph", "-T", "commit_id.short(12)", ).strip() # Not snapshotted yet: git sees it, so the colocated answer does. (colocated / "later.txt").write_text("later\n") operations = self.operations(colocated) session = self.session(colocated) found = self.answer(session) self.assertEqual(found["checkout"]["kind"], "jj") self.assertEqual(found["checkout"]["branch"], "main") self.assertEqual( found["base"], {"spec": "uncommitted", "label": "@-", "commit": parent} ) self.assertEqual( [(f["path"], f["status"]) for f in found["files"]], [("a.txt", "modified"), ("b.txt", "deleted"), ("later.txt", "added")], ) upstream = self.answer(session, "upstream") # Without a remote, trunk() is the root: everything is new. self.assertEqual( upstream["base"], {"spec": "upstream", "label": "trunk()", "commit": None} ) self.assertEqual( [(f["path"], f["status"]) for f in upstream["files"]], [("a.txt", "added"), ("later.txt", "added")], ) self.assertEqual(self.operations(colocated), operations) alone = self.root / "alone" alone.mkdir() sh(alone, "jj", "git", "init", "--no-colocate") (alone / "a.txt").write_text("one\n") sh(alone, "jj", "commit", "-m", "one") (alone / "a.txt").write_text("two\n") (alone / "c.txt").write_text("sea\n") sh(alone, "jj", "status") (alone / "later.txt").write_text("later\n") operations = self.operations(alone) found = self.answer(self.session(alone)) # Without git, the answer is as fresh as jj's last snapshot. self.assertEqual( [ (f["path"], f["status"], f["added"], f["removed"]) for f in found["files"] ], [("a.txt", "modified", 1, 1), ("c.txt", "added", 1, 0)], ) self.assertIn("+two", found["files"][0]["diff"]) self.assertEqual(self.operations(alone), operations) @staticmethod def operations(repo): return sh( repo, "jj", "op", "log", "--no-graph", "--ignore-working-copy", "-T", 'id ++ "\\n"', ) if __name__ == "__main__": unittest.main()