Pulse/scripts/release_control/subsystem_contracts_test.py

import os
import unittest
from unittest.mock import patch
from pathlib import Path
import subprocess
import tempfile

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:
        env = os.environ.copy()
        env.pop("GIT_INDEX_FILE", None)
        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],
                )


if __name__ == "__main__":
    unittest.main()