Something went wrong. Try again.
fork of prime agent with changes i want but are not upstreamable probqbly (and some fixes)
Something went wrong. Try again.
16 kB · 323 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324from __future__ import annotations
import subprocessimport tempfileimport unittestfrom pathlib import Path
import git
class GitSurfaceTests(unittest.TestCase): def test_grep_streams_only_the_bounded_revision_matches(self) -> None: with tempfile.TemporaryDirectory(prefix="git-grep-test-") as temporary: root = Path(temporary) subprocess.run(["git", "init", "-q", str(root)], check=True) (root / "source.txt").write_text("needle one\nneedle two\n", encoding="utf-8") (root / "ignored.md").write_text("needle three\n", encoding="utf-8") subprocess.run(["git", "-C", str(root), "add", "source.txt", "ignored.md"], check=True) subprocess.run( [ "git", "-C", str(root), "-c", "user.name=test", "-c", "user.email=test@example.com", "commit", "-qm", "fixture", ], check=True, )
result = git.grep("needle", globs=("*.txt",), max_hits=1, cwd=str(root))
self.assertEqual(result, [("source.txt", 1, "needle one")])
def test_grep_rejects_unbounded_hit_limits(self) -> None: with self.assertRaisesRegex(ValueError, "max_hits must be between 1 and 1000"): git.grep("needle", max_hits=0)
def test_commit_list_renders_pretty_and_indexes_dicts(self) -> None: raw = [ { "hash": "1234567890abcdef", "short": "123456789", "author": "Alice", "email": "alice@example.com", "date": "2026-09-03T12:00:00Z", "refs": ["HEAD -> main", "origin/main"], "subject": "initial commit", } ] cl = git.CommitList(raw) self.assertEqual(len(cl), 1) self.assertEqual(cl[0]["hash"], "1234567890abcdef") self.assertEqual(cl[0]["subject"], "initial commit") self.assertIn("123456789 (HEAD -> main, origin/main) initial commit [Alice, 2026-09-03]", repr(cl)) self.assertIn("123456789", str(cl))
def test_run_rejects_subcommands_with_native_functions(self) -> None: for subcmd in ("diff", "log", "status", "add", "commit", "checkout", "switch", "reset", "restore"): with self.assertRaises(git.GitError) as ctx: git.run(subcmd) self.assertIn(f"git.run('{subcmd}', ...) is rejected", str(ctx.exception))
def test_run_allows_unmapped_subcommands(self) -> None: with tempfile.TemporaryDirectory(prefix="git-run-test-") as temporary: root = Path(temporary) subprocess.run(["git", "init", "-q", str(root)], check=True) res = git.run("rev-parse", "--is-inside-work-tree", cwd=str(root)) self.assertEqual(res.strip(), "true")
def test_diff_name_only_and_extra_flags(self) -> None: with tempfile.TemporaryDirectory(prefix="git-diff-test-") as temporary: root = Path(temporary) subprocess.run(["git", "init", "-q", str(root)], check=True) (root / "file.txt").write_text("hello", encoding="utf-8") subprocess.run(["git", "-C", str(root), "add", "file.txt"], check=True) subprocess.run( ["git", "-C", str(root), "-c", "user.name=test", "-c", "user.email=t@example.com", "commit", "-qm", "init"], check=True, ) (root / "file.txt").write_text("world", encoding="utf-8") res = git.diff("file.txt", cwd=str(root), name_only=True) self.assertIn("file.txt", res)
def test_log_with_format(self) -> None: with tempfile.TemporaryDirectory(prefix="git-log-test-") as temporary: root = Path(temporary) subprocess.run(["git", "init", "-q", str(root)], check=True) (root / "file.txt").write_text("hello", encoding="utf-8") subprocess.run(["git", "-C", str(root), "add", "file.txt"], check=True) subprocess.run( ["git", "-C", str(root), "-c", "user.name=test", "-c", "user.email=t@example.com", "commit", "-qm", "my_subject"], check=True, ) res = git.log(cwd=str(root), format="%s") self.assertEqual(res.strip(), "my_subject")
def test_awaitable_git_results(self) -> None: import asyncio with tempfile.TemporaryDirectory(prefix="git-await-test-") as temporary: root = Path(temporary) subprocess.run(["git", "init", "-q", str(root)], check=True) (root / "source.txt").write_text("hello\n", encoding="utf-8") subprocess.run(["git", "-C", str(root), "add", "source.txt"], check=True) subprocess.run( [ "git", "-C", str(root), "-c", "user.name=test", "-c", "user.email=test@example.com", "commit", "-qm", "fixture", ], check=True, ) async def _test(): st = await git.status(cwd=str(root)) lg = await git.log(cwd=str(root)) return st, lg status, log_entries = asyncio.run(_test()) self.assertIn("##", status) self.assertEqual(len(log_entries), 1)
def test_cwd_keyword_is_the_run_directory(self) -> None: with tempfile.TemporaryDirectory(prefix="git-cwd-test-") as temporary: root = Path(temporary) subprocess.run(["git", "init", "-q", str(root)], check=True) res = git.root(cwd=str(root)) self.assertEqual(Path(res).resolve(), root.resolve())
def test_path_target_is_not_a_cwd(self) -> None: with tempfile.TemporaryDirectory(prefix="git-notdir-test-") as temporary: root = Path(temporary) subprocess.run(["git", "init", "-q", str(root)], check=True) file_path = root / "some_file.rs" file_path.write_text("fn main() {}", encoding="utf-8") with self.assertRaises(git.GitError) as ctx: git.diff(cwd=str(file_path)) # The error names the parameter, not just the failed git invocation. self.assertIn("cwd=", str(ctx.exception)) self.assertIn("paths=[...]", str(ctx.exception))
def test_log_limit_keyword(self) -> None: with tempfile.TemporaryDirectory(prefix="git-limit-test-") as temporary: root = Path(temporary) subprocess.run(["git", "init", "-q", str(root)], check=True) for i in range(3): (root / f"f{i}.txt").write_text(f"{i}", encoding="utf-8") subprocess.run(["git", "-C", str(root), "add", f"f{i}.txt"], check=True) subprocess.run( ["git", "-C", str(root), "-c", "user.name=t", "-c", "user.email=t@e.com", "commit", "-qm", f"msg {i}"], check=True, ) entries = git.log(cwd=str(root), limit=2) self.assertEqual(len(entries), 2)
def test_diff_paths_keyword(self) -> None: with tempfile.TemporaryDirectory(prefix="git-diff-paths-") as temporary: root = Path(temporary) subprocess.run(["git", "init", "-q", str(root)], check=True) (root / "a.txt").write_text("hello", encoding="utf-8") (root / "b.txt").write_text("world", encoding="utf-8") subprocess.run(["git", "-C", str(root), "add", "."], check=True) subprocess.run( ["git", "-C", str(root), "-c", "user.name=t", "-c", "user.email=t@e.com", "commit", "-qm", "init"], check=True, ) (root / "a.txt").write_text("hello2", encoding="utf-8") (root / "b.txt").write_text("world2", encoding="utf-8") res_a = git.diff(paths=["a.txt"], cwd=str(root), name_only=True) self.assertIn("a.txt", res_a) self.assertNotIn("b.txt", res_a) res_both = git.diff(paths=["a.txt", "b.txt"], cwd=str(root), name_only=True) self.assertIn("a.txt", res_both) self.assertIn("b.txt", res_both)
def test_grep_reports_truncation(self) -> None: with tempfile.TemporaryDirectory(prefix="git-grep-trunc-") as temporary: root = Path(temporary) subprocess.run(["git", "init", "-q", str(root)], check=True) (root / "many.txt").write_text("\n".join(f"match {i}" for i in range(10)), encoding="utf-8") subprocess.run(["git", "-C", str(root), "add", "many.txt"], check=True) subprocess.run( ["git", "-C", str(root), "-c", "user.name=t", "-c", "user.email=t@e.com", "commit", "-qm", "matches"], check=True, ) res = git.grep("match", cwd=str(root), max_hits=3) self.assertEqual(len(res), 3) self.assertTrue(res.truncated)
def test_subagent_push_attenuation(self) -> None: import os with tempfile.TemporaryDirectory(prefix="git-push-att-") as temporary: root = Path(temporary) subprocess.run(["git", "init", "-q", str(root)], check=True) old_depth = os.environ.get("RLM_DEPTH") old_allow = os.environ.get("RLM_ALLOW_PUSH") try: os.environ["RLM_DEPTH"] = "1" os.environ.pop("RLM_ALLOW_PUSH", None) with self.assertRaises(PermissionError) as ctx: git.push("origin", "main", cwd=str(root)) self.assertIn("git.push is disabled in subagents", str(ctx.exception)) self.assertIn("agent_message.send", str(ctx.exception))
# Exception via kwarg # Dry run against non-existent remote will fail with GitError not PermissionError with self.assertRaises(git.GitError): git.push("origin", "main", cwd=str(root), allow_push=True)
# Exception via env var os.environ["RLM_ALLOW_PUSH"] = "1" with self.assertRaises(git.GitError): git.push("origin", "main", cwd=str(root)) finally: if old_depth is None: os.environ.pop("RLM_DEPTH", None) else: os.environ["RLM_DEPTH"] = old_depth if old_allow is None: os.environ.pop("RLM_ALLOW_PUSH", None) else: os.environ["RLM_ALLOW_PUSH"] = old_allow
def test_subagent_reset_attenuation(self) -> None: import os with tempfile.TemporaryDirectory(prefix="git-reset-att-") as temporary: root = Path(temporary) subprocess.run(["git", "init", "-q", str(root)], check=True) (root / "file.txt").write_text("hello", encoding="utf-8") subprocess.run(["git", "-C", str(root), "add", "file.txt"], check=True) subprocess.run( ["git", "-C", str(root), "-c", "user.name=t", "-c", "user.email=t@e.com", "commit", "-qm", "init"], check=True, ) # Make working tree dirty (root / "file.txt").write_text("dirty content", encoding="utf-8")
old_depth = os.environ.get("RLM_DEPTH") try: os.environ["RLM_DEPTH"] = "1" with self.assertRaises(PermissionError) as ctx: git.reset("HEAD", cwd=str(root)) self.assertIn("force=True", str(ctx.exception))
# With force=True, it succeeds git.reset("HEAD", cwd=str(root), force=True) finally: if old_depth is None: os.environ.pop("RLM_DEPTH", None) else: os.environ["RLM_DEPTH"] = old_depth
def test_path_fails_loudly_with_both_corrections(self) -> None: with tempfile.TemporaryDirectory(prefix="git-path-synonym-") as temporary: root = Path(temporary) subprocess.run(["git", "init", "-q", str(root)], check=True) (root / "a.txt").write_text("hello", encoding="utf-8") (root / "b.txt").write_text("world", encoding="utf-8") subprocess.run(["git", "-C", str(root), "add", "."], check=True) subprocess.run( ["git", "-C", str(root), "-c", "user.name=t", "-c", "user.email=t@e.com", "commit", "-qm", "init"], check=True, ) (root / "a.txt").write_text("hello2", encoding="utf-8")
# `paths=` is the only target keyword; the old `path=` fails loudly with both # corrections instead of quietly filtering the wrong thing (it used to mean cwd). for call in ( lambda: git.diff(path="a.txt", cwd=str(root), name_only=True), lambda: git.diff(path=["a.txt", "b.txt"], cwd=str(root), name_only=True), lambda: git.log(path="a.txt", cwd=str(root)), lambda: git.diff(paths=["a.txt"], path="a.txt", cwd=str(root)), ): with self.assertRaisesRegex(TypeError, "path= was renamed") as caught: call() self.assertIn("cwd=", str(caught.exception)) self.assertIn("paths=[...]", str(caught.exception)) self.assertEqual( git.diff(paths=["a.txt"], cwd=str(root), name_only=True), git.diff(paths=["a.txt"], cwd=str(root), name_only=True), )
def test_settled_results_await_to_the_same_value(self) -> None: import asyncio import inspect
with tempfile.TemporaryDirectory(prefix="git-settled-") as temporary: root = Path(temporary) subprocess.run(["git", "init", "-q", str(root)], check=True) (root / "source.txt").write_text("needle\n", encoding="utf-8") subprocess.run(["git", "-C", str(root), "add", "source.txt"], check=True) subprocess.run( ["git", "-C", str(root), "-c", "user.name=t", "-c", "user.email=t@e.com", "commit", "-qm", "fixture"], check=True, ) calls = { "root": git.root(cwd=str(root)), "status": git.status(cwd=str(root)), "diff": git.diff(cwd=str(root)), "grep": git.grep("needle", cwd=str(root)), "log": git.log(cwd=str(root)), } for name, value in calls.items(): self.assertTrue(inspect.isawaitable(value), name)
async def _test(): return { "root": await git.root(cwd=str(root)), "status": await git.status(cwd=str(root)), "diff": await git.diff(cwd=str(root)), "grep": await git.grep("needle", cwd=str(root)), "log": await git.log(cwd=str(root)), }
awaited = asyncio.run(_test())
for name in ("root", "status", "diff"): self.assertEqual(awaited[name], calls[name], name) self.assertIsInstance(awaited[name], str) self.assertEqual(list(awaited["grep"]), list(calls["grep"])) self.assertEqual(awaited["log"], calls["log"]) # The skill's own result type and repr survive the settle. self.assertIs(type(awaited["log"]), git.CommitList) self.assertEqual(repr(awaited["log"]), repr(calls["log"]))
if __name__ == "__main__": unittest.main()