276 lines
9.7 KiB
Python
276 lines
9.7 KiB
Python
import os
|
|
import sys
|
|
import subprocess
|
|
import pytest
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
|
|
def run_diff_collect(repo_root, base_commit="", env=None):
|
|
script_path = Path(__file__).parent.parent / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "diff_collect.sh"
|
|
cmd = ["bash", str(script_path), str(repo_root)]
|
|
if base_commit:
|
|
cmd.append(base_commit)
|
|
run_env = dict(os.environ)
|
|
if env:
|
|
run_env.update(env)
|
|
res = subprocess.run(cmd, capture_output=True, text=True, env=run_env)
|
|
return res
|
|
|
|
|
|
@pytest.fixture
|
|
def b7_repo(tmp_path):
|
|
repo = tmp_path / "b7_repo"
|
|
repo.mkdir()
|
|
subprocess.run(["git", "init"], cwd=str(repo), capture_output=True)
|
|
subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=str(repo), capture_output=True)
|
|
subprocess.run(["git", "config", "user.name", "Test User"], cwd=str(repo), capture_output=True)
|
|
|
|
(repo / ".gitignore").write_text("ignored.log\n")
|
|
(repo / "tracked.txt").write_text("line 1\n")
|
|
subprocess.run(["git", "add", "."], cwd=str(repo), capture_output=True)
|
|
subprocess.run(["git", "commit", "-m", "initial commit"], cwd=str(repo), capture_output=True)
|
|
|
|
base_commit = subprocess.run(["git", "rev-parse", "HEAD"], cwd=str(repo), capture_output=True, text=True).stdout.strip()
|
|
return repo, base_commit
|
|
|
|
|
|
def test_b7_1_untracked_new_file_content(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
(repo / "newfile.py").write_text("print('hello')\n")
|
|
|
|
res = run_diff_collect(repo, base_commit)
|
|
assert res.returncode == 0
|
|
assert "newfile.py" in res.stdout
|
|
assert "print('hello')" in res.stdout
|
|
|
|
|
|
def test_b7_2_tracked_file_modification(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
with open(repo / "tracked.txt", "a") as f:
|
|
f.write("line 2\n")
|
|
|
|
res = run_diff_collect(repo, base_commit)
|
|
assert res.returncode == 0
|
|
assert "tracked.txt" in res.stdout
|
|
assert "+line 2" in res.stdout
|
|
|
|
|
|
def test_b7_3_gitignore_respected(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
(repo / "ignored.log").write_text("secret log\n")
|
|
|
|
res = run_diff_collect(repo, base_commit)
|
|
assert res.returncode == 0
|
|
assert "ignored.log" not in res.stdout
|
|
|
|
|
|
def test_b7_4_index_not_mutated(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
(repo / "untracked.txt").write_text("untracked\n")
|
|
|
|
run_diff_collect(repo, base_commit)
|
|
|
|
status = subprocess.run(["git", "status", "--porcelain"], cwd=str(repo), capture_output=True, text=True).stdout
|
|
assert "??" in status
|
|
assert "A untracked.txt" not in status
|
|
|
|
|
|
def test_b7_5_git_commit_am_safety(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
(repo / "newfile.txt").write_text("new content\n")
|
|
with open(repo / "tracked.txt", "a") as f:
|
|
f.write("mod\n")
|
|
|
|
run_diff_collect(repo, base_commit)
|
|
|
|
subprocess.run(["git", "commit", "-am", "wip"], cwd=str(repo), capture_output=True)
|
|
|
|
log_files = subprocess.run(["git", "show", "--stat", "HEAD"], cwd=str(repo), capture_output=True, text=True).stdout
|
|
assert "tracked.txt" in log_files
|
|
assert "newfile.txt" not in log_files
|
|
|
|
|
|
def test_b7_6_cwd_independence(b7_repo, tmp_path):
|
|
repo, base_commit = b7_repo
|
|
(repo / "new.py").write_text("x = 1\n")
|
|
|
|
res_root = run_diff_collect(repo, base_commit)
|
|
assert res_root.returncode == 0
|
|
assert "new.py" in res_root.stdout
|
|
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
res_out = subprocess.run(
|
|
["bash", str(Path(__file__).parent.parent / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "diff_collect.sh"), str(repo), base_commit],
|
|
capture_output=True, text=True, cwd=str(outside)
|
|
)
|
|
assert res_out.returncode == 0
|
|
assert "new.py" in res_out.stdout
|
|
|
|
|
|
def test_b7_7_non_git_repo_fails_with_banner(tmp_path):
|
|
non_repo = tmp_path / "not_git"
|
|
non_repo.mkdir()
|
|
|
|
res = run_diff_collect(non_repo)
|
|
assert res.returncode == 2
|
|
assert "!!! CHANGE SET UNAVAILABLE !!!" in res.stdout
|
|
assert "reason:" in res.stdout
|
|
|
|
|
|
def test_b7_8_clean_repo_returns_no_changes(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
res = run_diff_collect(repo, base_commit)
|
|
assert res.returncode == 0
|
|
assert "(no changes since base commit)" in res.stdout
|
|
|
|
|
|
def test_b7_9_truncation_limits(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
large_text = "\n".join([f"line {i}" for i in range(100)])
|
|
(repo / "big.txt").write_text(large_text)
|
|
|
|
res = run_diff_collect(repo, base_commit, env={"MAM_DIFF_MAX_LINES": "50"})
|
|
assert res.returncode == 0
|
|
assert "!!! DIFF TRUNCATED !!!" in res.stdout
|
|
assert "over the review limit" in res.stdout
|
|
|
|
|
|
def test_b7_10_within_limits_untruncated(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
(repo / "small.txt").write_text("small\n")
|
|
|
|
res = run_diff_collect(repo, base_commit, env={"MAM_DIFF_MAX_LINES": "4000"})
|
|
assert res.returncode == 0
|
|
assert "!!! DIFF TRUNCATED !!!" not in res.stdout
|
|
assert "small" in res.stdout
|
|
|
|
|
|
def test_b7_11_filename_with_spaces(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
(repo / "two words.py").write_text("val = 42\n")
|
|
|
|
res = run_diff_collect(repo, base_commit)
|
|
assert res.returncode == 0
|
|
assert "two words.py" in res.stdout
|
|
|
|
|
|
def test_b7_12_run_loop_refuses_review_on_error(tmp_path):
|
|
non_repo = tmp_path / "non_repo_dir"
|
|
non_repo.mkdir()
|
|
|
|
script_path = Path(__file__).parent.parent / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "diff_collect.sh"
|
|
cmd_str = f"source '{script_path}' && mam_collect_changes_diff '{non_repo}'"
|
|
res = subprocess.run(["bash", "-c", cmd_str], capture_output=True, text=True)
|
|
|
|
assert res.returncode == 2
|
|
assert "!!! CHANGE SET UNAVAILABLE !!!" in res.stdout
|
|
assert "not a git repository" in res.stdout
|
|
|
|
|
|
def test_b7_13_single_collection_per_iteration():
|
|
run_loop_path = Path(__file__).parent.parent / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
|
|
content = run_loop_path.read_text()
|
|
|
|
assert "CHANGES_DIFF=$(mam_collect_changes_diff" in content
|
|
# Ensure mam_collect_changes_diff is outside for rev in REVIEWERS loop
|
|
outside_idx = content.find("CHANGES_DIFF=$(mam_collect_changes_diff")
|
|
loop_idx = content.find("for rev in \"${REVIEWERS[@]}\"; do")
|
|
assert outside_idx != -1
|
|
assert loop_idx != -1
|
|
assert outside_idx < loop_idx
|
|
|
|
|
|
def test_b7_14_bash_syntax_clean():
|
|
base = Path(__file__).parent.parent / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts"
|
|
diff_script = base / "diff_collect.sh"
|
|
loop_script = base / "run_loop.sh"
|
|
|
|
res1 = subprocess.run(["bash", "-n", str(diff_script)], capture_output=True)
|
|
res2 = subprocess.run(["bash", "-n", str(loop_script)], capture_output=True)
|
|
|
|
assert res1.returncode == 0, res1.stderr
|
|
assert res2.returncode == 0, res2.stderr
|
|
|
|
|
|
def test_b7_15_untracked_binary_summary(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
(repo / "data.bin").write_bytes(bytes(range(256)) * 10)
|
|
|
|
res = run_diff_collect(repo, base_commit)
|
|
assert res.returncode == 0
|
|
assert "Binary files" in res.stdout or "data.bin" in res.stdout
|
|
assert "GIT binary patch" not in res.stdout
|
|
|
|
|
|
def test_b7_16_nested_git_repo_file_included(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
nested = repo / "nested_repo"
|
|
nested.mkdir()
|
|
subprocess.run(["git", "init"], cwd=str(nested), capture_output=True)
|
|
(nested / "nested_file.py").write_text("print('nested')\n")
|
|
|
|
res = run_diff_collect(repo, base_commit)
|
|
assert res.returncode == 0
|
|
assert "nested_file.py" in res.stdout or "nested_repo" in res.stdout
|
|
|
|
|
|
def test_b7_17_nested_git_repo_gitignore_respected(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
nested = repo / "nested_repo"
|
|
nested.mkdir()
|
|
subprocess.run(["git", "init"], cwd=str(nested), capture_output=True)
|
|
subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=str(nested), capture_output=True)
|
|
subprocess.run(["git", "config", "user.name", "Test User"], cwd=str(nested), capture_output=True)
|
|
(nested / ".gitignore").write_text("nested_ignored.log\n")
|
|
subprocess.run(["git", "add", ".gitignore"], cwd=str(nested), capture_output=True)
|
|
subprocess.run(["git", "commit", "-m", "add ignore"], cwd=str(nested), capture_output=True)
|
|
|
|
(nested / "nested_ignored.log").write_text("ignored\n")
|
|
(nested / "nested_valid.py").write_text("valid\n")
|
|
|
|
res = run_diff_collect(repo, base_commit)
|
|
assert res.returncode == 0
|
|
assert "nested_ignored.log" not in res.stdout
|
|
|
|
|
|
def test_b7_18_nested_git_marker_announced(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
nested = repo / "nested_repo"
|
|
nested.mkdir()
|
|
subprocess.run(["git", "init"], cwd=str(nested), capture_output=True)
|
|
(nested / "foo.py").write_text("bar\n")
|
|
|
|
res = run_diff_collect(repo, base_commit)
|
|
assert res.returncode == 0
|
|
assert "!!! NOT EXPANDED IN THIS DIFF !!!" in res.stdout or "NESTED GIT REPOSITORY" in res.stdout
|
|
|
|
|
|
def test_b7_19_directory_symlink_announced(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
target_dir = repo / "target_dir"
|
|
target_dir.mkdir()
|
|
(target_dir / "target_file.txt").write_text("data\n")
|
|
|
|
symlink_path = repo / "link_dir"
|
|
try:
|
|
os.symlink("target_dir", str(symlink_path))
|
|
except Exception:
|
|
pytest.skip("Symlinks not supported")
|
|
|
|
res = run_diff_collect(repo, base_commit)
|
|
assert res.returncode == 0
|
|
assert "symlink" in res.stdout.lower() or "NOT EXPANDED" in res.stdout or "link_dir" in res.stdout
|
|
|
|
|
|
def test_b7_20_plain_file_diff_preserved(b7_repo):
|
|
repo, base_commit = b7_repo
|
|
(repo / "plain.txt").write_text("plain content\n")
|
|
|
|
res = run_diff_collect(repo, base_commit)
|
|
assert res.returncode == 0
|
|
assert "plain.txt" in res.stdout
|
|
assert "+plain content" in res.stdout
|