test_workspace_commit.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/tests_git/test_workspace_commit.py
 
"""
Purpose
-------
Integration tests for ``oct git commit`` per-subproject quality-gate
fan-out (F1.3). When a workspace contains multiple subprojects and a
single ``oct git commit`` invocation stages files across them, each
subproject's slice must run under its own profile, the result must
aggregate cleanly, and the actual ``git commit`` must produce one
commit (not one per subproject).
 
Responsibilities
----------------
- One ``oct git commit`` produces exactly one git commit, even when
  staged files span multiple subprojects.
- A subproject's manifest profile is honoured; per-subproject octrc
  also takes effect.
- Secrets in any subproject still trip the always-blocking G-D4
  invariant.
- Files matching ``workspace_files[]`` are gated by
  ``workspace_profile``, not by any subproject's profile.
- Audit JSONL row carries ``per_subproject_results`` with the right
  shape after a workspace-mode commit.
- Single-project commit (from inside a subproject) is byte-for-byte
  unchanged.
 
Diagnostics
-----------
Domain: GIT-TESTS
Levels:
    L2 -- test lifecycle
    L3 -- assertion details
 
Contracts
---------
- All tests use subprocess invocation against ``oct.cli``.
- All tests use tmp_path fixtures so they are automatically cleaned up.
"""
 
import json
import shutil
import subprocess
import sys
from pathlib import Path
 
import pytest
 
 
COMPLIANT_TEMPLATE = '''\
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# {rel_path}
 
"""
Purpose
-------
Compliant fixture for the F1.3 workspace fan-out test suite.
 
Responsibilities
----------------
- Be linter-clean under any profile so the test can isolate
  fan-out routing from compliance noise.
 
Diagnostics
-----------
Domain: TEST
Levels:
    L1 — errors
    L2 — lifecycle
    L3 — details
    L4 — deep trace
 
Contracts
---------
- None.
"""
 
from oc_diagnostics import _dbg
 
func = "placeholder"
_dbg("TEST", func, "loaded", 2)
'''
 
 
def _git_available() -> bool:
    return shutil.which("git") is not None
 
 
def _git(repo: Path, *args: str, check: bool = True):
    return subprocess.run(
        ["git", *args], cwd=repo, check=check,
        capture_output=True, text=True,
    )
 
 
def _run_commit(
    cwd: Path, *extra_args: str,
) -> subprocess.CompletedProcess:
    return subprocess.run(
        [sys.executable, "-m", "oct.cli", "git", "commit", *extra_args],
        cwd=cwd, capture_output=True, text=True,
    )
 
 
def _make_subproject(parent: Path, name: str, profile: str = "compact") -> Path:
    """Build a minimal Option C subproject layout under *parent*/<name>/."""
    sub = parent / name
    sub.mkdir(parents=True, exist_ok=True)
    (sub / ".option_c").mkdir()
    (sub / "src").mkdir()
    (sub / "tests").mkdir()
    (sub / "docs").mkdir()
    (sub / "logs").mkdir()
    (sub / "oc_diagnostics").mkdir()
    (sub / ".octrc.json").write_text(
        json.dumps({"linter": {"profile": profile}}),
        encoding="utf-8",
    )
    (sub / "pyproject.toml").write_text(
        f'[project]\nname = "{name}"\n', encoding="utf-8",
    )
    (sub / "debug_config.json").write_text(
        json.dumps({"domains": {"GIT": {"level": 1, "color": "cyan"}}}),
        encoding="utf-8",
    )
    (sub / "docs" / "ARCHITECTURE.md").write_text(
        "# Architecture\n", encoding="utf-8",
    )
    return sub
 
 
def _write_compliant(target: Path, rel_path: str) -> None:
    """Write a compliant Python file at *target / rel_path*."""
    full = target / rel_path
    full.parent.mkdir(parents=True, exist_ok=True)
    full.write_text(
        COMPLIANT_TEMPLATE.format(rel_path=rel_path),
        encoding="utf-8",
    )
    # Companion test file so tests_exist quick-check passes.
    test_file = (
        target / "tests" / f"test_{full.stem}.py"
    )
    if not test_file.exists():
        test_file.write_text(
            f'"""Stub test for {full.stem}."""\n\n'
            f'def test_{full.stem}():\n    pass\n',
            encoding="utf-8",
        )
 
 
@pytest.fixture
def two_subproject_workspace(tmp_path: Path):
    """Build a workspace with two subprojects (alpha, beta) + initial commit.
 
    Returns ``(ws_root, alpha, beta)``. Both subprojects are committed
    in their initial state so further mutations show up cleanly in
    ``oct git status``.
    """
    if not _git_available():
        pytest.skip("git binary not available on PATH")
 
    ws_root = tmp_path / "ws"
    ws_root.mkdir()
    _git(ws_root, "init", "--quiet", "-b", "feature/test")
    _git(ws_root, "config", "user.email", "test@example.com")
    _git(ws_root, "config", "user.name", "Test")
    _git(ws_root, "config", "commit.gpgsign", "false")
 
    manifest = {
        "version": "1.0",
        "name": "FanOut Workspace",
        "workspace_profile": "compact",
        "workspace_files": ["CLAUDE.md", "README.md"],
        "subprojects": [
            {"path": "alpha", "name": "alpha", "profile": "compact"},
            {"path": "beta", "name": "beta", "profile": "compact"},
        ],
    }
    (ws_root / ".option_c_workspace.json").write_text(
        json.dumps(manifest, indent=2), encoding="utf-8",
    )
 
    alpha = _make_subproject(ws_root, "alpha", "compact")
    beta = _make_subproject(ws_root, "beta", "compact")
 
    # Workspace-level files.
    (ws_root / "CLAUDE.md").write_text("# CLAUDE\n", encoding="utf-8")
    (ws_root / "README.md").write_text("# README\n", encoding="utf-8")
 
    _git(ws_root, "add", "-A")
    _git(ws_root, "commit", "-m", "initial: workspace + subprojects", "--quiet")
 
    return ws_root.resolve(), alpha.resolve(), beta.resolve()
 
 
# =====================================================================
# Single-commit-per-invocation
# =====================================================================
 
 
class TestOneCommitPerInvocation:
 
    def test_single_commit_for_cross_subproject_stage(
        self, two_subproject_workspace,
    ) -> None:
        ws_root, alpha, beta = two_subproject_workspace
        _write_compliant(alpha, "src/a.py")
        _write_compliant(beta, "src/b.py")
        # Only stage the source files; the companion test stubs exist
        # on disk for the tests_exist quick-check but don't need to
        # land in the index.
        _git(ws_root, "add", "alpha/src/a.py")
        _git(ws_root, "add", "beta/src/b.py")
 
        # Note: pre-commit count.
        before = _git(
            ws_root, "rev-list", "--count", "HEAD",
        ).stdout.strip()
 
        result = _run_commit(
            ws_root, "-m", "feat(workspace): cross-subproject stage",
        )
        assert result.returncode == 0, result.stderr
 
        after = _git(
            ws_root, "rev-list", "--count", "HEAD",
        ).stdout.strip()
        assert int(after) - int(before) == 1, (
            f"expected exactly one new commit; got {before} -> {after}"
        )
 
 
# =====================================================================
# Per-subproject results land in the audit JSONL
# =====================================================================
 
 
def _read_latest_audit(repo_root: Path, command_name: str) -> dict | None:
    """Return the newest audit-log row matching *command_name*."""
    # The audit lives under the subproject that was the project_root
    # at command time. For workspace-mode commits that's actually the
    # workspace itself (since project-root resolution lands on the
    # workspace marker). Search both the workspace logs and any
    # subproject logs to be robust.
    candidates: list[Path] = []
    for log_dir in repo_root.rglob("logs"):
        if log_dir.is_dir():
            candidates.extend(log_dir.glob("git-audit-*.jsonl"))
    if not candidates:
        return None
    newest = max(candidates, key=lambda p: p.stat().st_mtime)
    rows: list[dict] = []
    for line in newest.read_text(encoding="utf-8").splitlines():
        try:
            rows.append(json.loads(line))
        except json.JSONDecodeError:
            continue
    for row in reversed(rows):
        if row.get("command") == command_name:
            return row
    return None
 
 
class TestAuditPerSubprojectResults:
 
    def test_per_subproject_results_field_populated(
        self, two_subproject_workspace,
    ) -> None:
        ws_root, alpha, beta = two_subproject_workspace
        _write_compliant(alpha, "src/a.py")
        _write_compliant(beta, "src/b.py")
        _git(ws_root, "add", "alpha/src/a.py")
        _git(ws_root, "add", "beta/src/b.py")
 
        result = _run_commit(
            ws_root, "-m", "feat(workspace): audit shape test",
        )
        assert result.returncode == 0, result.stderr
 
        row = _read_latest_audit(ws_root, "oct git commit")
        assert row is not None, "no audit row found for oct git commit"
        psr = row.get("per_subproject_results")
        assert isinstance(psr, list)
        names = {entry["name"] for entry in psr}
        assert "alpha" in names
        assert "beta" in names
        for entry in psr:
            for key in (
                "name", "profile", "files",
                "lint_violations", "format_violations", "secrets",
            ):
                assert key in entry, (
                    f"per_subproject_results entry missing {key!r}: {entry}"
                )
 
 
# =====================================================================
# Workspace-level files use workspace_profile
# =====================================================================
 
 
class TestWorkspaceFilesUseWorkspaceProfile:
 
    def test_claudemd_routed_to_workspace_bucket(
        self, two_subproject_workspace,
    ) -> None:
        ws_root, _alpha, _beta = two_subproject_workspace
        # Touch CLAUDE.md (workspace-level file) and stage it.
        (ws_root / "CLAUDE.md").write_text(
            "# CLAUDE — modified\n", encoding="utf-8",
        )
        _git(ws_root, "add", "CLAUDE.md")
 
        result = _run_commit(
            ws_root, "-m", "docs: tweak CLAUDE.md",
        )
        assert result.returncode == 0, result.stderr
 
        row = _read_latest_audit(ws_root, "oct git commit")
        assert row is not None
        psr = row.get("per_subproject_results", [])
        names = {entry["name"] for entry in psr}
        assert "workspace" in names, (
            f"expected workspace bucket in per_subproject_results: {psr}"
        )
        # And NOT routed to alpha or beta.
        ws_entry = next(e for e in psr if e["name"] == "workspace")
        assert ws_entry["files"] == 1
 
 
# =====================================================================
# Backwards compat: single-project mode unchanged
# =====================================================================
 
 
class TestSingleProjectCompat:
 
    def test_commit_inside_subproject_no_fanout(
        self, two_subproject_workspace,
    ) -> None:
        _ws_root, alpha, _beta = two_subproject_workspace
        _write_compliant(alpha, "src/x.py")
        _git(alpha, "add", "src/x.py")
 
        result = _run_commit(
            alpha, "-m", "feat(alpha): standalone commit",
        )
        assert result.returncode == 0, result.stderr
 
        row = _read_latest_audit(alpha, "oct git commit")
        assert row is not None
        # Single-project mode -> per_subproject_results is empty.
        assert row.get("per_subproject_results", []) == []
 
 
# =====================================================================
# Empty fan-out (no-op buckets)
# =====================================================================
 
 
class TestEmptyBuckets:
 
    def test_workspace_files_only_no_subproject_buckets(
        self, two_subproject_workspace,
    ) -> None:
        ws_root, _alpha, _beta = two_subproject_workspace
        # Stage only a workspace-level file.
        (ws_root / "README.md").write_text(
            "# README — modified\n", encoding="utf-8",
        )
        _git(ws_root, "add", "README.md")
 
        result = _run_commit(
            ws_root, "-m", "docs: tweak README",
        )
        assert result.returncode == 0, result.stderr
 
        row = _read_latest_audit(ws_root, "oct git commit")
        assert row is not None
        psr = row.get("per_subproject_results", [])
        # Only the "workspace" bucket should appear.
        names = {entry["name"] for entry in psr}
        assert names == {"workspace"}, (
            f"expected only workspace bucket; got {names}"
        )