cc-ci/tests/unit/test_warmsnap.py

"""Unit tests for the WC3 snapshot helper's pure path/meta logic (runner/harness/warmsnap.py).

The docker/tar snapshot+restore round-trip is integration (proven live on cc-ci against a real
keycloak volume, W0.5). Here we cover the layout, meta read, and the has_snapshot completeness check.
"""

from __future__ import annotations

import json
import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "runner"))
from harness import warmsnap  # noqa: E402


def test_warm_root_env_override(monkeypatch):
    monkeypatch.setenv("CCCI_WARM_ROOT", "/tmp/ci-warm-test")
    assert warmsnap.warm_root() == "/tmp/ci-warm-test"
    assert warmsnap.app_dir("keycloak") == "/tmp/ci-warm-test/keycloak"
    # snapshot lives in a subdir so sibling state (last_good) survives the atomic swap.
    assert warmsnap.snap_dir("keycloak") == "/tmp/ci-warm-test/keycloak/snapshot"
    assert warmsnap.meta_path("keycloak") == "/tmp/ci-warm-test/keycloak/snapshot/meta.json"
    assert warmsnap.volumes_dir("keycloak") == "/tmp/ci-warm-test/keycloak/snapshot/volumes"


def test_warm_root_default(monkeypatch):
    monkeypatch.delenv("CCCI_WARM_ROOT", raising=False)
    assert warmsnap.warm_root() == "/var/lib/ci-warm"


def test_read_meta_absent(monkeypatch, tmp_path):
    monkeypatch.setenv("CCCI_WARM_ROOT", str(tmp_path))
    assert warmsnap.read_meta("keycloak") is None
    assert warmsnap.has_snapshot("keycloak") is False


def _write_snapshot(tmp_path, recipe, volumes):
    snapdir = tmp_path / recipe / "snapshot"
    (snapdir / "volumes").mkdir(parents=True)
    (snapdir / "meta.json").write_text(json.dumps({"recipe": recipe, "volumes": volumes}))
    for v in volumes:
        (snapdir / "volumes" / f"{v}.tar").write_bytes(b"fake")


def test_has_snapshot_complete(monkeypatch, tmp_path):
    monkeypatch.setenv("CCCI_WARM_ROOT", str(tmp_path))
    _write_snapshot(tmp_path, "keycloak", ["a_mariadb", "a_providers"])
    assert warmsnap.has_snapshot("keycloak") is True
    meta = warmsnap.read_meta("keycloak")
    assert meta["volumes"] == ["a_mariadb", "a_providers"]


def test_last_good_survives_sibling_of_snapshot(monkeypatch, tmp_path):
    # The reconciler's last_good lives in <recipe>/ (sibling of snapshot/) — a test that the layout
    # keeps them separate so a snapshot swap can't clobber last_good.
    monkeypatch.setenv("CCCI_WARM_ROOT", str(tmp_path))
    appdir = tmp_path / "keycloak"
    assert warmsnap.snap_dir("keycloak").startswith(str(appdir))
    assert os.path.dirname(warmsnap.snap_dir("keycloak")) == str(appdir)


def test_has_snapshot_incomplete_missing_tar(monkeypatch, tmp_path):
    monkeypatch.setenv("CCCI_WARM_ROOT", str(tmp_path))
    # meta lists two volumes but only one tar present -> incomplete -> not usable.
    snapdir = tmp_path / "keycloak" / "snapshot"
    (snapdir / "volumes").mkdir(parents=True)
    (snapdir / "meta.json").write_text(json.dumps({"recipe": "keycloak", "volumes": ["a", "b"]}))
    (snapdir / "volumes" / "a.tar").write_bytes(b"fake")
    assert warmsnap.has_snapshot("keycloak") is False


# --------------------------------------------------------------- F-redfix-4: slot ≠ recipe


def test_live_slot_is_the_recipe():
    assert warmsnap.live_slot("keycloak") == "keycloak"


def test_snapshot_refuses_to_clobber_another_domains_slot(monkeypatch, tmp_path):
    """F-redfix-4 regression: a slot holding domain A's known-good must not be overwritten by a
    snapshot of domain B. Before the fix this silently destroyed the other deployment's snapshot."""
    monkeypatch.setenv("CCCI_WARM_ROOT", str(tmp_path))
    snapdir = tmp_path / "keycloak" / "snapshot"
    (snapdir / "volumes").mkdir(parents=True)
    (snapdir / "meta.json").write_text(
        json.dumps({"slot": "keycloak", "domain": "warm-keycloak.ci.x", "volumes": []})
    )
    # Never reaches docker: the foreign-slot guard fires before _assert_undeployed.
    try:
        warmsnap.snapshot("keycloak", "warm-canon-keycloak.ci.x")
    except warmsnap.SnapshotError as e:
        assert "warm-keycloak.ci.x" in str(e) and "refusing to clobber" in str(e)
    else:
        raise AssertionError("snapshot() overwrote a foreign slot")
    # the original known-good is intact
    assert warmsnap.read_meta("keycloak")["domain"] == "warm-keycloak.ci.x"


def test_snapshot_may_reclaim_its_own_slot(monkeypatch, tmp_path):
    # Same domain → not foreign → the guard must not fire (it would break every re-snapshot).
    monkeypatch.setenv("CCCI_WARM_ROOT", str(tmp_path))
    snapdir = tmp_path / "keycloak" / "snapshot"
    (snapdir / "volumes").mkdir(parents=True)
    (snapdir / "meta.json").write_text(
        json.dumps({"slot": "keycloak", "domain": "warm-keycloak.ci.x", "volumes": []})
    )
    warmsnap._assert_slot_not_foreign("keycloak", "warm-keycloak.ci.x")  # no raise
    warmsnap._assert_slot_not_foreign("fresh-slot", "anything.ci.x")  # empty slot is free to claim


def test_restore_refuses_a_foreign_slot(monkeypatch, tmp_path):
    """restore() must reject a slot recorded against another domain BEFORE touching volumes."""
    monkeypatch.setenv("CCCI_WARM_ROOT", str(tmp_path))
    monkeypatch.setattr(warmsnap, "_assert_undeployed", lambda d: None)
    _write_snapshot(tmp_path, "keycloak", ["warm-keycloak_ci_x_mariadb"])
    snapdir = tmp_path / "keycloak" / "snapshot"
    meta = json.loads((snapdir / "meta.json").read_text())
    meta["domain"] = "warm-keycloak.ci.x"
    (snapdir / "meta.json").write_text(json.dumps(meta))
    called = []
    monkeypatch.setattr(warmsnap, "stack_volumes", lambda d: called.append(d) or [])
    try:
        warmsnap.restore("keycloak", "warm-canon-keycloak.ci.x")
    except warmsnap.SnapshotError as e:
        assert "refusing to clobber" in str(e)
    else:
        raise AssertionError("restore() accepted a foreign slot")
    assert called == [], "restore() must fail before inspecting the target stack's volumes"