Validate repository transaction results
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b6f-7db1-7222-918b-e813a6bda38d
This commit is contained in:
parent
eaae4357eb
commit
2e16504e1f
3 changed files with 573 additions and 4 deletions
|
|
@ -12,6 +12,8 @@ import pytest
|
|||
from rein_aharness.repository_transaction import (
|
||||
DirtyRepositoryError,
|
||||
GitRepositoryError,
|
||||
RepositoryAcceptanceError,
|
||||
RepositoryAcceptancePolicy,
|
||||
RepositoryBusyError,
|
||||
RepositoryTransaction,
|
||||
)
|
||||
|
|
@ -215,6 +217,250 @@ def test_busy_error_exposes_bounded_owner_metadata(tmp_path: Path) -> None:
|
|||
assert len(excinfo.value.owner["correlation_id"]) == 200
|
||||
|
||||
|
||||
def test_acceptance_proves_one_descendant_commit_on_granted_paths(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
policy = RepositoryAcceptancePolicy(allowed_paths=("docs/",))
|
||||
|
||||
with RepositoryTransaction(repo, state_dir=tmp_path / "state") as tx:
|
||||
_commit(repo, "docs/result.md", "accepted\n")
|
||||
accepted = tx.validate_acceptance(policy)
|
||||
|
||||
assert len(accepted.commits) == 1
|
||||
assert accepted.changed_paths == ("docs/result.md",)
|
||||
evidence = tx.evidence()["acceptance"]
|
||||
assert evidence["accepted"] is True
|
||||
assert evidence["policy_id"] == policy.policy_id
|
||||
assert evidence["changed_paths"] == ["docs/result.md"]
|
||||
assert evidence["clean_post_state"] is True
|
||||
assert evidence["remote_refs_unchanged"] is True
|
||||
|
||||
|
||||
def test_acceptance_rejects_ungranted_extra_path(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
|
||||
with RepositoryTransaction(repo, state_dir=tmp_path / "state") as tx:
|
||||
_commit(repo, "UNRELATED.md", "not granted\n")
|
||||
with pytest.raises(RepositoryAcceptanceError) as excinfo:
|
||||
tx.validate_acceptance(
|
||||
RepositoryAcceptancePolicy(allowed_paths=("docs/",))
|
||||
)
|
||||
|
||||
assert excinfo.value.code == "path-not-granted"
|
||||
assert "UNRELATED.md" in excinfo.value.detail
|
||||
|
||||
|
||||
def test_acceptance_rejects_more_commits_than_policy_allows(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
|
||||
with RepositoryTransaction(repo, state_dir=tmp_path / "state") as tx:
|
||||
_commit(repo, "docs/one.md", "one\n")
|
||||
_commit(repo, "docs/two.md", "two\n")
|
||||
with pytest.raises(RepositoryAcceptanceError) as excinfo:
|
||||
tx.validate_acceptance(
|
||||
RepositoryAcceptancePolicy(allowed_paths=("docs/",))
|
||||
)
|
||||
|
||||
assert excinfo.value.code == "commit-count"
|
||||
assert "actual=2" in excinfo.value.detail
|
||||
|
||||
|
||||
def test_acceptance_rejects_dirty_post_state(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
|
||||
with RepositoryTransaction(repo, state_dir=tmp_path / "state") as tx:
|
||||
_commit(repo, "docs/result.md", "committed\n")
|
||||
(repo / "docs" / "result.md").write_text("left dirty\n", encoding="utf-8")
|
||||
with pytest.raises(RepositoryAcceptanceError) as excinfo:
|
||||
tx.validate_acceptance(
|
||||
RepositoryAcceptancePolicy(allowed_paths=("docs/",))
|
||||
)
|
||||
|
||||
assert excinfo.value.code == "dirty-post-state"
|
||||
|
||||
|
||||
def test_acceptance_rejects_remote_tracking_ref_movement(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
subprocess.run(
|
||||
["git", "update-ref", "refs/remotes/origin/main", "HEAD"],
|
||||
cwd=repo,
|
||||
check=True,
|
||||
)
|
||||
|
||||
with RepositoryTransaction(repo, state_dir=tmp_path / "state") as tx:
|
||||
_commit(repo, "docs/result.md", "committed\n")
|
||||
subprocess.run(
|
||||
["git", "update-ref", "refs/remotes/origin/main", "HEAD"],
|
||||
cwd=repo,
|
||||
check=True,
|
||||
)
|
||||
with pytest.raises(RepositoryAcceptanceError) as excinfo:
|
||||
tx.validate_acceptance(
|
||||
RepositoryAcceptancePolicy(allowed_paths=("docs/",))
|
||||
)
|
||||
|
||||
assert excinfo.value.code == "remote-refs-changed"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("metadata_kind", ["config", "hook"])
|
||||
def test_acceptance_rejects_protected_git_metadata_changes(
|
||||
tmp_path: Path,
|
||||
metadata_kind: str,
|
||||
) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
|
||||
with RepositoryTransaction(repo, state_dir=tmp_path / "state") as tx:
|
||||
_commit(repo, "docs/result.md", "committed\n")
|
||||
if metadata_kind == "config":
|
||||
subprocess.run(
|
||||
["git", "config", "rein-aharness.test", "changed"],
|
||||
cwd=repo,
|
||||
check=True,
|
||||
)
|
||||
else:
|
||||
hook = repo / ".git" / "hooks" / "pre-commit"
|
||||
hook.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
||||
with pytest.raises(RepositoryAcceptanceError) as excinfo:
|
||||
tx.validate_acceptance(
|
||||
RepositoryAcceptancePolicy(allowed_paths=("docs/",))
|
||||
)
|
||||
|
||||
assert excinfo.value.code == "git-metadata-changed"
|
||||
|
||||
|
||||
def test_acceptance_rejects_non_descendant_head(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
branch = subprocess.run(
|
||||
["git", "branch", "--show-current"],
|
||||
cwd=repo,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
).stdout.strip()
|
||||
|
||||
with RepositoryTransaction(repo, state_dir=tmp_path / "state") as tx:
|
||||
subprocess.run(["git", "checkout", "--orphan", "unrelated"], cwd=repo, check=True)
|
||||
subprocess.run(["git", "rm", "-qf", "README.md"], cwd=repo, check=True)
|
||||
(repo / "docs").mkdir()
|
||||
(repo / "docs" / "result.md").write_text("unrelated\n", encoding="utf-8")
|
||||
subprocess.run(["git", "add", "."], cwd=repo, check=True)
|
||||
subprocess.run(
|
||||
[
|
||||
"git",
|
||||
"-c",
|
||||
"user.email=test@example.invalid",
|
||||
"-c",
|
||||
"user.name=test",
|
||||
"commit",
|
||||
"-qm",
|
||||
"unrelated",
|
||||
],
|
||||
cwd=repo,
|
||||
check=True,
|
||||
)
|
||||
subprocess.run(["git", "checkout", "-qB", branch], cwd=repo, check=True)
|
||||
with pytest.raises(RepositoryAcceptanceError) as excinfo:
|
||||
tx.validate_acceptance(
|
||||
RepositoryAcceptancePolicy(allowed_paths=("docs/",))
|
||||
)
|
||||
|
||||
assert excinfo.value.code == "non-descendant-head"
|
||||
|
||||
|
||||
def test_acceptance_rejects_branch_change_and_unchanged_head(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
policy = RepositoryAcceptancePolicy(allowed_paths=("docs/",))
|
||||
|
||||
with RepositoryTransaction(repo, state_dir=tmp_path / "state") as tx:
|
||||
with pytest.raises(RepositoryAcceptanceError) as excinfo:
|
||||
tx.validate_acceptance(policy)
|
||||
assert excinfo.value.code == "head-unchanged"
|
||||
|
||||
with RepositoryTransaction(repo, state_dir=tmp_path / "state") as tx:
|
||||
subprocess.run(["git", "checkout", "-qb", "other"], cwd=repo, check=True)
|
||||
_commit(repo, "docs/result.md", "committed\n")
|
||||
with pytest.raises(RepositoryAcceptanceError) as excinfo:
|
||||
tx.validate_acceptance(policy)
|
||||
assert excinfo.value.code == "branch-changed"
|
||||
|
||||
|
||||
def test_acceptance_evidence_bounds_changed_path_list(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
policy = RepositoryAcceptancePolicy(
|
||||
allowed_paths=("docs/",),
|
||||
max_evidence_paths=2,
|
||||
)
|
||||
|
||||
with RepositoryTransaction(repo, state_dir=tmp_path / "state") as tx:
|
||||
_commit(repo, "docs/a.md", "a\n", extra={"docs/b.md": "b\n", "docs/c.md": "c\n"})
|
||||
evidence = tx.validate_acceptance(policy).evidence()
|
||||
|
||||
assert evidence["changed_path_count"] == 3
|
||||
assert len(evidence["changed_paths"]) == 2
|
||||
assert evidence["changed_paths_truncated"] is True
|
||||
assert len(evidence["changed_paths_digest"]) == 64
|
||||
|
||||
|
||||
def test_acceptance_path_glob_does_not_cross_directory_boundaries(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
|
||||
with RepositoryTransaction(repo, state_dir=tmp_path / "state") as tx:
|
||||
_commit(repo, "docs/nested/result.md", "nested\n")
|
||||
with pytest.raises(RepositoryAcceptanceError) as excinfo:
|
||||
tx.validate_acceptance(
|
||||
RepositoryAcceptancePolicy(allowed_paths=("docs/*.md",))
|
||||
)
|
||||
assert excinfo.value.code == "path-not-granted"
|
||||
|
||||
|
||||
def test_acceptance_double_star_path_glob_allows_nested_paths(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
|
||||
with RepositoryTransaction(repo, state_dir=tmp_path / "state") as tx:
|
||||
_commit(repo, "docs/nested/result.md", "nested\n")
|
||||
accepted = tx.validate_acceptance(
|
||||
RepositoryAcceptancePolicy(allowed_paths=("docs/**/*.md",))
|
||||
)
|
||||
assert accepted.changed_paths == ("docs/nested/result.md",)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"pattern",
|
||||
["/absolute", "../escape", ".git/config", "docs\\windows"],
|
||||
)
|
||||
def test_acceptance_policy_rejects_unsafe_path_patterns(pattern: str) -> None:
|
||||
with pytest.raises(ValueError, match="path pattern"):
|
||||
RepositoryAcceptancePolicy(allowed_paths=(pattern,))
|
||||
|
||||
|
||||
def _commit(
|
||||
repo: Path,
|
||||
relative_path: str,
|
||||
content: str,
|
||||
*,
|
||||
extra: dict[str, str] | None = None,
|
||||
) -> None:
|
||||
files = {relative_path: content, **(extra or {})}
|
||||
for name, value in files.items():
|
||||
path = repo / name
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(value, encoding="utf-8")
|
||||
subprocess.run(["git", "add", "."], cwd=repo, check=True)
|
||||
subprocess.run(
|
||||
[
|
||||
"git",
|
||||
"-c",
|
||||
"user.email=test@example.invalid",
|
||||
"-c",
|
||||
"user.name=test",
|
||||
"commit",
|
||||
"-qm",
|
||||
"task result",
|
||||
],
|
||||
cwd=repo,
|
||||
check=True,
|
||||
)
|
||||
|
||||
|
||||
def _status(repo: Path) -> str:
|
||||
return subprocess.run(
|
||||
["git", "status", "--porcelain=v2", "--untracked-files=all"],
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue