rein-aharness/tests/test_repository_artifact.py

"""Actual Git export/import across separate workspaces; no gateway writes source."""

from __future__ import annotations

import os
import shutil
import subprocess
from types import SimpleNamespace
from unittest.mock import patch

import pytest

from rein_aharness.execution_cancel import ExecutionCancel, ExecutionCancelled
from rein_aharness.repository_artifact import (
    RepositoryArtifactError,
    RepositoryArtifactTransfer,
)
from rein_aharness.repository_grant import RepositoryGrant
from rein_aharness.repository_transaction import (
    RepositoryAcceptanceError,
    RepositoryTransaction,
)


def git(repo, *args):
    return (
        subprocess.check_output(
            ["git", "-C", str(repo), *args], stderr=subprocess.DEVNULL
        )
        .decode()
        .strip()
    )


def commit(repo, name="result.txt", content="accepted change\n"):
    path = repo / name
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content)
    git(repo, "add", name)
    git(
        repo,
        "-c",
        "user.name=Fixture",
        "-c",
        "user.email=fixture@example.invalid",
        "commit",
        "-qm",
        "bounded result",
    )
    return git(repo, "rev-parse", "HEAD")


def fixture(tmp_path):
    source = tmp_path / "source"
    source.mkdir()
    git(source, "init", "-q")
    commit(source, "README.md", "baseline\n")
    sandbox = tmp_path / "sandbox"
    git(tmp_path, "clone", "--quiet", "--no-local", str(source), str(sandbox))
    grant = RepositoryGrant.from_mapping(
        {
            "version": "1",
            "allowed_paths": ["result.txt"],
            "commit_count": {"min": 1, "max": 1},
            "publish": False,
        }
    )
    return source, sandbox, grant


class Transport:
    """Local transport double only; exporter and Git run unmodified."""

    kind = "local_namespace"

    def __init__(self, workspace):
        self.workspace = str(workspace)

    def run(self, argv, *, timeout):
        return subprocess.run(
            argv, cwd=self.workspace, capture_output=True, text=True, timeout=timeout
        )


def capture(transfer, sandbox, head):
    with patch(
        "glas_harness.transport.transport_from_sandbox", return_value=Transport(sandbox)
    ):
        transfer.capture(object(), SimpleNamespace(commit_sha=head))


def test_commit_survives_sandbox_teardown_and_import_preserves_identity(tmp_path):
    source, sandbox, grant = fixture(tmp_path)
    with RepositoryTransaction(source, state_dir=tmp_path / "state") as tx:
        baseline = tx.baseline.head
        transfer = RepositoryArtifactTransfer(tx, grant)
        head = commit(sandbox)
        capture(transfer, sandbox, head)
        assert git(source, "rev-parse", "HEAD") == baseline
        assert not (source / "result.txt").exists()
        shutil.rmtree(sandbox)
        result = transfer.import_after_teardown(head)
        assert result["head"] == head and result["imported"]
        assert git(source, "rev-parse", "HEAD") == head
        assert (source / "result.txt").read_text() == "accepted change\n"
        assert not git(source, "status", "--porcelain")
        assert tx.acceptance.changed_paths == ("result.txt",)
        assert "bundle" not in result


@pytest.mark.parametrize(
    "fault",
    [
        "ungranted",
        "two_commits",
        "wrong_reported_head",
        "dirty",
        "symlink",
        "expanded_size",
    ],
)
def test_invalid_artifact_cannot_change_source(tmp_path, fault):
    source, sandbox, grant = fixture(tmp_path)
    with RepositoryTransaction(source, state_dir=tmp_path / "state") as tx:
        baseline = tx.baseline.head
        transfer = RepositoryArtifactTransfer(tx, grant)
        head = commit(
            sandbox,
            "outside.txt" if fault == "ungranted" else "result.txt",
            "x" * (3 * 1024 * 1024) if fault == "expanded_size" else "result\n",
        )
        if fault == "two_commits":
            head = commit(sandbox, content="again\n")
        if fault == "dirty":
            (sandbox / "untracked.txt").write_text("dirty")
        if fault == "symlink":
            git(sandbox, "rm", "result.txt")
            os.symlink("/tmp/outside", sandbox / "result.txt")
            git(sandbox, "add", "result.txt")
            git(
                sandbox,
                "-c",
                "user.name=Fixture",
                "-c",
                "user.email=fixture@example.invalid",
                "commit",
                "--amend",
                "--no-edit",
                "-q",
            )
            head = git(sandbox, "rev-parse", "HEAD")
        with pytest.raises((RepositoryArtifactError, RepositoryAcceptanceError)):
            capture(transfer, sandbox, head)
            transfer.import_after_teardown(
                "f" * 40 if fault == "wrong_reported_head" else head
            )
        assert git(source, "rev-parse", "HEAD") == baseline
        assert not (source / "result.txt").exists()
        assert not (source / "outside.txt").exists()


def test_lease_loss_after_capture_refuses_import(tmp_path):
    source, sandbox, grant = fixture(tmp_path)
    cancel = ExecutionCancel()
    with RepositoryTransaction(source, state_dir=tmp_path / "state") as tx:
        transfer = RepositoryArtifactTransfer(tx, grant, cancel=cancel)
        head = commit(sandbox)
        capture(transfer, sandbox, head)
        cancel.cancel("lease-loss")
        with pytest.raises(ExecutionCancelled):
            transfer.import_after_teardown(head)
        assert git(source, "rev-parse", "HEAD") == tx.baseline.head


def test_changed_source_baseline_refuses_import(tmp_path):
    source, sandbox, grant = fixture(tmp_path)
    with RepositoryTransaction(source, state_dir=tmp_path / "state") as tx:
        transfer = RepositoryArtifactTransfer(tx, grant)
        head = commit(sandbox)
        capture(transfer, sandbox, head)
        human_head = commit(source, "human.txt", "new human work\n")
        with pytest.raises(RepositoryAcceptanceError, match="baseline-changed"):
            transfer.import_after_teardown(head)
        assert git(source, "rev-parse", "HEAD") == human_head
        assert (source / "human.txt").read_text() == "new human work\n"
        assert not (source / "result.txt").exists()


def test_gateway_failure_after_capture_does_not_import(tmp_path):
    from rein_aharness.glas_execution import execute_profiled_run
    from rein_aharness.ops_run_client import OpsRun, OpsRunConfig

    source, sandbox, grant = fixture(tmp_path)
    run = OpsRun(
        id="run",
        activity_definition_id="def",
        idempotency_key="key",
        target_repo=str(source),
        title="fixture",
        description="fixture",
        state="claimed",
        harness_profile_ref="harness.agent-dev-local@1.0.0",
        repository_grant=grant,
    )
    with RepositoryTransaction(source, state_dir=tmp_path / "state") as tx:
        head = commit(sandbox)

        def gateway(request, *, artifact_capture):
            with patch(
                "glas_harness.transport.transport_from_sandbox",
                return_value=Transport(sandbox),
            ):
                artifact_capture(object(), SimpleNamespace(commit_sha=head))
            return {
                "ok": False,
                "evidence": {
                    "commit_sha": head,
                    "outcome": "failed",
                    "sandbox_destroy": "failed",
                },
            }

        result = execute_profiled_run(
            run,
            OpsRunConfig(repo_roots=(str(tmp_path),)),
            report_to_hub=False,
            transaction=tx,
            request_factory=lambda **kw: kw,
            gateway=gateway,
        )
        assert not result["ok"]
        assert git(source, "rev-parse", "HEAD") == tx.baseline.head


def test_success_without_artifact_capture_refuses(tmp_path):
    from rein_aharness.glas_execution import execute_profiled_run
    from rein_aharness.ops_run_client import OpsRun, OpsRunConfig

    source, sandbox, grant = fixture(tmp_path)
    run = OpsRun(
        id="run",
        activity_definition_id="def",
        idempotency_key="key",
        target_repo=str(source),
        title="fixture",
        description="fixture",
        state="claimed",
        harness_profile_ref="harness.agent-dev-local@1.0.0",
        repository_grant=grant,
    )
    with RepositoryTransaction(source, state_dir=tmp_path / "state") as tx:
        with pytest.raises(RepositoryArtifactError, match="captured artifact"):
            execute_profiled_run(
                run,
                OpsRunConfig(repo_roots=(str(tmp_path),)),
                transaction=tx,
                request_factory=lambda **kw: kw,
                gateway=lambda *a, **kw: {
                    "ok": True,
                    "evidence": {
                        "commit_sha": "a" * 40,
                        "session_cleanup": "succeeded",
                        "sandbox_destroy": "succeeded",
                    },
                },
            )
        assert git(source, "rev-parse", "HEAD") == tx.baseline.head


def test_missing_repository_ownership_refuses():
    grant = RepositoryGrant.from_mapping(
        {
            "version": "1",
            "allowed_paths": ["result.txt"],
            "commit_count": {"min": 1, "max": 1},
            "publish": False,
        }
    )
    with pytest.raises(RepositoryArtifactError, match="active repository transaction"):
        RepositoryArtifactTransfer(None, grant)


@pytest.mark.parametrize("corruption", ["truncated", "substituted_head"])
def test_corrupted_capture_cannot_change_source(tmp_path, corruption):
    source, sandbox, grant = fixture(tmp_path)
    with RepositoryTransaction(source, state_dir=tmp_path / "state") as tx:
        transfer = RepositoryArtifactTransfer(tx, grant)
        head = commit(sandbox)
        capture(transfer, sandbox, head)
        if corruption == "truncated":
            transfer._bundle = transfer._bundle[:100]
        else:
            transfer._head = "f" * 40
        with pytest.raises(RepositoryArtifactError):
            transfer.import_after_teardown(transfer._head)
        assert git(source, "rev-parse", "HEAD") == tx.baseline.head
        assert not (source / "result.txt").exists()


def test_ignored_sandbox_cache_is_not_imported(tmp_path):
    source, sandbox, grant = fixture(tmp_path)
    commit(source, ".gitignore", "cache/\n")
    git(sandbox, "pull", "--ff-only", "--quiet")
    (sandbox / "cache").mkdir()
    (sandbox / "cache" / "runtime.txt").write_text("discarded runtime cache")
    with RepositoryTransaction(source, state_dir=tmp_path / "state") as tx:
        transfer = RepositoryArtifactTransfer(tx, grant)
        head = commit(sandbox)
        capture(transfer, sandbox, head)
        transfer.import_after_teardown(head)
        assert not (source / "cache").exists()
        assert tx.acceptance.changed_paths == ("result.txt",)