src/tests/test_artifact_promotion.py

from __future__ import annotations

import gzip
import json
from datetime import datetime, timezone
from pathlib import Path

import pytest

from ctx.core.wiki import artifact_promotion


def test_promote_staged_artifact_validates_replaces_and_records_metadata(
    tmp_path: Path,
) -> None:
    target = tmp_path / "artifact.txt"
    staged = tmp_path / "candidate.txt"
    target.write_bytes(b"old\n")
    staged.write_bytes(b"new\n")
    validated: list[Path] = []

    result = artifact_promotion.promote_staged_artifact(
        staged,
        target,
        validate=lambda path: validated.append(path),
        now=datetime(2026, 5, 4, tzinfo=timezone.utc),
    )

    metadata = json.loads(result.metadata_path.read_text(encoding="utf-8"))
    assert validated == [staged]
    assert target.read_bytes() == b"new\n"
    assert not staged.exists()
    assert metadata["status"] == "promoted"
    assert metadata["previous"]["exists"] is True
    assert metadata["previous"]["size"] == 4
    assert metadata["candidate"]["size"] == 4
    assert metadata["current"]["sha256"] == metadata["candidate"]["sha256"]
    assert metadata["last_good"] == metadata["previous"]
    assert metadata["rollback"]["available"] is True
    assert metadata["rollback"]["sha256"] == metadata["previous"]["sha256"]


def test_promote_staged_artifact_validation_failure_preserves_target(
    tmp_path: Path,
) -> None:
    target = tmp_path / "artifact.txt"
    staged = tmp_path / "candidate.txt"
    target.write_bytes(b"old\n")
    staged.write_bytes(b"new\n")

    def fail_validation(_path: Path) -> None:
        raise ValueError("candidate failed validation")

    with pytest.raises(ValueError, match="candidate failed validation"):
        artifact_promotion.promote_staged_artifact(
            staged,
            target,
            validate=fail_validation,
        )

    assert target.read_bytes() == b"old\n"
    assert staged.exists()
    assert not target.with_name("artifact.txt.promotion.json").exists()

    malformed_json = tmp_path / "broken.json"
    malformed_json.write_text("{", encoding="utf-8")
    with pytest.raises(ValueError, match="invalid JSON artifact"):
        artifact_promotion.validate_json_artifact(malformed_json)

    truncated_gzip = tmp_path / "broken.json.gz"
    truncated_gzip.write_bytes(b"\x1f\x8b")
    with pytest.raises(ValueError, match="invalid gzip JSON artifact"):
        artifact_promotion.validate_gzip_json_artifact(truncated_gzip)

    valid_gzip = tmp_path / "catalog.json.gz"
    with gzip.open(valid_gzip, "wt", encoding="utf-8") as fh:
        json.dump({"skills": []}, fh)
    artifact_promotion.validate_gzip_json_artifact(valid_gzip, required_keys=("skills",))


def test_promote_staged_artifact_replace_failure_preserves_target_and_last_good(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    target = tmp_path / "artifact.txt"
    staged = tmp_path / "candidate.txt"
    target.write_bytes(b"old\n")
    staged.write_bytes(b"new\n")

    def locked_replace(_src: Path, _dst: Path) -> None:
        raise PermissionError("locked")

    monkeypatch.setattr(artifact_promotion, "_replace_with_retry", locked_replace)

    with pytest.raises(PermissionError, match="locked"):
        artifact_promotion.promote_staged_artifact(staged, target)

    metadata = json.loads(
        target.with_name("artifact.txt.promotion.json").read_text(encoding="utf-8")
    )
    assert target.read_bytes() == b"old\n"
    assert staged.exists()
    assert metadata["status"] == "staged"
    assert metadata["previous"]["sha256"] != metadata["candidate"]["sha256"]
    assert metadata["last_good"] == metadata["previous"]
    assert metadata["rollback"]["available"] is True


def test_promote_staged_artifact_recovers_after_post_replace_crash(
    tmp_path: Path,
) -> None:
    target = tmp_path / "artifact.txt"
    staged = tmp_path / "candidate.txt"
    metadata_path = target.with_name("artifact.txt.promotion.json")
    target.write_bytes(b"new\n")
    candidate = artifact_promotion._snapshot(target)
    previous = {
        "path": str(target),
        "exists": True,
        "size": 4,
        "sha256": "old-sha",
        "mtime_ns": 1,
    }
    metadata_path.write_text(
        json.dumps({
            "schema_version": 1,
            "status": "staged",
            "target": str(target),
            "started_at": "2026-05-04T00:00:00+00:00",
            "previous": previous,
            "candidate": candidate,
        }),
        encoding="utf-8",
    )

    result = artifact_promotion.promote_staged_artifact(staged, target)

    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
    assert result.current["sha256"] == candidate["sha256"]
    assert metadata["status"] == "promoted"
    assert metadata["previous"] == previous
    assert metadata["candidate"]["sha256"] == candidate["sha256"]
    assert metadata["current"]["sha256"] == candidate["sha256"]