import json
import os
from pathlib import Path
import subprocess

import pytest


MODULE = Path(__file__).resolve().parents[1]
HELPER = MODULE / "lib" / "notifier-retirement.py"
HOME_FILES = MODULE / "lib" / "user-home-files.py"
SERVICES = ("mem-pressure-notify.service", "disk-fill-notify.service")


def write_executable(path: Path, content: str):
    path.write_text(content)
    path.chmod(0o755)


def fixture(
    tmp_path: Path,
    *,
    fail_action="",
    fail_service="",
    fail_once=True,
    not_found=(),
    restart_on_reload="",
    failed_service="",
):
    home = tmp_path / "home"
    state = tmp_path / "state"
    fakebin = tmp_path / "bin"
    fakebin.mkdir()
    calls = tmp_path / "calls"
    service_state = tmp_path / "services"
    service_state.mkdir()
    failures = tmp_path / "failures"
    failures.mkdir()

    for service in SERVICES:
        if service not in not_found:
            (service_state / f"{service}.load").write_text("loaded")
            (service_state / f"{service}.active").write_text("active")
            (service_state / f"{service}.enabled").write_text("enabled")

    write_executable(
        fakebin / "runuser",
        """#!/bin/sh
while [ "$#" -gt 0 ] && [ "$1" != systemctl ]; do shift; done
exec "$@"
""",
    )
    write_executable(
        fakebin / "systemctl",
        """#!/bin/sh
set -eu
[ "${1:-}" = --user ] && shift
printf '%s\n' "$*" >> "$CALLS"
action=${1:-}; shift || true
service=${2:-${1:-}}
if [ "$action" = "${FAIL_ACTION:-}" ] && [ "$service" = "${FAIL_SERVICE:-}" ]; then
  marker="$FAILURES/$action-$service"
  if [ "${FAIL_ONCE:-1}" != 1 ] || [ ! -e "$marker" ]; then
    touch "$marker"
    exit 1
  fi
fi
case "$action" in
  disable)
    [ "${1:-}" = --now ] && shift
    service=$1
    rm -f "$SERVICE_STATE/$service.active" "$SERVICE_STATE/$service.enabled"
    ;;
  enable) touch "$SERVICE_STATE/$1.enabled" ;;
  start) touch "$SERVICE_STATE/$1.active" ;;
  stop) rm -f "$SERVICE_STATE/$1.active" ;;
  show)
    service=$1; property=$3
    case "$property" in
      LoadState) [ -e "$SERVICE_STATE/$service.load" ] && printf 'loaded\n' || printf 'not-found\n' ;;
      ActiveState)
        if [ "$service" = "${FAILED_SERVICE:-}" ]; then printf 'failed\n'
        elif [ -e "$SERVICE_STATE/$service.active" ]; then printf 'active\n'
        else printf 'inactive\n'; fi
        ;;
      UnitFileState)
        if [ ! -e "$SERVICE_STATE/$service.load" ]; then printf 'not-found\n'
        elif [ -e "$SERVICE_STATE/$service.enabled" ]; then printf 'enabled\n'
        else printf 'disabled\n'; fi
        ;;
      *) exit 64 ;;
    esac
    ;;
  daemon-reload)
    for unit in mem-pressure-notify.service disk-fill-notify.service; do
      if [ -e "$UNIT_DIR/$unit" ] || [ -L "$UNIT_DIR/$unit" ]; then
        touch "$SERVICE_STATE/$unit.load"
      else
        rm -f "$SERVICE_STATE/$unit.load" "$SERVICE_STATE/$unit.active" "$SERVICE_STATE/$unit.enabled"
      fi
    done
    if [ -n "${RESTART_ON_RELOAD:-}" ]; then
      touch "$SERVICE_STATE/$RESTART_ON_RELOAD.load" "$SERVICE_STATE/$RESTART_ON_RELOAD.active"
    fi
    ;;
  *) exit 64 ;;
esac
""",
    )

    env = os.environ.copy()
    env.update(
        PATH=f"{fakebin}:{env['PATH']}",
        CALLS=str(calls),
        SERVICE_STATE=str(service_state),
        UNIT_DIR=str(home / ".config/systemd/user"),
        FAILURES=str(failures),
        FAIL_ACTION=fail_action,
        FAIL_SERVICE=fail_service,
        FAIL_ONCE="1" if fail_once else "0",
        RESTART_ON_RELOAD=restart_on_reload,
        FAILED_SERVICE=failed_service,
    )
    return home, state, calls, service_state, env


def populate_paths(home: Path):
    for path in (
        home / ".local/bin/mem-pressure-notify",
        home / ".local/bin/disk-fill-notify",
        home / ".config/systemd/user/mem-pressure-notify.service",
    ):
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(path.name)
    dangling = home / ".config/systemd/user/disk-fill-notify.service"
    dangling.symlink_to(home / "missing-unit")


def run_helper(action, home, state, env, *, check=True, dry_run=False):
    command = [
        "python3",
        str(HELPER),
        action,
        "--user",
        "user",
        "--uid",
        "1000",
        "--home",
        str(home),
        "--state-dir",
        str(state),
    ]
    if dry_run:
        command.append("--dry-run")
    return subprocess.run(
        command,
        check=check,
        capture_output=True,
        env=env,
        text=True,
    )


def assert_original_paths(home):
    assert (home / ".local/bin/mem-pressure-notify").read_text() == "mem-pressure-notify"
    assert (home / ".local/bin/disk-fill-notify").read_text() == "disk-fill-notify"
    assert (home / ".config/systemd/user/disk-fill-notify.service").is_symlink()


def assert_original_services(service_state):
    for service in SERVICES:
        assert (service_state / f"{service}.active").exists()
        assert (service_state / f"{service}.enabled").exists()


def generation(state):
    current = state / "retired/current"
    return current.parent / current.readlink()


def test_retires_both_services_and_all_paths_including_dangling_link(tmp_path):
    home, state, calls, service_state, env = fixture(tmp_path)
    populate_paths(home)

    result = run_helper("retire", home, state, env, check=False)
    assert result.returncode == 0, result.stderr

    for service in SERVICES:
        assert f"disable --now {service}" in calls.read_text()
        assert not (service_state / f"{service}.active").exists()
        assert not (service_state / f"{service}.enabled").exists()
    for name in ("mem-pressure-notify", "disk-fill-notify"):
        assert not (home / ".local/bin" / name).exists()
        assert not (home / ".config/systemd/user" / f"{name}.service").is_symlink()
        assert (generation(state) / name).exists()
        archived = generation(state) / f"{name}.service"
        assert archived.exists() or archived.is_symlink()


def test_second_disable_failure_rolls_back_first_service_and_files(tmp_path):
    home, state, _, service_state, env = fixture(
        tmp_path,
        fail_action="disable",
        fail_service="disk-fill-notify.service",
    )
    populate_paths(home)

    result = run_helper("retire", home, state, env, check=False)

    assert result.returncode != 0
    assert_original_services(service_state)
    assert_original_paths(home)
    assert not (state / "retired/current").exists()
    assert not list((state / "retired").glob("generation-*"))


def test_failed_retirement_does_not_publish_recovery_generation(tmp_path):
    home, state, _, _, env = fixture(
        tmp_path,
        fail_action="disable",
        fail_service="disk-fill-notify.service",
    )
    populate_paths(home)

    run_helper("retire", home, state, env, check=False)

    assert not (state / "retired/current").exists()
    assert not list((state / "retired").glob("generation-*"))


def test_failed_service_state_aborts_before_mutation(tmp_path):
    home, state, calls, service_state, env = fixture(
        tmp_path,
        failed_service="disk-fill-notify.service",
    )
    populate_paths(home)
    (service_state / "disk-fill-notify.service.active").unlink()

    result = run_helper("retire", home, state, env, check=False)

    assert result.returncode != 0
    assert "unsupported ActiveState" in result.stderr
    assert "disable --now" not in calls.read_text()
    assert_original_paths(home)


def test_real_not_found_unit_is_accepted_without_disable(tmp_path):
    home, state, calls, _, env = fixture(
        tmp_path,
        not_found=("disk-fill-notify.service",),
    )
    populate_paths(home)

    run_helper("retire", home, state, env)

    assert "disable --now disk-fill-notify.service" not in calls.read_text()
    receipt = json.loads((generation(state) / "notifier-state.json").read_text())
    assert receipt["services"]["disk-fill-notify.service"] == {
        "load": "not-found",
        "active": "inactive",
        "enabled": "not-found",
    }


def test_rerun_preserves_first_generation_and_receipt(tmp_path):
    home, state, _, _, env = fixture(tmp_path)
    populate_paths(home)
    run_helper("retire", home, state, env)
    first_link = (state / "retired/current").readlink()
    first_receipt = (generation(state) / "notifier-state.json").read_bytes()

    run_helper("retire", home, state, env)

    assert (state / "retired/current").readlink() == first_link
    assert (generation(state) / "notifier-state.json").read_bytes() == first_receipt


def test_daemon_reload_failure_rolls_back_files_and_services(tmp_path):
    home, state, _, service_state, env = fixture(
        tmp_path,
        fail_action="daemon-reload",
        fail_once=True,
    )
    populate_paths(home)

    result = run_helper("retire", home, state, env, check=False)

    assert result.returncode != 0
    assert_original_paths(home)
    assert_original_services(service_state)
    assert not (state / "retired/current").exists()
    assert not list((state / "retired").glob("generation-*"))


def test_post_reload_restart_is_detected_and_rolled_back(tmp_path):
    home, state, _, service_state, env = fixture(
        tmp_path,
        restart_on_reload="disk-fill-notify.service",
    )
    populate_paths(home)

    result = run_helper("retire", home, state, env, check=False)

    assert result.returncode != 0
    assert "obsolete service remains after retirement" in result.stderr
    assert_original_paths(home)
    assert_original_services(service_state)


def test_restore_start_failure_restores_prior_files_and_service_states(tmp_path):
    home, state, _, service_state, env = fixture(tmp_path)
    populate_paths(home)
    run_helper("retire", home, state, env)
    replacements = {
        home / ".local/bin/mem-pressure-notify": "replacement-mem",
        home / ".local/bin/disk-fill-notify": "replacement-disk",
    }
    for path, text in replacements.items():
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(text)
    for service in SERVICES:
        (service_state / f"{service}.load").write_text("loaded")
        (service_state / f"{service}.enabled").write_text("enabled")
    env["FAIL_ACTION"] = "start"
    env["FAIL_SERVICE"] = "disk-fill-notify.service"

    result = run_helper("restore", home, state, env, check=False)

    assert result.returncode != 0
    for path, text in replacements.items():
        assert path.read_text() == text, result.stderr
    for service in SERVICES:
        assert (service_state / f"{service}.enabled").exists()
        assert not (service_state / f"{service}.active").exists()


def test_restore_preserves_valid_and_dangling_unit_symlinks(tmp_path):
    home, state, _, _, env = fixture(tmp_path)
    populate_paths(home)
    valid_target = tmp_path / "valid.service"
    valid_target.write_text("unit")
    valid = home / ".config/systemd/user/mem-pressure-notify.service"
    valid.unlink()
    valid.symlink_to(valid_target)
    dangling = home / ".config/systemd/user/disk-fill-notify.service"
    dangling_target = dangling.readlink()
    run_helper("retire", home, state, env)

    run_helper("restore", home, state, env)

    restored = home / ".config/systemd/user"
    assert (restored / "mem-pressure-notify.service").is_symlink()
    assert (restored / "mem-pressure-notify.service").readlink() == valid_target
    assert (restored / "disk-fill-notify.service").is_symlink()
    assert (restored / "disk-fill-notify.service").readlink() == dangling_target


def test_restore_rejects_externally_removed_directories(tmp_path):
    home, state, _, _, env = fixture(tmp_path)
    populate_paths(home)
    bin_dir = home / ".local/bin"
    unit_dir = home / ".config/systemd/user"
    bin_dir.chmod(0o751)
    unit_dir.chmod(0o750)
    run_helper("retire", home, state, env)
    bin_dir.rmdir()
    unit_dir.rmdir()

    result = run_helper("restore", home, state, env, check=False)

    assert result.returncode != 0
    assert "restore target directory is missing" in result.stderr
    assert not bin_dir.exists()
    assert not unit_dir.exists()


def test_missing_target_directories_remain_absent(tmp_path):
    home, state, _, _, env = fixture(
        tmp_path,
        not_found=SERVICES,
    )
    home.mkdir()

    result = run_helper("retire", home, state, env, check=False)
    assert result.returncode == 0, result.stderr

    assert not (home / ".local/bin").exists()
    assert not (home / ".config/systemd/user").exists()


def test_symlinked_target_directory_is_rejected_before_mutation(tmp_path):
    home, state, calls, _, env = fixture(tmp_path)
    outside = tmp_path / "outside"
    outside.mkdir()
    (home / ".local").mkdir(parents=True)
    (home / ".local/bin").symlink_to(outside)
    unit_dir = home / ".config/systemd/user"
    unit_dir.mkdir(parents=True)
    (unit_dir / "mem-pressure-notify.service").write_text("unit")
    (unit_dir / "disk-fill-notify.service").write_text("unit")

    result = run_helper("retire", home, state, env, check=False)

    assert result.returncode != 0
    assert "contains symlink" in result.stderr
    assert "disable --now" not in calls.read_text()
    assert not any(outside.iterdir())


def test_installer_dry_run_uses_target_account_home_everywhere(tmp_path):
    fakebin = tmp_path / "bin"
    fakebin.mkdir()
    target_home = tmp_path / "target-home"
    wrong_home = tmp_path / "wrong-home"
    write_executable(
        fakebin / "getent",
        f"#!/bin/sh\nprintf 'user:x:1000:1000::%s:/bin/sh\\n' '{target_home}'\n",
    )
    env = os.environ.copy()
    env.update(PATH=f"{fakebin}:{env['PATH']}", HOME=str(wrong_home))

    result = subprocess.run(
        ["sh", str(MODULE / "install.sh"), "--dry-run"],
        check=True,
        capture_output=True,
        env=env,
        text=True,
    )

    assert f"DRY-RUN: rm {target_home}/.local/bin/disk-fill-notify" in result.stdout
    assert str(wrong_home) not in result.stdout


def test_leaf_replacement_after_backup_aborts_without_deleting_replacement(tmp_path, monkeypatch):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_retirement_leaf", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    home, state, _, service_state, env = fixture(tmp_path)
    populate_paths(home)
    monkeypatch.setenv("PATH", env["PATH"])
    for key in (
        "CALLS",
        "SERVICE_STATE",
        "UNIT_DIR",
        "FAILURES",
        "FAIL_ACTION",
        "FAIL_SERVICE",
        "FAIL_ONCE",
        "RESTART_ON_RELOAD",
        "FAILED_SERVICE",
    ):
        monkeypatch.setenv(key, env[key])
    target = home / ".local/bin/disk-fill-notify"
    original = module.disable_services

    def replace_then_disable(*args):
        original(*args)
        target.unlink()
        target.write_text("concurrent replacement")

    monkeypatch.setattr(module, "disable_services", replace_then_disable)

    with pytest.raises(RuntimeError, match="object identity changed"):
        module.retire("user", 1000, home, state)

    assert target.read_text() == "concurrent replacement"
    assert_original_services(service_state)


def test_failed_restore_removes_files_whose_destinations_were_absent(tmp_path, monkeypatch):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_restore_absent", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    home, state, _, _, env = fixture(tmp_path)
    populate_paths(home)
    monkeypatch.setenv("PATH", env["PATH"])
    for key in (
        "CALLS",
        "SERVICE_STATE",
        "UNIT_DIR",
        "FAILURES",
        "FAIL_ACTION",
        "FAIL_SERVICE",
        "FAIL_ONCE",
        "RESTART_ON_RELOAD",
        "FAILED_SERVICE",
    ):
        monkeypatch.setenv(key, env[key])
    module.retire("user", 1000, home, state)
    original = module.replace_from_source
    calls = 0

    def fail_second_publish(*args):
        nonlocal calls
        calls += 1
        if calls == 2:
            raise OSError("injected restore publication failure")
        return original(*args)

    monkeypatch.setattr(module, "replace_from_source", fail_second_publish)

    with pytest.raises(OSError, match="injected restore publication failure"):
        module.restore("user", 1000, home, state)

    for destination in module.notifier_paths(home).values():
        assert not module.exists(destination)


def test_capture_cleanup_consumes_deleted_entry_before_fsync_failure(tmp_path, monkeypatch):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_capture_cleanup", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    directory = tmp_path / "targets"
    directory.mkdir()
    destinations = {
        "first": directory / "first",
        "second": directory / "second",
    }
    captures = {"first": ".first.capture", "second": ".second.capture"}
    for capture in captures.values():
        (directory / capture).write_text(capture)
    descriptor = os.open(directory, os.O_RDONLY | os.O_DIRECTORY)
    original = module.os.fsync
    calls = 0

    def fail_first_fsync(fd):
        nonlocal calls
        calls += 1
        if calls == 1:
            raise OSError("injected capture fsync failure")
        return original(fd)

    monkeypatch.setattr(module.os, "fsync", fail_first_fsync)
    try:
        with pytest.raises(RuntimeError, match="capture cleanup failed"):
            module.delete_pending_captures(captures, destinations, {directory: descriptor})
    finally:
        os.close(descriptor)

    assert "first" in captures
    assert "second" not in captures
    assert (directory / ".first.capture.cleanup").exists()
    assert not (directory / ".second.capture").exists()


def test_restore_preserves_post_publication_concurrent_replacement(tmp_path, monkeypatch):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_restore_post_publish", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    home, state, _, _, env = fixture(tmp_path)
    populate_paths(home)
    monkeypatch.setenv("PATH", env["PATH"])
    for key in (
        "CALLS",
        "SERVICE_STATE",
        "UNIT_DIR",
        "FAILURES",
        "FAIL_ACTION",
        "FAIL_SERVICE",
        "FAIL_ONCE",
        "RESTART_ON_RELOAD",
        "FAILED_SERVICE",
    ):
        monkeypatch.setenv(key, env[key])
    module.retire("user", 1000, home, state)
    target = home / ".local/bin/disk-fill-notify"
    original = module.run_systemctl

    def replace_before_reload(user, uid, *args):
        if args == ("daemon-reload",):
            target.unlink()
            target.write_text("post-publication replacement")
            raise RuntimeError("injected reload failure")
        return original(user, uid, *args)

    monkeypatch.setattr(module, "run_systemctl", replace_before_reload)

    with pytest.raises(RuntimeError, match="published notifier object identity changed"):
        module.restore("user", 1000, home, state)

    assert target.read_text() == "post-publication replacement"


def test_failed_restore_removes_created_directories(tmp_path, monkeypatch):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_restore_directories", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    home, state, _, _, env = fixture(tmp_path)
    populate_paths(home)
    monkeypatch.setenv("PATH", env["PATH"])
    for key in (
        "CALLS",
        "SERVICE_STATE",
        "UNIT_DIR",
        "FAILURES",
        "FAIL_ACTION",
        "FAIL_SERVICE",
        "FAIL_ONCE",
        "RESTART_ON_RELOAD",
        "FAILED_SERVICE",
    ):
        monkeypatch.setenv(key, env[key])
    module.retire("user", 1000, home, state)
    (home / ".local/bin").rmdir()
    (home / ".config/systemd/user").rmdir()
    env["FAIL_ACTION"] = "start"
    env["FAIL_SERVICE"] = "disk-fill-notify.service"
    monkeypatch.setenv("FAIL_ACTION", env["FAIL_ACTION"])
    monkeypatch.setenv("FAIL_SERVICE", env["FAIL_SERVICE"])

    with pytest.raises(Exception):
        module.restore("user", 1000, home, state)

    assert not (home / ".local/bin").exists()
    assert not (home / ".config/systemd/user").exists()


def test_capture_fsync_failure_restores_recorded_capture_and_services(tmp_path, monkeypatch):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_capture_fsync", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    home, state, _, service_state, env = fixture(tmp_path)
    populate_paths(home)
    monkeypatch.setenv("PATH", env["PATH"])
    for key in (
        "CALLS",
        "SERVICE_STATE",
        "UNIT_DIR",
        "FAILURES",
        "FAIL_ACTION",
        "FAIL_SERVICE",
        "FAIL_ONCE",
        "RESTART_ON_RELOAD",
        "FAILED_SERVICE",
    ):
        monkeypatch.setenv(key, env[key])
    original = module.os.fsync
    injected = False

    def fail_after_capture(fd):
        nonlocal injected
        target = Path(f"/proc/self/fd/{fd}")
        if not injected and target.is_dir() and any(".retire-" in item.name for item in target.iterdir()):
            injected = True
            raise OSError("injected capture fsync failure")
        return original(fd)

    monkeypatch.setattr(module.os, "fsync", fail_after_capture)

    with pytest.raises(OSError, match="injected capture fsync failure"):
        module.retire("user", 1000, home, state)

    assert_original_paths(home)
    assert_original_services(service_state)


def test_failed_restore_restores_preexisting_directory_metadata(tmp_path, monkeypatch):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_restore_metadata", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    home, state, _, _, env = fixture(tmp_path)
    populate_paths(home)
    monkeypatch.setenv("PATH", env["PATH"])
    for key in (
        "CALLS",
        "SERVICE_STATE",
        "UNIT_DIR",
        "FAILURES",
        "FAIL_ACTION",
        "FAIL_SERVICE",
        "FAIL_ONCE",
        "RESTART_ON_RELOAD",
        "FAILED_SERVICE",
    ):
        monkeypatch.setenv(key, env[key])
    archived_bin = home / ".local/bin"
    archived_units = home / ".config/systemd/user"
    archived_bin.chmod(0o751)
    archived_units.chmod(0o750)
    module.retire("user", 1000, home, state)
    archived_bin.chmod(0o711)
    archived_units.chmod(0o700)
    monkeypatch.setenv("FAIL_ACTION", "start")
    monkeypatch.setenv("FAIL_SERVICE", "disk-fill-notify.service")

    with pytest.raises(Exception):
        module.restore("user", 1000, home, state)

    assert archived_bin.stat().st_mode & 0o777 == 0o711
    assert archived_units.stat().st_mode & 0o777 == 0o700


def test_ensure_directories_partial_failure_keeps_creation_journal(tmp_path, monkeypatch):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_directory_journal", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    home = tmp_path / "home"
    home.mkdir()
    first = home / "first"
    second = home / "second"
    metadata = {
        first: (0o755, os.getuid(), os.getgid()),
        second: (0o755, os.getuid(), os.getgid()),
    }
    created = {}
    original = module.os.mkdir

    def fail_second(name, *args, **kwargs):
        if name == "second":
            raise OSError("injected directory creation failure")
        return original(name, *args, **kwargs)

    monkeypatch.setattr(module.os, "mkdir", fail_second)

    with pytest.raises(OSError, match="injected directory creation failure"):
        module.ensure_directories(home, metadata, created)

    assert first in created
    module.remove_created_directories(home, created)
    assert not first.exists()


def test_ensure_directories_journals_intent_before_mkdir(tmp_path, monkeypatch):
    module = load_module("notifier_directory_intent")
    home = tmp_path / "home"
    home.mkdir()
    target = home / "target"
    created = {}
    receipts = []
    original = module.os.mkdir

    def observe_mkdir(*args, **kwargs):
        receipts.append(dict(created))
        return original(*args, **kwargs)

    monkeypatch.setattr(module.os, "mkdir", observe_mkdir)
    module.ensure_directories(
        home,
        {target: (0o755, os.getuid(), os.getgid())},
        created,
        expected_current={target: None},
        journal_created=lambda values: receipts.append(dict(values)),
    )

    assert receipts[0] == {target: None}
    assert receipts[1] == {target: None}
    assert created[target] == (target.stat().st_dev, target.stat().st_ino)


def test_expected_absent_directory_concurrent_presence_is_rejected(tmp_path):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_absent_directory", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    home = tmp_path / "home"
    home.mkdir()
    target = home / "target"
    target.mkdir()
    metadata = {target: (0o755, os.getuid(), os.getgid())}
    expected_current = {target: None}

    with pytest.raises(RuntimeError, match="expected absent target directory appeared"):
        module.ensure_directories(home, metadata, {}, expected_current)

    assert target.exists()


def test_expected_present_directory_replacement_is_rejected(tmp_path):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_present_directory", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    home = tmp_path / "home"
    target = home / "target"
    target.mkdir(parents=True)
    original = target.stat()
    expected = {
        target: (
            original.st_mode & 0o7777,
            original.st_uid,
            original.st_gid,
            original.st_dev,
            original.st_ino,
        )
    }
    original_path = home / "original-target"
    target.rename(original_path)
    target.mkdir()

    with pytest.raises(RuntimeError, match="target directory identity changed"):
        module.ensure_directories(home, expected, {}, expected)


def test_mkdir_child_open_failure_removes_unjournaled_claim(tmp_path, monkeypatch):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_child_open", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    home = tmp_path / "home"
    home.mkdir()
    target = home / "target"
    metadata = {target: (0o755, os.getuid(), os.getgid())}
    original = module.os.open

    def fail_child_open(path, *args, **kwargs):
        if path == "target":
            raise OSError("injected child open failure")
        return original(path, *args, **kwargs)

    monkeypatch.setattr(module.os, "open", fail_child_open)

    with pytest.raises(OSError, match="injected child open failure"):
        module.ensure_directories(home, metadata, {})

    assert not target.exists()


def test_same_inode_same_mtime_content_change_is_rejected(tmp_path, monkeypatch):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_content_identity", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    home, state, _, service_state, env = fixture(tmp_path)
    populate_paths(home)
    monkeypatch.setenv("PATH", env["PATH"])
    for key in (
        "CALLS",
        "SERVICE_STATE",
        "UNIT_DIR",
        "FAILURES",
        "FAIL_ACTION",
        "FAIL_SERVICE",
        "FAIL_ONCE",
        "RESTART_ON_RELOAD",
        "FAILED_SERVICE",
    ):
        monkeypatch.setenv(key, env[key])
    target = home / ".local/bin/disk-fill-notify"
    original = module.disable_services

    def rewrite_then_disable(*args):
        before = target.stat()
        target.write_text("changed-fill-notify")
        os.utime(target, ns=(before.st_atime_ns, before.st_mtime_ns))
        original(*args)

    monkeypatch.setattr(module, "disable_services", rewrite_then_disable)

    with pytest.raises(RuntimeError, match="object identity changed"):
        module.retire("user", 1000, home, state)

    assert target.read_text() == "changed-fill-notify"
    assert_original_services(service_state)


def test_restore_preserves_setgid_directory_mode(tmp_path):
    home, state, _, _, env = fixture(tmp_path)
    populate_paths(home)
    bin_dir = home / ".local/bin"
    bin_dir.chmod(0o2751)
    run_helper("retire", home, state, env)
    bin_dir.chmod(0o700)

    run_helper("restore", home, state, env)

    assert bin_dir.stat().st_mode & 0o7777 == 0o2751


def test_next_retire_recovers_durable_crash_manifest(tmp_path, monkeypatch):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_crash_recovery", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    home, state, _, service_state, env = fixture(tmp_path)
    populate_paths(home)
    monkeypatch.setenv("PATH", env["PATH"])
    for key in (
        "CALLS",
        "SERVICE_STATE",
        "UNIT_DIR",
        "FAILURES",
        "FAIL_ACTION",
        "FAIL_SERVICE",
        "FAIL_ONCE",
        "RESTART_ON_RELOAD",
        "FAILED_SERVICE",
    ):
        monkeypatch.setenv(key, env[key])
    original = module.capture_leaf
    crashed = False

    def crash_after_first_capture(*args):
        nonlocal crashed
        capture = original(*args)
        if not crashed:
            crashed = True
            raise KeyboardInterrupt("injected crash")
        return capture

    monkeypatch.setattr(module, "capture_leaf", crash_after_first_capture)
    with pytest.raises(KeyboardInterrupt, match="injected crash"):
        module.retire("user", 1000, home, state)

    monkeypatch.setattr(module, "capture_leaf", original)
    module.retire("user", 1000, home, state)

    assert not (state / "retired/.notifier-transaction.json").exists()
    for destination in module.notifier_paths(home).values():
        assert not module.exists(destination)
    for service in SERVICES:
        assert not (service_state / f"{service}.active").exists()


def test_restore_leaf_replacement_after_backup_is_not_overwritten(tmp_path, monkeypatch):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_restore_leaf", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    home, state, _, _, env = fixture(tmp_path)
    populate_paths(home)
    monkeypatch.setenv("PATH", env["PATH"])
    for key in (
        "CALLS",
        "SERVICE_STATE",
        "UNIT_DIR",
        "FAILURES",
        "FAIL_ACTION",
        "FAIL_SERVICE",
        "FAIL_ONCE",
        "RESTART_ON_RELOAD",
        "FAILED_SERVICE",
    ):
        monkeypatch.setenv(key, env[key])
    module.retire("user", 1000, home, state)
    target = home / ".local/bin/disk-fill-notify"
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text("pre-restore")
    original = module.directory_metadata

    def replace_after_backup(*args, **kwargs):
        result = original(*args, **kwargs)
        target.unlink()
        target.write_text("concurrent restore replacement")
        return result

    monkeypatch.setattr(module, "directory_metadata", replace_after_backup)

    with pytest.raises(RuntimeError, match="object identity changed"):
        module.restore("user", 1000, home, state)

    assert target.read_text() == "concurrent restore replacement"


def test_copy_failure_before_mutation_keeps_files_and_services(tmp_path, monkeypatch):
    import importlib.util

    spec = importlib.util.spec_from_file_location("notifier_retirement", HELPER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    home, state, _, service_state, env = fixture(tmp_path)
    populate_paths(home)
    monkeypatch.setenv("PATH", env["PATH"])
    for key in (
        "CALLS",
        "SERVICE_STATE",
        "UNIT_DIR",
        "FAILURES",
        "FAIL_ACTION",
        "FAIL_SERVICE",
        "FAIL_ONCE",
        "RESTART_ON_RELOAD",
        "FAILED_SERVICE",
    ):
        monkeypatch.setenv(key, env[key])
    original = module.copy_object
    calls = 0

    def fail_second_copy(*args, **kwargs):
        nonlocal calls
        calls += 1
        if calls == 2:
            raise OSError("injected partial copy")
        return original(*args, **kwargs)

    monkeypatch.setattr(module, "copy_object", fail_second_copy)

    with pytest.raises(OSError, match="injected partial copy"):
        module.retire("user", 1000, home, state)

    assert_original_paths(home)
    assert_original_services(service_state)


def load_module(name, path=HELPER):
    import importlib.util

    spec = importlib.util.spec_from_file_location(name, path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_expected_absent_leaf_capture_uses_prejournaled_name(tmp_path):
    module = load_module("notifier_prejournaled_capture")
    parent = tmp_path / "parent"
    parent.mkdir()
    destination = parent / "notifier"
    capture = ".notifier.retire-known"
    descriptor = os.open(parent, os.O_RDONLY | os.O_DIRECTORY)
    try:
        assert module.capture_leaf(destination, descriptor, capture) is None
        destination.write_text("concurrent")
        assert not (parent / capture).exists()
    finally:
        os.close(descriptor)


def test_recovery_accepts_missing_journaled_created_directory(tmp_path, monkeypatch):
    module = load_module("notifier_missing_created_recovery")
    home, state, _, _, env = fixture(tmp_path)
    home.mkdir()
    archive = state / "retired"
    backup = archive / "notifier-restore-test"
    backup.mkdir(parents=True)
    destinations = module.notifier_paths(home)
    directories = {
        str(path.relative_to(home)): None
        for path in {item.parent for item in destinations.values()}
    }
    manifest = {
        "version": 1,
        "action": "restore",
        "home": str(home),
        "backup": backup.name,
        "states": {
            service: {"load": "not-found", "active": "inactive", "enabled": "not-found"}
            for service in SERVICES
        },
        "directories": directories,
        "captures": {name: None for name in destinations},
        "capture_identities": {},
        "published": {},
        "expected": {},
        "created": {".local/bin": [1, 2]},
        "phase": "mutating",
    }
    module.write_manifest(archive, manifest)
    monkeypatch.setenv("PATH", env["PATH"])
    for key in (
        "CALLS",
        "SERVICE_STATE",
        "UNIT_DIR",
        "FAILURES",
        "FAIL_ACTION",
        "FAIL_SERVICE",
        "FAIL_ONCE",
        "RESTART_ON_RELOAD",
        "FAILED_SERVICE",
    ):
        monkeypatch.setenv(key, env[key])

    module.recover_transaction("user", 1000, home, archive)

    assert module.read_manifest(archive) is None


@pytest.mark.parametrize("relative", ["", "."])
def test_directory_metadata_rejects_home_alias(relative, tmp_path):
    module = load_module("notifier_directory_metadata_validation")

    with pytest.raises(RuntimeError, match="invalid retired notifier directory metadata"):
        module.decode_directory_metadata({relative: None}, tmp_path)


def test_user_home_directory_creation_rejects_symlink_component(tmp_path):
    module = load_module("user_home_symlink_directory", HOME_FILES)
    home = tmp_path / "home"
    outside = tmp_path / "outside"
    home.mkdir()
    outside.mkdir()
    (home / ".local").symlink_to(outside, target_is_directory=True)

    with pytest.raises(OSError):
        module.ensure_directory(home, ".local/bin", os.getuid(), os.getgid(), 0o755)

    assert not (outside / "bin").exists()


def test_user_home_unlink_rejects_symlink_component(tmp_path):
    module = load_module("user_home_symlink_unlink", HOME_FILES)
    home = tmp_path / "home"
    outside = tmp_path / "outside"
    target = outside / "systemd/user/agent-guard.service"
    target.parent.mkdir(parents=True)
    target.write_text("outside")
    home.mkdir()
    (home / ".config").symlink_to(outside, target_is_directory=True)

    with pytest.raises(OSError):
        module.unlink_leaf(
            home,
            ".config/systemd/user/agent-guard.service",
            os.getuid(),
            os.getgid(),
            "/expected/agent-guard.service",
        )

    assert target.read_text() == "outside"