mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-24 12:42:49 +00:00
rcourtman
5f5f7ac7fc
The completion guard's CI mode now compares contract text against the evaluated commit (#2502), but it still decided which contracts a commit owed from HEAD's registry and contract index. CI walks a branch with HEAD at the tip (or at main after a merge), so a later commit could change an earlier commit's verdict. If a later commit added a Shared Boundaries reference to a file, an earlier commit that was complete when written got blocked. If a later commit dropped ownership of a file, an earlier commit that changed it without its contract passed. The pre-commit hook read both from the commit's own index, so CI and the hook could still disagree. repo_file_io and subsystem_contracts can now read from a revision as well as the working tree or index. A revision read uses only that commit's tree and fails closed rather than falling back to the disk. In diff-base mode the guard reads the registry and contract index from the evaluated commit, as does the Prettier-only inference that --commit enables. Replaying CI's per-commit loop over the last 120 commits on main gives the same verdicts as before.
300 lines
10 KiB
Python
300 lines
10 KiB
Python
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
|
|
from repo_file_io import strip_local_git_env
|
|
|
|
from subsystem_contracts import (
|
|
contract_reference_matches_path,
|
|
load_contract_index,
|
|
parse_contract_text,
|
|
referenced_contracts_for_path,
|
|
tracked_contract_paths,
|
|
tracked_contract_files,
|
|
)
|
|
|
|
|
|
class SubsystemContractsTest(unittest.TestCase):
|
|
def git(self, repo_root: Path, *args: str) -> subprocess.CompletedProcess:
|
|
# Scrub the full hook environment: with only GIT_INDEX_FILE removed, a
|
|
# pre-commit run from a linked worktree exports an absolute GIT_DIR and
|
|
# "git init" here re-initializes the REAL repository as bare.
|
|
env = strip_local_git_env(os.environ.copy())
|
|
return subprocess.run(
|
|
["git", *args],
|
|
cwd=repo_root,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
|
|
def test_parse_contract_text_extracts_metadata_and_path_references(self) -> None:
|
|
parsed, errors = parse_contract_text(
|
|
"docs/release-control/v6/internal/subsystems/example.md",
|
|
"""# Example Contract
|
|
|
|
## Contract Metadata
|
|
|
|
```json
|
|
{
|
|
"subsystem_id": "example"
|
|
}
|
|
```
|
|
|
|
## Purpose
|
|
|
|
Own example truth.
|
|
|
|
## Canonical Files
|
|
|
|
1. `internal/example/runtime.go`
|
|
2. `go.mod`
|
|
3. `go.sum`
|
|
|
|
## Shared Boundaries
|
|
|
|
1. `internal/shared/runtime.go` shared with `other-subsystem`: shared runtime boundary.
|
|
|
|
## Extension Points
|
|
|
|
1. Add new adapters in `internal/example/`
|
|
|
|
## Forbidden Paths
|
|
|
|
1. None
|
|
|
|
## Completion Obligations
|
|
|
|
1. Update contract
|
|
|
|
## Current State
|
|
|
|
Stable.
|
|
""",
|
|
)
|
|
|
|
self.assertEqual(errors, [])
|
|
self.assertEqual(parsed["metadata"]["subsystem_id"], "example")
|
|
self.assertEqual(
|
|
parsed["path_references"],
|
|
[
|
|
{
|
|
"heading": "## Canonical Files",
|
|
"path": "internal/example/runtime.go",
|
|
"line": 17,
|
|
"heading_line": 15,
|
|
},
|
|
{
|
|
"heading": "## Canonical Files",
|
|
"path": "go.mod",
|
|
"line": 18,
|
|
"heading_line": 15,
|
|
},
|
|
{
|
|
"heading": "## Canonical Files",
|
|
"path": "go.sum",
|
|
"line": 19,
|
|
"heading_line": 15,
|
|
},
|
|
{
|
|
"heading": "## Shared Boundaries",
|
|
"path": "internal/shared/runtime.go",
|
|
"line": 23,
|
|
"heading_line": 21,
|
|
},
|
|
{
|
|
"heading": "## Extension Points",
|
|
"path": "internal/example/",
|
|
"line": 27,
|
|
"heading_line": 25,
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_contract_reference_matches_path_supports_prefixes(self) -> None:
|
|
self.assertTrue(contract_reference_matches_path("internal/example/", "internal/example/runtime.go"))
|
|
self.assertFalse(contract_reference_matches_path("internal/example/", "internal/other/runtime.go"))
|
|
self.assertTrue(contract_reference_matches_path("internal/example/runtime.go", "internal/example/runtime.go"))
|
|
self.assertFalse(contract_reference_matches_path("internal/example/runtime.go", "internal/example/other.go"))
|
|
|
|
def test_referenced_contracts_for_path_returns_matching_subsystem(self) -> None:
|
|
contract_index = load_contract_index(
|
|
{
|
|
"docs/release-control/v6/internal/subsystems/example.md": """# Example Contract
|
|
|
|
## Contract Metadata
|
|
|
|
```json
|
|
{
|
|
"subsystem_id": "example"
|
|
}
|
|
```
|
|
|
|
## Purpose
|
|
|
|
Own example truth.
|
|
|
|
## Canonical Files
|
|
|
|
1. `internal/example/runtime.go`
|
|
2. `go.mod`
|
|
3. `go.sum`
|
|
|
|
## Shared Boundaries
|
|
|
|
1. `internal/shared/runtime.go` shared with `other-subsystem`: shared runtime boundary.
|
|
|
|
## Extension Points
|
|
|
|
1. Add new adapters in `internal/example/`
|
|
|
|
## Forbidden Paths
|
|
|
|
1. None
|
|
|
|
## Completion Obligations
|
|
|
|
1. Update contract
|
|
|
|
## Current State
|
|
|
|
Stable.
|
|
"""
|
|
}
|
|
)
|
|
|
|
matches = referenced_contracts_for_path("internal/example/runtime.go", contract_index)
|
|
self.assertEqual(len(matches), 1)
|
|
self.assertEqual(matches[0]["subsystem_id"], "example")
|
|
self.assertEqual(
|
|
matches[0]["matched_references"],
|
|
[
|
|
{
|
|
"heading": "## Canonical Files",
|
|
"path": "internal/example/runtime.go",
|
|
"line": 17,
|
|
"heading_line": 15,
|
|
},
|
|
{
|
|
"heading": "## Extension Points",
|
|
"path": "internal/example/",
|
|
"line": 27,
|
|
"heading_line": 25,
|
|
},
|
|
],
|
|
)
|
|
|
|
shared_matches = referenced_contracts_for_path("internal/shared/runtime.go", contract_index)
|
|
self.assertEqual(len(shared_matches), 1)
|
|
self.assertEqual(shared_matches[0]["subsystem_id"], "example")
|
|
self.assertEqual(
|
|
shared_matches[0]["matched_references"],
|
|
[
|
|
{
|
|
"heading": "## Shared Boundaries",
|
|
"path": "internal/shared/runtime.go",
|
|
"line": 23,
|
|
"heading_line": 21,
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_tracked_contract_files_can_read_staged_contract_content(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_root = Path(tmpdir)
|
|
contracts_dir = repo_root / "docs" / "release-control" / "v6" / "internal" / "subsystems"
|
|
contracts_dir.mkdir(parents=True, exist_ok=True)
|
|
contract_path = contracts_dir / "example.md"
|
|
contract_rel = "docs/release-control/v6/internal/subsystems/example.md"
|
|
|
|
self.git(repo_root, "init")
|
|
contract_path.write_text("# staged version\n", encoding="utf-8")
|
|
self.git(repo_root, "add", contract_rel)
|
|
contract_path.write_text("# working tree version\n", encoding="utf-8")
|
|
|
|
with (
|
|
patch("subsystem_contracts.REPO_ROOT", repo_root),
|
|
patch("subsystem_contracts.CONTRACTS_DIR", contracts_dir),
|
|
):
|
|
self.assertEqual(tracked_contract_files()[contract_rel], "# working tree version\n")
|
|
self.assertEqual(tracked_contract_paths(staged=True), [contract_rel])
|
|
self.assertEqual(tracked_contract_files(staged=True)[contract_rel], "# staged version\n")
|
|
|
|
def test_tracked_contract_paths_staged_ignores_untracked_working_tree_contracts(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_root = Path(tmpdir)
|
|
contracts_dir = repo_root / "docs" / "release-control" / "v6" / "internal" / "subsystems"
|
|
contracts_dir.mkdir(parents=True, exist_ok=True)
|
|
tracked_rel = "docs/release-control/v6/internal/subsystems/tracked.md"
|
|
untracked_rel = "docs/release-control/v6/internal/subsystems/untracked.md"
|
|
|
|
self.git(repo_root, "init")
|
|
(repo_root / tracked_rel).write_text("# tracked\n", encoding="utf-8")
|
|
self.git(repo_root, "add", tracked_rel)
|
|
(repo_root / untracked_rel).write_text("# untracked\n", encoding="utf-8")
|
|
|
|
with (
|
|
patch("subsystem_contracts.REPO_ROOT", repo_root),
|
|
patch("subsystem_contracts.CONTRACTS_DIR", contracts_dir),
|
|
):
|
|
self.assertEqual(
|
|
tracked_contract_paths(staged=True),
|
|
[tracked_rel],
|
|
)
|
|
self.assertEqual(
|
|
sorted(tracked_contract_paths()),
|
|
[tracked_rel, untracked_rel],
|
|
)
|
|
|
|
|
|
def test_tracked_contract_files_can_read_a_commits_contracts(self) -> None:
|
|
# A CI check of an earlier commit on a branch must see that commit's
|
|
# contracts, not the ones a later commit, the index or the working
|
|
# tree hold.
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_root = Path(tmpdir)
|
|
contracts_dir = repo_root / "docs" / "release-control" / "v6" / "internal" / "subsystems"
|
|
contracts_dir.mkdir(parents=True, exist_ok=True)
|
|
first_rel = "docs/release-control/v6/internal/subsystems/first.md"
|
|
later_rel = "docs/release-control/v6/internal/subsystems/later.md"
|
|
identity = ("-c", "user.name=Pulse Test", "-c", "user.email=pulse-test@example.invalid")
|
|
|
|
self.git(repo_root, "init")
|
|
(repo_root / first_rel).write_text("# committed first\n", encoding="utf-8")
|
|
self.git(repo_root, "add", first_rel)
|
|
self.git(repo_root, *identity, "commit", "-q", "-m", "first")
|
|
evaluated = self.git(repo_root, "rev-parse", "HEAD").stdout.strip()
|
|
(repo_root / first_rel).write_text("# committed later\n", encoding="utf-8")
|
|
(repo_root / later_rel).write_text("# later\n", encoding="utf-8")
|
|
self.git(repo_root, "add", first_rel, later_rel)
|
|
self.git(repo_root, *identity, "commit", "-q", "-m", "later")
|
|
(repo_root / first_rel).write_text("# staged\n", encoding="utf-8")
|
|
self.git(repo_root, "add", first_rel)
|
|
(repo_root / first_rel).write_text("# working tree\n", encoding="utf-8")
|
|
|
|
with (
|
|
patch("subsystem_contracts.REPO_ROOT", repo_root),
|
|
patch("subsystem_contracts.CONTRACTS_DIR", contracts_dir),
|
|
):
|
|
self.assertEqual(tracked_contract_paths(revision=evaluated), [first_rel])
|
|
self.assertEqual(
|
|
tracked_contract_files(revision=evaluated),
|
|
{first_rel: "# committed first\n"},
|
|
)
|
|
self.assertEqual(
|
|
tracked_contract_files(staged=True, revision="HEAD"),
|
|
{first_rel: "# committed later\n", later_rel: "# later\n"},
|
|
)
|
|
# Unlike the index read, a commit without contracts does not
|
|
# borrow the working tree's.
|
|
empty_tree = self.git(repo_root, "hash-object", "-t", "tree", "/dev/null").stdout.strip()
|
|
self.assertEqual(tracked_contract_files(revision=empty_tree), {})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|