from __future__ import annotations

import copy
import json
import os
import shlex
import subprocess
from pathlib import Path
from unittest import mock

import pytest

from ops import gcp_leoclean_nosend_package as package
from ops import install_gcp_leoclean_nosend_release as installer
from ops import leoclean_nosend_service_control as control
from tests.test_gcp_leoclean_nosend_package import (
    build_push_receipt_fixture,
    docker_inspection_fixture,
    image_input_fixture,
    write_json,
)

# Representative production-shaped values from systemd's synthesized
# properties for a system service with LoadState=not-found. These are manager
# defaults, not evidence that a unit fragment, command, environment, runtime
# directory, or process exists.
PRODUCTION_SHAPED_NOT_FOUND_SECURITY = {
    "RuntimeDirectory": "",
    "RuntimeDirectoryMode": "0755",
    "RuntimeDirectoryPreserve": "no",
    "UMask": "0022",
    "NoNewPrivileges": "no",
    "CapabilityBoundingSet": (
        "cap_chown cap_dac_override cap_dac_read_search cap_fowner cap_fsetid cap_kill "
        "cap_setgid cap_setuid cap_setpcap cap_linux_immutable cap_net_bind_service "
        "cap_net_broadcast cap_net_admin cap_net_raw cap_ipc_lock cap_ipc_owner "
        "cap_sys_module cap_sys_rawio cap_sys_chroot cap_sys_ptrace cap_sys_pacct "
        "cap_sys_admin cap_sys_boot cap_sys_nice cap_sys_resource cap_sys_time "
        "cap_sys_tty_config cap_mknod cap_lease cap_audit_write cap_audit_control "
        "cap_setfcap cap_mac_override cap_mac_admin cap_syslog cap_wake_alarm "
        "cap_block_suspend cap_audit_read cap_perfmon cap_bpf cap_checkpoint_restore"
    ),
    "AmbientCapabilities": "",
    "PrivateDevices": "no",
    "PrivateTmp": "no",
    "ProtectSystem": "no",
    "ProtectHome": "no",
    "ProtectKernelTunables": "no",
    "ProtectKernelModules": "no",
    "ProtectKernelLogs": "no",
    "ProtectControlGroups": "no",
    "LockPersonality": "no",
    "RestrictSUIDSGID": "no",
    "RestrictRealtime": "no",
    "RestrictAddressFamilies": "",
    "SystemCallArchitectures": "",
}


def source_binding() -> dict[str, object]:
    stable: dict[str, object] = {
        "revision": "a" * 40,
        "files": [
            {"path": path, "sha256": package.sha256_file(installer.CONTRACT_ROOT / path)}
            for path in sorted(installer.SOURCE_PATHS)
        ],
    }
    return {**stable, "sha256": package.canonical_sha256(stable)}


def release_bundle(tmp_path: Path, digest_character: str) -> tuple[Path, dict[str, object], dict[str, object]]:
    fixture_root = tmp_path / f"input-{digest_character}"
    fixture_root.mkdir(mode=0o755)
    # Production image creation rejects any dirty reviewed source closure.
    # This synthetic fixture tests an uncommitted installer candidate without
    # weakening the production check itself.
    with mock.patch.object(package, "_verify_git_source_closure"):
        image_input = image_input_fixture(fixture_root)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{digest_character * 64}"
    config_digest = "sha256:" + chr(ord(digest_character) + 1) * 64
    image_inspection = docker_inspection_fixture(
        image_input,
        reference,
        config_digest=config_digest,
    )
    inspection = package.build_image_inspection(image_input, reference, image_inspection)
    release = package.build_release_descriptor(
        image_input,
        inspection,
        build_push_receipt_fixture(image_input, reference, config_digest=config_digest),
    )
    bundle = tmp_path / f"bundle-{digest_character}"
    bundle.mkdir(mode=0o755)
    write_json(bundle / "release.json", release, mode=0o644)
    (bundle / package.SERVICE).write_text(package.render_systemd_unit(release), encoding="utf-8")
    (bundle / package.SERVICE).chmod(0o644)
    return bundle, release, image_inspection


class FakeHost:
    def __init__(self, paths: installer.InstallPaths) -> None:
        self.paths = paths
        self.images: dict[str, dict[str, object]] = {}
        self.releases: dict[bytes, dict[str, object]] = {}
        self.active = False
        self.main_pid = 0
        self.n_restarts = 0
        self.invocation = ""
        self.container: dict[str, object] | None = None
        self.current_release: dict[str, object] | None = None
        self.loaded_release: dict[str, object] | None = None
        self.generation = 0
        self.restart_calls = 0
        self.stop_calls = 0
        self.fail_dropin_release: str | None = None
        self.race_release: str | None = None
        self.race_kind: str | None = None
        self.race_applied = False
        self.property_overrides: dict[str, str] = {}
        self.not_found_security_defaults: dict[str, str] = {}
        self.unstable_readbacks = False
        self.show_count = 0
        self.calls: list[list[str]] = []

    def add_release(self, release: dict[str, object], image_inspection: dict[str, object]) -> None:
        image = release["image"]
        assert isinstance(image, dict)
        self.images[str(image["reference"])] = copy.deepcopy(image_inspection)
        self.releases[package.render_systemd_unit(release).encode()] = release

    def _release_from_unit(self) -> dict[str, object]:
        value = self.releases.get(self.paths.unit.read_bytes())
        assert value is not None
        return value

    def _effective_fields(self) -> dict[str, list[str]]:
        assert self.loaded_release is not None
        return installer._rendered_service_fields(self.loaded_release)

    @staticmethod
    def _effective_exec(value: str) -> str:
        argv = shlex.split(value)
        return (
            f"{{ path={argv[0]} ; argv[]={value} ; ignore_errors=no ; "
            "start_time=[n/a] ; stop_time=[n/a] ; pid=0 ; code=(null) ; status=0/0 }"
        )

    def _write_process_environment(self) -> None:
        directory = self.paths.proc_root / str(self.main_pid)
        directory.mkdir(parents=True, exist_ok=True)
        (directory / "environ").write_bytes(f"PATH={package.SERVICE_PATH}\0INVOCATION_ID={self.invocation}\0".encode())
        fields = self._effective_fields()
        command = shlex.split(fields["ExecStart"][0])
        (directory / "exe").symlink_to(command[0])
        (directory / "cmdline").write_bytes(b"\0".join(item.encode() for item in command) + b"\0")
        (directory / "cgroup").write_text(f"0::/system.slice/{package.SERVICE}\n", encoding="utf-8")

    def _start(self) -> None:
        release = self.loaded_release
        assert release is not None
        image = release["image"]
        inspection = release["inspection"]
        assert isinstance(image, dict) and isinstance(inspection, dict)
        runtime = inspection["runtime_config"]
        assert isinstance(runtime, dict)
        self.generation += 1
        self.restart_calls += 1
        self.active = True
        self.main_pid = 1000 + self.generation
        self.invocation = f"{self.generation:032x}"
        container_id = f"{self.generation:064x}"
        labels = dict(image["labels"])
        labels.update(
            {
                "livingip.release.sha256": release["release_sha256"],
                "livingip.container.role": "leoclean-gcp-nosend",
            }
        )
        self.container = {
            "Id": container_id,
            "Name": f"/{package.CONTAINER_NAME}",
            "Image": image["config_digest"],
            "Path": runtime["entrypoint"][0],
            "Args": [*runtime["entrypoint"][1:], *runtime["cmd"]],
            "Config": {
                "Image": image["reference"],
                "Env": runtime["environment"],
                "Labels": labels,
            },
            "State": {"Running": True, "Pid": 2000 + self.generation, "Health": {"Status": "healthy"}},
            "HostConfig": {
                "ReadonlyRootfs": True,
                "NetworkMode": "bridge",
                "PidsLimit": 512,
                "CapDrop": ["ALL"],
                "CapAdd": ["CHOWN", "SETGID", "SETPCAP", "SETUID"],
                "SecurityOpt": ["no-new-privileges:true"],
            },
        }
        self.current_release = release
        self._write_process_environment()

    def _stop(self) -> None:
        self.stop_calls += 1
        self.active = False
        self.main_pid = 0
        self.invocation = ""
        self.container = None
        self.current_release = None

    def _show(self, arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
        self.show_count += 1
        loaded = self.loaded_release is not None
        release_sha = self.current_release.get("release_sha256") if self.current_release else None
        dropin = bool(self.active and release_sha == self.fail_dropin_release)
        value: dict[str, str] = {
            "LoadState": "loaded" if loaded else "not-found",
            "ActiveState": "active" if self.active else "inactive",
            "SubState": "running" if self.active else "dead",
            "MainPID": str(self.main_pid),
            "NRestarts": str(self.n_restarts),
            "InvocationID": self.invocation,
            "FragmentPath": str(self.paths.unit) if loaded else "",
            "DropInPaths": f"{self.paths.unit}.d/override.conf" if dropin else "",
            "EnvironmentFiles": "",
            "Environment": f"PATH={package.SERVICE_PATH}" if loaded else "",
            "ControlGroup": f"/system.slice/{package.SERVICE}" if self.active else "",
        }
        fields = self._effective_fields() if loaded else {}
        for property_name in installer.SERVICE_EXEC_PROPERTIES:
            value[property_name] = self._effective_exec(fields[property_name][0]) if loaded else ""
        for property_name in installer.SERVICE_SECURITY_PROPERTIES:
            value[property_name] = (
                fields[property_name][0] if loaded else self.not_found_security_defaults.get(property_name, "")
            )
        value.update(self.property_overrides)
        if self.unstable_readbacks:
            value["NRestarts"] = str(self.n_restarts + self.show_count % 2)
        payload = "".join(f"{name}={value[name]}\n" for name in installer.SERVICE_PROPERTIES).encode()
        return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"")

    def __call__(self, arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
        self.calls.append(arguments)
        if arguments[0] == installer.SYSTEMCTL:
            action = arguments[1]
            if action == "show":
                return self._show(arguments)
            if action == "daemon-reload":
                self.loaded_release = self._release_from_unit() if self.paths.unit.exists() else None
                return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"")
            if action == "reset-failed":
                self.n_restarts = 0
                return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"")
            if action == "restart":
                self._start()
                return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"")
            if action == "stop":
                self._stop()
                return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"")
            raise AssertionError(arguments)

        assert arguments[:3] == [
            package.DOCKER_BINARY,
            f"--host={package.DOCKER_HOST}",
            f"--config={self.paths.docker_config}",
        ]
        operation = arguments[3:]
        if operation[:2] == ["image", "inspect"]:
            image = self.images.get(operation[2])
            if image is None:
                return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=b"not found")
            return subprocess.CompletedProcess(arguments, 0, stdout=json.dumps([image]).encode(), stderr=b"")
        if operation[:2] == ["container", "inspect"]:
            if self.container is None or operation[2] not in {package.CONTAINER_NAME, self.container["Id"]}:
                return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=b"not found")
            observed = copy.deepcopy(self.container)
            release_sha = self.current_release.get("release_sha256") if self.current_release else None
            if operation[2] != package.CONTAINER_NAME and release_sha == self.race_release and not self.race_applied:
                self.race_applied = True
                if self.race_kind == "container_pid":
                    observed["State"]["Pid"] += 1
                elif self.race_kind == "service_restart":
                    self.main_pid += 1
                    self.n_restarts += 1
                    self.invocation = "f" * 32
                    self._write_process_environment()
            return subprocess.CompletedProcess(arguments, 0, stdout=json.dumps([observed]).encode(), stderr=b"")
        raise AssertionError(arguments)


def install_paths(tmp_path: Path) -> installer.InstallPaths:
    root = tmp_path / "root"
    root.mkdir(mode=0o700)
    return installer.InstallPaths.under(root, proc_root=root / "proc")


def test_dry_run_validates_bundle_without_creating_install_paths(tmp_path: Path) -> None:
    bundle, release, _image = release_bundle(tmp_path, "2")
    paths = install_paths(tmp_path)

    result = installer.install_release(
        bundle,
        paths=paths,
        source_binding=source_binding(),
        execute_restart=False,
    )

    assert result["mode"] == "dry_run"
    assert result["release"]["release_sha256"] == release["release_sha256"]
    assert not paths.unit.exists()
    assert not paths.state_dir.exists()


def test_dry_run_rejects_unbound_installer_source_without_creating_paths(tmp_path: Path) -> None:
    bundle, _release, _image = release_bundle(tmp_path, "2")
    paths = install_paths(tmp_path)
    binding = source_binding()
    binding["sha256"] = "f" * 64

    with pytest.raises(installer.InstallError, match="source binding"):
        installer.install_release(
            bundle,
            paths=paths,
            source_binding=binding,
            execute_restart=False,
        )

    assert not paths.unit.exists()
    assert not paths.state_dir.exists()


def test_alternate_repo_root_cannot_bind_the_executing_installer(tmp_path: Path) -> None:
    with pytest.raises(installer.InstallError, match="not the contract root"):
        installer.reviewed_source_binding(tmp_path)

    with pytest.raises(installer.InstallError, match="invalid arguments"):
        installer.parse_args(["--bundle", str(tmp_path), "--repo-root", str(tmp_path)])


def test_rehashed_alternate_source_bytes_cannot_bypass_actual_source_binding(tmp_path: Path) -> None:
    bundle, _release, _image = release_bundle(tmp_path, "2")
    paths = install_paths(tmp_path)
    binding = source_binding()
    files = copy.deepcopy(binding["files"])
    assert isinstance(files, list) and isinstance(files[0], dict)
    files[0]["sha256"] = "f" * 64
    stable = {"revision": binding["revision"], "files": files}
    forged = {**stable, "sha256": package.canonical_sha256(stable)}

    with pytest.raises(installer.InstallError, match="executing installer source differed"):
        installer.install_release(
            bundle,
            paths=paths,
            source_binding=forged,
            execute_restart=False,
        )


def test_disposable_root_install_is_digest_bound_idempotent_and_exact(tmp_path: Path) -> None:
    bundle, release, image = release_bundle(tmp_path, "2")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(release, image)

    receipt = installer.install_release(
        bundle,
        paths=paths,
        source_binding=source_binding(),
        runner=host,
        sleeper=lambda _seconds: None,
        execute_restart=True,
    )

    assert receipt["status"] == "pass"
    assert receipt["release_sha256"] == release["release_sha256"]
    assert host.restart_calls == 1
    assert paths.unit.read_text(encoding="utf-8") == package.render_systemd_unit(release)
    assert paths.unit.stat().st_mode & 0o777 == 0o644
    assert paths.helper.stat().st_mode & 0o777 == 0o755
    assert paths.release.stat().st_mode & 0o777 == 0o600
    assert paths.receipt.stat().st_mode & 0o777 == 0o600
    assert not list(paths.docker_config.iterdir())
    assert not list(paths.unit.parent.glob(f".{paths.unit.name}.*.tmp"))

    repeated = installer.install_release(
        bundle,
        paths=paths,
        source_binding=source_binding(),
        runner=host,
        sleeper=lambda _seconds: None,
        execute_restart=True,
    )

    assert repeated["result"] == "already_installed"
    assert host.restart_calls == 1


def test_first_install_accepts_production_shaped_systemd_not_found_defaults(tmp_path: Path) -> None:
    bundle, release, image = release_bundle(tmp_path, "2")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(release, image)
    host.not_found_security_defaults = dict(PRODUCTION_SHAPED_NOT_FOUND_SECURITY)

    receipt = installer.install_release(
        bundle,
        paths=paths,
        source_binding=source_binding(),
        runner=host,
        sleeper=lambda _seconds: None,
        execute_restart=True,
    )

    assert receipt["status"] == "pass"
    assert receipt["release_sha256"] == release["release_sha256"]
    assert host.restart_calls == 1


@pytest.mark.parametrize("property_name", installer.ABSENT_SERVICE_IDENTITY_PROPERTIES)
def test_first_install_rejects_identity_bearing_not_found_property(
    tmp_path: Path,
    property_name: str,
) -> None:
    bundle, release, image = release_bundle(tmp_path, "2")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(release, image)
    host.not_found_security_defaults = dict(PRODUCTION_SHAPED_NOT_FOUND_SECURITY)
    host.property_overrides[property_name] = "unexpected-authority"

    with pytest.raises(installer.InstallError, match="exactly absent"):
        installer.install_release(
            bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=host,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert host.restart_calls == 0
    assert not paths.unit.exists()
    assert not paths.state_dir.exists()


def test_execute_transaction_uses_exclusive_owner_only_lock(tmp_path: Path) -> None:
    bundle, release, image = release_bundle(tmp_path, "2")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(release, image)

    with installer._install_lock(paths):
        observed = paths.lock.stat(follow_symlinks=False)
        assert observed.st_uid == os.geteuid()
        assert observed.st_mode & 0o777 == 0o600
        with pytest.raises(installer.InstallError, match="holds the lock"):
            installer.install_release(
                bundle,
                paths=paths,
                source_binding=source_binding(),
                runner=host,
                sleeper=lambda _seconds: None,
                execute_restart=True,
            )

    assert not host.calls
    assert not paths.unit.exists()


def test_upgrade_stops_verified_prior_release_before_unit_replacement(tmp_path: Path) -> None:
    first_bundle, first_release, first_image = release_bundle(tmp_path, "2")
    second_bundle, second_release, second_image = release_bundle(tmp_path, "4")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(first_release, first_image)
    host.add_release(second_release, second_image)

    installer.install_release(
        first_bundle,
        paths=paths,
        source_binding=source_binding(),
        runner=host,
        sleeper=lambda _seconds: None,
        execute_restart=True,
    )
    start = len(host.calls)
    receipt = installer.install_release(
        second_bundle,
        paths=paths,
        source_binding=source_binding(),
        runner=host,
        sleeper=lambda _seconds: None,
        execute_restart=True,
    )

    actions = [call[1] for call in host.calls[start:] if call[0] == installer.SYSTEMCTL]
    assert actions.index("stop") < actions.index("daemon-reload") < actions.index("restart")
    assert receipt["release_sha256"] == second_release["release_sha256"]
    assert host.current_release == second_release
    assert host.stop_calls == 1


def test_stale_loaded_unit_is_rejected_before_trusting_old_exec_stop(tmp_path: Path) -> None:
    first_bundle, first_release, first_image = release_bundle(tmp_path, "2")
    second_bundle, second_release, second_image = release_bundle(tmp_path, "4")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(first_release, first_image)
    host.add_release(second_release, second_image)
    installer.install_release(
        first_bundle,
        paths=paths,
        source_binding=source_binding(),
        runner=host,
        sleeper=lambda _seconds: None,
        execute_restart=True,
    )
    host.loaded_release = second_release

    with pytest.raises(installer.InstallError, match="effective service command differed"):
        installer.install_release(
            second_bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=host,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert host.active
    assert host.current_release == first_release
    assert host.stop_calls == 0
    assert host.restart_calls == 1


def test_effective_security_drift_is_rejected_before_old_service_stop(tmp_path: Path) -> None:
    first_bundle, first_release, first_image = release_bundle(tmp_path, "2")
    second_bundle, second_release, second_image = release_bundle(tmp_path, "4")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(first_release, first_image)
    host.add_release(second_release, second_image)
    installer.install_release(
        first_bundle,
        paths=paths,
        source_binding=source_binding(),
        runner=host,
        sleeper=lambda _seconds: None,
        execute_restart=True,
    )
    host.property_overrides["NoNewPrivileges"] = "no"

    with pytest.raises(installer.InstallError, match="effective service security differed"):
        installer.install_release(
            second_bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=host,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert host.stop_calls == 0
    assert host.restart_calls == 1


@pytest.mark.parametrize(
    ("field", "error"),
    [
        ("exe", "supervising executable differed"),
        ("cmdline", "supervising command line differed"),
        ("cgroup", "cgroup differed from systemd"),
    ],
)
def test_supervising_process_drift_is_rejected_before_old_service_stop(
    tmp_path: Path,
    field: str,
    error: str,
) -> None:
    first_bundle, first_release, first_image = release_bundle(tmp_path, "2")
    second_bundle, second_release, second_image = release_bundle(tmp_path, "4")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(first_release, first_image)
    host.add_release(second_release, second_image)
    installer.install_release(
        first_bundle,
        paths=paths,
        source_binding=source_binding(),
        runner=host,
        sleeper=lambda _seconds: None,
        execute_restart=True,
    )
    process_root = paths.proc_root / str(host.main_pid)
    if field == "exe":
        executable = process_root / "exe"
        executable.unlink()
        executable.symlink_to("/bin/false")
    elif field == "cmdline":
        (process_root / "cmdline").write_bytes(b"/usr/bin/false\0")
    else:
        (process_root / "cgroup").write_text("0::/system.slice/other.service\n", encoding="utf-8")

    with pytest.raises(installer.InstallError, match=error):
        installer.install_release(
            second_bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=host,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert host.stop_calls == 0
    assert host.restart_calls == 1


def test_prior_managed_file_mode_drift_fails_before_service_mutation(tmp_path: Path) -> None:
    bundle, release, image = release_bundle(tmp_path, "2")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(release, image)
    installer.install_release(
        bundle,
        paths=paths,
        source_binding=source_binding(),
        runner=host,
        sleeper=lambda _seconds: None,
        execute_restart=True,
    )
    paths.receipt.chmod(0o644)

    with pytest.raises(installer.InstallError, match="file posture drifted"):
        installer.install_release(
            bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=host,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert host.active
    assert host.stop_calls == 0
    assert host.restart_calls == 1


def test_transitional_prior_service_posture_is_rejected_before_mutation(tmp_path: Path) -> None:
    first_bundle, first_release, first_image = release_bundle(tmp_path, "2")
    second_bundle, second_release, second_image = release_bundle(tmp_path, "4")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(first_release, first_image)
    host.add_release(second_release, second_image)
    installer.install_release(
        first_bundle,
        paths=paths,
        source_binding=source_binding(),
        runner=host,
        sleeper=lambda _seconds: None,
        execute_restart=True,
    )
    host.property_overrides.update({"ActiveState": "deactivating", "SubState": "stop-sigterm"})

    with pytest.raises(installer.InstallError, match="transitional or failed"):
        installer.install_release(
            second_bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=host,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert host.stop_calls == 0
    assert host.restart_calls == 1


def test_changing_prior_service_posture_fails_two_readback_gate(tmp_path: Path) -> None:
    first_bundle, first_release, first_image = release_bundle(tmp_path, "2")
    second_bundle, second_release, second_image = release_bundle(tmp_path, "4")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(first_release, first_image)
    host.add_release(second_release, second_image)
    installer.install_release(
        first_bundle,
        paths=paths,
        source_binding=source_binding(),
        runner=host,
        sleeper=lambda _seconds: None,
        execute_restart=True,
    )
    host.show_count = 0
    host.unstable_readbacks = True

    with pytest.raises(installer.InstallError, match="changed between readbacks"):
        installer.install_release(
            second_bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=host,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert host.stop_calls == 0
    assert host.restart_calls == 1


def test_wrong_local_image_is_rejected_without_service_restart(tmp_path: Path) -> None:
    bundle, release, image = release_bundle(tmp_path, "2")
    image["Id"] = "sha256:" + "9" * 64
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(release, image)

    with pytest.raises(installer.InstallError, match=r"finalized (release|inspection)"):
        installer.install_release(
            bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=host,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert host.restart_calls == 0
    assert not paths.unit.exists()
    assert not paths.state_dir.exists()


def test_preflight_reports_incomplete_directory_cleanup(tmp_path: Path) -> None:
    bundle, release, image = release_bundle(tmp_path, "2")
    image["Id"] = "sha256:" + "9" * 64
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(release, image)
    original_call = host.__call__

    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
        result = original_call(arguments)
        if arguments[3:5] == ["image", "inspect"]:
            (paths.state_dir / "unexpected").write_text("preserve", encoding="utf-8")
        return result

    with pytest.raises(installer.InstallError, match="cleanup was incomplete"):
        installer.install_release(
            bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=runner,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert host.restart_calls == 0
    assert (paths.state_dir / "unexpected").read_text(encoding="utf-8") == "preserve"


def test_post_rename_race_refuses_to_mutate_replaced_pathname(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    bundle, release, image = release_bundle(tmp_path, "2")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(release, image)
    original_fsync = installer._fsync_directory
    raced = False

    def replace_after_helper_rename(directory: Path) -> None:
        nonlocal raced
        if directory == paths.helper.parent and paths.helper.exists() and not raced:
            raced = True
            paths.helper.unlink()
            paths.helper.write_bytes(b"independent replacement\n")
            paths.helper.chmod(0o755)
            raise OSError("injected post-rename durability failure")
        original_fsync(directory)

    monkeypatch.setattr(installer, "_fsync_directory", replace_after_helper_rename)
    with pytest.raises(installer.InstallError, match="automatic rollback was incomplete"):
        installer.install_release(
            bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=host,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert raced
    assert paths.helper.read_bytes() == b"independent replacement\n"
    assert host.restart_calls == 0


@pytest.mark.parametrize("race_kind", ["container_pid", "service_restart"])
def test_pid_or_restart_race_fails_and_cleans_first_install(tmp_path: Path, race_kind: str) -> None:
    bundle, release, image = release_bundle(tmp_path, "2")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(release, image)
    host.race_release = str(release["release_sha256"])
    host.race_kind = race_kind

    with pytest.raises(installer.InstallError, match="changed during verification"):
        installer.install_release(
            bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=host,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert not host.active
    assert host.container is None
    assert not paths.unit.exists()
    assert not paths.helper.exists()
    assert not paths.state_dir.exists()


def test_restart_command_failure_after_side_effect_stops_candidate_and_cleans_install(tmp_path: Path) -> None:
    bundle, release, image = release_bundle(tmp_path, "2")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(release, image)
    failed_restart = False

    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
        nonlocal failed_restart
        result = host(arguments)
        if arguments[:2] == [installer.SYSTEMCTL, "restart"] and not failed_restart:
            failed_restart = True
            return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=b"restart failed")
        return result

    with pytest.raises(installer.InstallError, match="systemd restart failed"):
        installer.install_release(
            bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=runner,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert failed_restart
    assert not host.active
    assert host.stop_calls == 1
    assert not paths.unit.exists()
    assert not paths.state_dir.exists()


def test_later_dropin_failure_restores_previous_release_and_running_service(tmp_path: Path) -> None:
    first_bundle, first_release, first_image = release_bundle(tmp_path, "2")
    second_bundle, second_release, second_image = release_bundle(tmp_path, "4")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(first_release, first_image)
    host.add_release(second_release, second_image)

    installer.install_release(
        first_bundle,
        paths=paths,
        source_binding=source_binding(),
        runner=host,
        sleeper=lambda _seconds: None,
        execute_restart=True,
    )
    first_unit = paths.unit.read_bytes()
    first_state = paths.release.read_bytes()
    first_receipt = paths.receipt.read_bytes()
    host.fail_dropin_release = str(second_release["release_sha256"])

    with pytest.raises(installer.InstallError, match="later drop-in"):
        installer.install_release(
            second_bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=host,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert paths.unit.read_bytes() == first_unit
    assert paths.release.read_bytes() == first_state
    assert paths.receipt.read_bytes() == first_receipt
    assert host.active
    assert host.current_release == first_release
    assert host.stop_calls == 2
    assert host.restart_calls == 3
    assert not list(paths.state_dir.glob(".*.tmp"))


def test_failed_upgrade_restores_exact_inactive_prior_posture(tmp_path: Path) -> None:
    first_bundle, first_release, first_image = release_bundle(tmp_path, "2")
    second_bundle, second_release, second_image = release_bundle(tmp_path, "4")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(first_release, first_image)
    host.add_release(second_release, second_image)
    installer.install_release(
        first_bundle,
        paths=paths,
        source_binding=source_binding(),
        runner=host,
        sleeper=lambda _seconds: None,
        execute_restart=True,
    )
    host._stop()
    stop_calls = host.stop_calls
    restart_calls = host.restart_calls
    prior_n_restarts = host.n_restarts
    first_unit = paths.unit.read_bytes()
    host.fail_dropin_release = str(second_release["release_sha256"])

    with pytest.raises(installer.InstallError, match="effective later drop-in"):
        installer.install_release(
            second_bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=host,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert not host.active
    assert host.loaded_release == first_release
    assert host.n_restarts == prior_n_restarts
    assert host.stop_calls == stop_calls + 1
    assert host.restart_calls == restart_calls + 1
    assert paths.unit.read_bytes() == first_unit


def test_effective_process_environment_override_is_rejected_and_rolled_back(tmp_path: Path) -> None:
    bundle, release, image = release_bundle(tmp_path, "2")
    paths = install_paths(tmp_path)
    host = FakeHost(paths)
    host.add_release(release, image)
    original_start = host._start

    def start_with_override() -> None:
        original_start()
        environment = paths.proc_root / str(host.main_pid) / "environ"
        environment.write_bytes(f"PATH={package.SERVICE_PATH}\0DOCKER_HOST=tcp://unsafe.invalid:2375\0".encode())

    host._start = start_with_override  # type: ignore[method-assign]

    with pytest.raises(installer.InstallError, match="forbidden environment"):
        installer.install_release(
            bundle,
            paths=paths,
            source_binding=source_binding(),
            runner=host,
            sleeper=lambda _seconds: None,
            execute_restart=True,
        )

    assert not host.active
    assert not paths.state_dir.exists()


def test_control_helper_stops_only_trusted_full_container_id(tmp_path: Path) -> None:
    cid_parent = tmp_path / "runtime"
    cid_parent.mkdir(mode=0o700)
    cidfile = cid_parent / "container.cid"
    container_id = "c" * 64
    cidfile.write_text(container_id, encoding="ascii")
    cidfile.chmod(0o600)
    release_sha = "d" * 64
    image_reference = f"{package.IMAGE_REPOSITORY}@sha256:{'e' * 64}"
    container: dict[str, object] | None = {
        "Id": container_id,
        "Name": f"/{package.CONTAINER_NAME}",
        "Image": "sha256:" + "f" * 64,
        "Config": {
            "Image": image_reference,
            "Labels": {
                "livingip.release.sha256": release_sha,
                "livingip.container.role": "leoclean-gcp-nosend",
            },
        },
        "State": {"Running": True},
    }
    calls: list[list[str]] = []

    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
        nonlocal container
        calls.append(arguments)
        operation = arguments[3:]
        if operation[:2] == ["container", "inspect"]:
            if container is None or operation[2] not in {package.CONTAINER_NAME, container_id}:
                error = f"Error: No such container: {operation[2]}\n".encode()
                return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=error)
            return subprocess.CompletedProcess(arguments, 0, stdout=json.dumps([container]).encode(), stderr=b"")
        if operation[:2] == ["container", "stop"]:
            assert operation == ["container", "stop", "--time=30", container_id]
            container = None
            return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"")
        raise AssertionError(arguments)

    result = control.control_container(
        "stop",
        release_sha,
        image_reference,
        "sha256:" + "f" * 64,
        runner=runner,
        sleeper=lambda _seconds: None,
        cidfile=cidfile,
    )

    assert result == {"action": "stop", "status": "pass", "container": "stopped"}
    assert not cidfile.exists()
    stop_calls = [call[3:] for call in calls if call[3:5] == ["container", "stop"]]
    assert stop_calls == [["container", "stop", "--time=30", container_id]]


def test_control_helper_refuses_config_digest_mismatch_without_stopping(tmp_path: Path) -> None:
    cid_parent = tmp_path / "runtime"
    cid_parent.mkdir(mode=0o700)
    cidfile = cid_parent / "container.cid"
    container_id = "c" * 64
    cidfile.write_text(container_id, encoding="ascii")
    cidfile.chmod(0o600)
    release_sha = "d" * 64
    image_reference = f"{package.IMAGE_REPOSITORY}@sha256:{'e' * 64}"
    container = {
        "Id": container_id,
        "Name": f"/{package.CONTAINER_NAME}",
        "Image": "sha256:" + "1" * 64,
        "Config": {
            "Image": image_reference,
            "Labels": {
                "livingip.release.sha256": release_sha,
                "livingip.container.role": "leoclean-gcp-nosend",
            },
        },
        "State": {"Running": True},
    }
    calls: list[list[str]] = []

    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
        calls.append(arguments)
        return subprocess.CompletedProcess(arguments, 0, stdout=json.dumps([container]).encode(), stderr=b"")

    with pytest.raises(control.ControlError, match="identity did not match"):
        control.control_container(
            "stop",
            release_sha,
            image_reference,
            "sha256:" + "2" * 64,
            runner=runner,
            sleeper=lambda _seconds: None,
            cidfile=cidfile,
        )

    assert cidfile.exists()
    assert all(call[3:5] != ["container", "stop"] for call in calls)


def test_control_helper_retains_cid_when_name_rebinds_after_trusted_id_disappears(tmp_path: Path) -> None:
    cid_parent = tmp_path / "runtime"
    cid_parent.mkdir(mode=0o700)
    cidfile = cid_parent / "container.cid"
    container_id = "c" * 64
    rebound_id = "b" * 64
    cidfile.write_text(container_id, encoding="ascii")
    cidfile.chmod(0o600)
    release_sha = "d" * 64
    image_reference = f"{package.IMAGE_REPOSITORY}@sha256:{'e' * 64}"
    config_digest = "sha256:" + "f" * 64
    trusted = {
        "Id": container_id,
        "Name": f"/{package.CONTAINER_NAME}",
        "Image": config_digest,
        "Config": {
            "Image": image_reference,
            "Labels": {
                "livingip.release.sha256": release_sha,
                "livingip.container.role": "leoclean-gcp-nosend",
            },
        },
        "State": {"Running": True},
    }
    stopped = False
    rebound = False
    calls: list[list[str]] = []

    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
        nonlocal stopped, rebound
        calls.append(arguments)
        operation = arguments[3:]
        if operation[:2] == ["container", "stop"]:
            stopped = True
            return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"")
        if operation[:2] == ["container", "inspect"]:
            reference = operation[2]
            if reference == container_id and stopped:
                rebound = True
                error = f"Error: No such container: {reference}\n".encode()
                return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=error)
            if reference == package.CONTAINER_NAME and rebound:
                replacement = {"Id": rebound_id, "Name": f"/{package.CONTAINER_NAME}"}
                return subprocess.CompletedProcess(
                    arguments,
                    0,
                    stdout=json.dumps([replacement]).encode(),
                    stderr=b"",
                )
            return subprocess.CompletedProcess(arguments, 0, stdout=json.dumps([trusted]).encode(), stderr=b"")
        raise AssertionError(arguments)

    with pytest.raises(control.ControlError, match="name was rebound"):
        control.control_container(
            "stop",
            release_sha,
            image_reference,
            config_digest,
            runner=runner,
            sleeper=lambda _seconds: None,
            cidfile=cidfile,
        )

    assert stopped and rebound
    assert cidfile.read_text(encoding="ascii") == container_id
    assert [call[3:5] for call in calls].count(["container", "stop"]) == 1


def test_control_helper_refuses_untrusted_name_collision_without_mutation(tmp_path: Path) -> None:
    cid_parent = tmp_path / "runtime"
    cid_parent.mkdir(mode=0o700)
    calls: list[list[str]] = []

    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
        calls.append(arguments)
        operation = arguments[3:]
        if operation == ["container", "inspect", package.CONTAINER_NAME]:
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=json.dumps([{"Id": "f" * 64, "Name": f"/{package.CONTAINER_NAME}"}]).encode(),
                stderr=b"",
            )
        raise AssertionError(arguments)

    with pytest.raises(control.ControlError, match="without a trusted ID"):
        control.control_container(
            "prestart",
            "d" * 64,
            f"{package.IMAGE_REPOSITORY}@sha256:{'e' * 64}",
            "sha256:" + "f" * 64,
            runner=runner,
            sleeper=lambda _seconds: None,
            cidfile=cid_parent / "container.cid",
        )

    assert all(call[3:5] != ["container", "stop"] for call in calls)


def test_control_helper_does_not_treat_daemon_failure_as_container_absence(tmp_path: Path) -> None:
    cid_parent = tmp_path / "runtime"
    cid_parent.mkdir(mode=0o700)
    cidfile = cid_parent / "container.cid"

    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
        return subprocess.CompletedProcess(
            arguments,
            1,
            stdout=b"",
            stderr=b"Cannot connect to the Docker daemon\n",
        )

    with pytest.raises(control.ControlError, match="absence could not be proven"):
        control.control_container(
            "prestart",
            "d" * 64,
            f"{package.IMAGE_REPOSITORY}@sha256:{'e' * 64}",
            "sha256:" + "f" * 64,
            runner=runner,
            sleeper=lambda _seconds: None,
            cidfile=cidfile,
        )

    assert not cidfile.exists()


def test_release_bundle_symlink_is_rejected(tmp_path: Path) -> None:
    bundle, _release, _image = release_bundle(tmp_path, "2")
    alias = tmp_path / "bundle-alias"
    alias.symlink_to(bundle, target_is_directory=True)

    with pytest.raises(installer.InstallError, match="missing or unsafe"):
        installer.validate_release_bundle(alias)


def test_service_unit_binds_local_docker_authority_and_private_cidfile(tmp_path: Path) -> None:
    _bundle, release, _image = release_bundle(tmp_path, "2")
    unit = package.render_systemd_unit(release)

    assert f"--host={control.DOCKER_HOST}" in unit
    assert f"--config={control.DOCKER_CONFIG}" in unit
    assert f"--cidfile={control.CIDFILE}" in unit
    assert str(release["image"]["config_digest"]) in unit
    assert "UMask=0077" in unit
    assert "docker rm" not in unit
    assert "--force" not in unit