feat: codegraph

tests/test_codegraph.py

@@ -0,0 +1,57 @@
+from pathlib import Path
+
+from agentci.adapters.codegraph import CodeGraphClient
+
+
+class FakeProcess:
+    returncode = 0
+
+    async def communicate(self) -> tuple[bytes, bytes]:
+        return b"", b""
+
+
+async def test_initializes_index_and_excludes_it_from_git(
+    tmp_path: Path, monkeypatch
+) -> None:
+    workspace = tmp_path / "repo"
+    (workspace / ".git" / "info").mkdir(parents=True)
+    calls: list[tuple[object, ...]] = []
+
+    async def create_subprocess_exec(*args, **_kwargs):
+        calls.append(args)
+        return FakeProcess()
+
+    monkeypatch.setattr(
+        "agentci.adapters.codegraph.asyncio.create_subprocess_exec",
+        create_subprocess_exec,
+    )
+
+    await CodeGraphClient().prepare(workspace)
+
+    assert calls == [("codegraph", "init", str(workspace))]
+    assert (workspace / ".git" / "info" / "exclude").read_text() == ".codegraph/\n"
+
+
+async def test_syncs_an_existing_index_without_duplicating_exclude(
+    tmp_path: Path, monkeypatch
+) -> None:
+    workspace = tmp_path / "repo"
+    (workspace / ".git" / "info").mkdir(parents=True)
+    (workspace / ".codegraph").mkdir()
+    exclude = workspace / ".git" / "info" / "exclude"
+    exclude.write_text("# local excludes\n.codegraph/\n")
+    calls: list[tuple[object, ...]] = []
+
+    async def create_subprocess_exec(*args, **_kwargs):
+        calls.append(args)
+        return FakeProcess()
+
+    monkeypatch.setattr(
+        "agentci.adapters.codegraph.asyncio.create_subprocess_exec",
+        create_subprocess_exec,
+    )
+
+    await CodeGraphClient().prepare(workspace)
+
+    assert calls == [("codegraph", "sync", str(workspace))]
+    assert exclude.read_text() == "# local excludes\n.codegraph/\n"