87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise agent Git workflow against a local temporary repository."""
|
|
import asyncio
|
|
import importlib.util
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
GIT_OPS_PATH = ROOT / "agent" / "git_operations.py"
|
|
|
|
spec = importlib.util.spec_from_file_location("git_operations", GIT_OPS_PATH)
|
|
git_operations = importlib.util.module_from_spec(spec)
|
|
assert spec and spec.loader
|
|
spec.loader.exec_module(git_operations)
|
|
GitOperations = git_operations.GitOperations
|
|
|
|
|
|
def run(command: list[str], cwd: Optional[Path] = None):
|
|
result = subprocess.run(
|
|
command,
|
|
cwd=cwd,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(
|
|
f"Command failed: {' '.join(command)}\nSTDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"
|
|
)
|
|
return result
|
|
|
|
|
|
async def main():
|
|
with tempfile.TemporaryDirectory(prefix="swarm-git-test-") as tmp:
|
|
tmpdir = Path(tmp)
|
|
source = tmpdir / "source"
|
|
remote = tmpdir / "remote.git"
|
|
workspace = tmpdir / "workspace"
|
|
|
|
source.mkdir()
|
|
run(["git", "init", "-b", "main"], cwd=source)
|
|
run(["git", "config", "user.name", "Test User"], cwd=source)
|
|
run(["git", "config", "user.email", "test@example.com"], cwd=source)
|
|
|
|
(source / "hello.py").write_text(
|
|
"def hello_world():\n return 'hello'\n",
|
|
encoding="utf-8",
|
|
)
|
|
run(["git", "add", "hello.py"], cwd=source)
|
|
run(["git", "commit", "-m", "Initial commit"], cwd=source)
|
|
run(["git", "clone", "--bare", str(source), str(remote)])
|
|
|
|
os.environ["GIT_BASE_BRANCH"] = "main"
|
|
git_ops = GitOperations(str(workspace), "test-agent")
|
|
|
|
assert await git_ops.clone_workspace(str(remote))
|
|
assert await git_ops.is_git_workspace()
|
|
assert await git_ops.create_result_branch("task-123")
|
|
|
|
(workspace / "hello.py").write_text(
|
|
"def hello_world():\n return 'hello from agent'\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
commit_sha = await git_ops.commit_changes("Task task-123: update hello")
|
|
assert commit_sha
|
|
|
|
branch_name = await git_ops.push_results()
|
|
assert branch_name
|
|
|
|
refs = run(["git", "for-each-ref", "--format=%(refname:short)", "refs/heads"], cwd=remote)
|
|
assert branch_name in refs.stdout.splitlines()
|
|
|
|
print("Git workflow test passed")
|
|
print(f"branch={branch_name}")
|
|
print(f"commit={commit_sha}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|