coulomb-social/tests/test_backup_install.py

"""Use disposable files only: prove source installation and recoverable rollback."""

import importlib.util
import io
import json
import os
import tarfile
from pathlib import Path

import pytest

spec = importlib.util.spec_from_file_location(
    "backup_install_test", Path(__file__).parents[1] / "scripts/backup_install.py")
install = importlib.util.module_from_spec(spec)
spec.loader.exec_module(install)


def bundle(change=None):
    contents = {name: ("# synthetic " + name + "\n").encode() for name in install.TARGETS}
    entries = [{"member": n, "target": t, "mode": "0644" if n.startswith("units/") else "0600",
                "sha256": install.sha(contents[n]), "bytes": len(contents[n])}
               for n, t in install.TARGETS.items()]
    manifest = {"schema": "coulomb.backup-operating-source-bundle.v1", "files": entries}
    if change:
        change(manifest, contents)
    raw_manifest = json.dumps(manifest).encode()
    stream = io.BytesIO()
    with tarfile.open(fileobj=stream, mode="w") as archive:
        for name, data in {**contents, "manifest.json": raw_manifest}.items():
            info = tarfile.TarInfo(name)
            info.size = len(data)
            archive.addfile(info, io.BytesIO(data))
    return stream.getvalue(), install.sha(raw_manifest)


@pytest.fixture
def prepared(tmp_path):
    raw, manifest_hash = bundle()
    manifest, contents = install.read_bundle(raw, install.sha(raw), manifest_hash)
    root = tmp_path / "root"
    root.mkdir(mode=0o700)
    for name in ("consumer/paired-backup.py", "consumer/snapshot-content-and-db.py",
                 "platform/primary_application_archive.py"):
        path = install.path_for(root, install.TARGETS[name])
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(b"# old paired worker\n" if "paired-backup" in name else contents[name])
        path.chmod(0o700)
    before = install.observe(root)
    return root, manifest, contents, before, tmp_path / "receipt"


def apply(prepared, reload=lambda: None, inactive=lambda: None):
    root, manifest, contents, before, directory = prepared
    return install.install(manifest, contents, before, directory, root=root,
                           reload_manager=reload, verify_inactive=inactive)


@pytest.mark.parametrize("mutation", ["target", "extra", "hash", "mode", "duplicate"])
def test_archive_rejects_unreviewed_targets_members_and_bytes(mutation):
    def change(manifest, contents):
        first = manifest["files"][0]
        if mutation == "target":
            first["target"] = "/etc/unrelated"
        elif mutation == "extra":
            contents["../escape"] = b"must not extract"
        elif mutation == "hash":
            first["sha256"] = "0" * 64
        elif mutation == "mode":
            first["mode"] = "0777"
        else:
            manifest["files"][-1] = dict(first)
    raw, manifest_hash = bundle(change)
    with pytest.raises(install.Refused):
        install.read_bundle(raw, install.sha(raw), manifest_hash)


def test_install_preserves_unchanged_inodes_and_finishes_inactive(prepared):
    calls = []
    record = apply(prepared, reload=lambda: calls.append("reload"),
                   inactive=lambda: calls.append("inactive"))
    root, manifest, contents, before, directory = prepared
    assert record["status"] == "sources_installed_units_inactive_verified"
    assert record["verified_targets"] == 19
    assert calls == ["inactive", "reload", "inactive"]
    assert record["timers_activated"] is False and record["capture_executed"] is False
    assert json.loads((directory / "receipt.json").read_text()) == record
    after = install.observe(root)
    for name in ("consumer/snapshot-content-and-db.py", "platform/primary_application_archive.py"):
        target = install.TARGETS[name]
        assert after[target] == before[target]
    assert after[install.TARGETS["consumer/paired-backup.py"]]["mode"] == 0o700
    for entry in manifest["files"]:
        if entry["target"]:
            assert install.path_for(root, entry["target"]).read_bytes() == contents[entry["member"]]


def test_active_units_and_native_drift_refuse_before_writes(prepared):
    def active():
        raise install.Refused("unit active")
    with pytest.raises(install.Refused, match="unit active"):
        apply(prepared, inactive=active)
    root, _, _, before, directory = prepared
    assert install.observe(root) == before and not directory.exists()
    path = install.path_for(root, install.TARGETS["consumer/paired-backup.py"])
    path.write_bytes(b"changed after review")
    with pytest.raises(install.Refused, match="source_drift"):
        apply(prepared)
    assert not directory.exists()


def test_shared_platform_helper_is_never_overwritten_on_drift(prepared):
    root, manifest, contents, _, directory = prepared
    path = install.path_for(root, install.TARGETS["platform/primary_application_archive.py"])
    path.write_bytes(b"other owner's version")
    before = install.observe(root)
    with pytest.raises(install.Refused, match="shared_platform"):
        apply((root, manifest, contents, before, directory))
    assert install.observe(root) == before and not directory.exists()


def test_failed_reload_has_durable_preimages_and_explicit_rollback(prepared):
    def failed():
        raise RuntimeError("synthetic reload failure")
    with pytest.raises(RuntimeError):
        apply(prepared, reload=failed)
    root, _, _, before, directory = prepared
    record = json.loads((directory / "receipt.json").read_text())
    assert record["status"] == "installation_requires_review"
    # Simulate a process exit during rollback after replacements, then resume
    # from its persisted expected restoration identities.
    with pytest.raises(RuntimeError):
        install.restore(root, directory, record, failed)
    record = json.loads((directory / "receipt.json").read_text())
    restored = install.restore(root, directory, record, lambda: None)
    assert restored["status"] == "prior_files_restored_verified"
    after = install.observe(root)
    for target, old in before.items():
        keys = ("exists", "sha256", "mode", "uid", "gid") if old["exists"] else ("exists",)
        assert all(after[target].get(k) == old.get(k) for k in keys)


def test_rollback_refuses_interference_before_undoing_any_file(prepared):
    record = apply(prepared)
    root, _, _, _, directory = prepared
    path = install.path_for(root, install.TARGETS["consumer/paired-backup.py"])
    path.write_bytes(b"concurrent owner change")
    before_rollback = install.observe(root)
    with pytest.raises(install.Refused, match="rollback_target_changed"):
        install.restore(root, directory, record, lambda: None)
    assert install.observe(root) == before_rollback


def test_corrupt_rollback_preimage_refuses_before_any_restoration(prepared):
    record = apply(prepared)
    root, _, _, _, directory = prepared
    entry = next(c for c in record["changes"] if c["before"]["exists"])
    (directory / entry["preimage"]).write_bytes(b"corrupt")
    before_rollback = install.observe(root)
    with pytest.raises(install.Refused, match="preimage_differs"):
        install.restore(root, directory, record, lambda: None)
    assert install.observe(root) == before_rollback


def test_corrupt_staging_reference_cannot_expand_rollback_scope(prepared):
    record = apply(prepared)
    root, _, _, _, directory = prepared
    record["changes"][0]["temporary"] = str(root / "unrelated")
    before_rollback = install.observe(root)
    with pytest.raises(install.Refused, match="outside_reviewed_parent"):
        install.restore(root, directory, record, lambda: None)
    assert install.observe(root) == before_rollback


@pytest.mark.parametrize("kind", ["symlink", "hardlink", "writable_parent"])
def test_unsafe_filesystem_entries_refuse(prepared, kind):
    root, _, _, _, directory = prepared
    path = install.path_for(root, install.TARGETS["consumer/paired-backup.py"])
    if kind == "symlink":
        path.unlink()
        path.symlink_to(root / "outside")
    elif kind == "hardlink":
        os.link(path, path.with_suffix(".duplicate"))
    else:
        path.parent.chmod(0o777)
    with pytest.raises((install.Refused, OSError)):
        apply(prepared)
    assert not directory.exists()