from __future__ import annotations

import importlib.util
import os
import shutil
import stat
import uuid
from pathlib import Path
from types import ModuleType

import pytest

ROOT = Path(__file__).resolve().parents[2]
BROKER_ROOT = ROOT / "ops" / "gitea-deploy-broker"
INSTALLER_PATH = BROKER_ROOT / "prepare_managed_appdata_root.py"


def _load_module(name: str, path: Path) -> ModuleType | None:
    helper_path = BROKER_ROOT / "managed_appdata_paths.py"
    if not path.is_file() or not helper_path.is_file():
        return None
    if str(BROKER_ROOT) not in os.sys.path:
        os.sys.path.insert(0, str(BROKER_ROOT))
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None or spec.loader is None:
        return None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


@pytest.fixture
def root_owned_test_tree():
    if os.name != "posix" or not hasattr(os, "geteuid") or os.geteuid() != 0:
        pytest.skip("requires a POSIX root-owned test tree")
    parent = Path("/root")
    if parent.is_symlink() or not parent.is_dir():
        pytest.skip("requires the fixed root-owned /root test parent")
    base = parent / f"appops-managed-root-test-{uuid.uuid4().hex}"
    base.mkdir(mode=0o700)
    base.chmod(0o700)
    try:
        yield base
    finally:
        if base.parent == parent and base.name.startswith("appops-managed-root-test-"):
            shutil.rmtree(base)


def _root_safe_path(base: Path, *components: str) -> Path:
    current = base
    for component in components:
        current = current / component
        current.mkdir(mode=0o700)
        current.chmod(0o700)
    return current


def test_open_fixed_directory_walks_real_owned_components_without_symlinks(
    root_owned_test_tree, monkeypatch
):
    paths = _load_module("managed_appdata_open_real_chain_test", BROKER_ROOT / "managed_appdata_paths.py")
    assert paths is not None
    target = _root_safe_path(root_owned_test_tree, "docker", "volumes", "itworx-managed-appdata", "_data")
    monkeypatch.setattr(paths, "MANAGED_APPDATA_ROOT", target)

    descriptor = paths._open_fixed_directory(target)
    try:
        info = os.fstat(descriptor)
        assert (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) == (0, 0, 0o700)
    finally:
        os.close(descriptor)


def test_open_fixed_directory_rejects_symlink_component(root_owned_test_tree, monkeypatch):
    paths = _load_module("managed_appdata_open_symlink_test", BROKER_ROOT / "managed_appdata_paths.py")
    assert paths is not None
    real_target = _root_safe_path(root_owned_test_tree, "real", "managed")
    alias = root_owned_test_tree / "alias"
    alias.symlink_to(real_target.parent, target_is_directory=True)
    target = alias / real_target.name
    monkeypatch.setattr(paths, "MANAGED_APPDATA_ROOT", target)

    with pytest.raises(OSError):
        paths._open_fixed_directory(target)


def test_open_fixed_directory_rejects_world_writable_ancestor(root_owned_test_tree, monkeypatch):
    paths = _load_module("managed_appdata_open_world_writable_test", BROKER_ROOT / "managed_appdata_paths.py")
    assert paths is not None
    unsafe = root_owned_test_tree / "unsafe"
    unsafe.mkdir(mode=0o777)
    unsafe.chmod(0o777)
    target = _root_safe_path(unsafe, "managed")
    monkeypatch.setattr(paths, "MANAGED_APPDATA_ROOT", target)

    with pytest.raises(paths.UnsafeManagedAppdataRoot):
        paths._open_fixed_directory(target)


@pytest.mark.parametrize("owner_field", ["uid", "gid"])
def test_open_fixed_directory_rejects_wrong_owner_or_group(
    root_owned_test_tree, monkeypatch, owner_field
):
    paths = _load_module(f"managed_appdata_open_wrong_{owner_field}_test", BROKER_ROOT / "managed_appdata_paths.py")
    assert paths is not None
    owner = root_owned_test_tree / "wrong-owner"
    target = _root_safe_path(root_owned_test_tree, "wrong-owner", "managed")
    if not hasattr(os, "chown"):
        pytest.skip("requires POSIX chown for the owned test directory")
    if owner_field == "uid":
        os.chown(owner, 65534, 0)
    else:
        os.chown(owner, 0, 65534)
    monkeypatch.setattr(paths, "MANAGED_APPDATA_ROOT", target)

    with pytest.raises(paths.UnsafeManagedAppdataRoot):
        paths._open_fixed_directory(target)


@pytest.mark.parametrize("acl_name", ["system.posix_acl_access", "system.posix_acl_default"])
def test_open_fixed_directory_rejects_reported_acl(root_owned_test_tree, monkeypatch, acl_name):
    paths = _load_module("managed_appdata_open_acl_test", BROKER_ROOT / "managed_appdata_paths.py")
    assert paths is not None
    acl_component = root_owned_test_tree / "acl"
    target = _root_safe_path(root_owned_test_tree, "acl", "managed")
    acl_stat = acl_component.stat()
    original_listxattr = os.listxattr

    def listxattr(path, *args, **kwargs):
        if isinstance(path, int):
            info = os.fstat(path)
            if (info.st_dev, info.st_ino) == (acl_stat.st_dev, acl_stat.st_ino):
                return [acl_name]
        return original_listxattr(path, *args, **kwargs)

    monkeypatch.setattr(paths, "MANAGED_APPDATA_ROOT", target)
    monkeypatch.setattr(paths.os, "listxattr", listxattr)

    with pytest.raises(paths.UnsafeManagedAppdataRoot, match="unreviewed ACL"):
        paths._open_fixed_directory(target)


def test_fixed_controller_installer_creates_private_root_and_refuses_drift(tmp_path, monkeypatch):
    installer = _load_module("managed_appdata_installer_test", INSTALLER_PATH)
    assert installer is not None, "the fixed managed-appdata installer must ship with the controller artifact"
    if os.name != "posix" or not hasattr(os, "geteuid") or os.geteuid() != 0:
        pytest.skip("requires POSIX root ownership semantics")

    target = tmp_path / "volume-root"
    target.mkdir(mode=0o755)
    target.chmod(0o755)
    monkeypatch.setattr(installer, "MANAGED_APPDATA_ROOT", target)
    monkeypatch.setattr(installer, "validate_initialization_prerequisites", lambda: True)
    monkeypatch.setattr(installer, "configured_appdata_root", lambda: target)
    monkeypatch.setattr(installer, "_open_fixed_directory", lambda _path: os.open(target, os.O_RDONLY | os.O_DIRECTORY))

    assert installer.ensure_managed_appdata_root() == "initialized"
    info = target.lstat()
    assert stat.S_ISDIR(info.st_mode)
    assert (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) == (0, 0, 0o700)
    marker = target / installer.MANAGED_ROOT_MARKER
    assert marker.read_bytes() == installer.MANAGED_ROOT_MARKER_BYTES
    assert stat.S_IMODE(marker.lstat().st_mode) == 0o400
    assert installer.ensure_managed_appdata_root() == "already_initialized"

    target.chmod(0o777)
    with pytest.raises(installer.UnsafeManagedAppdataRoot):
        installer.ensure_managed_appdata_root()
    assert stat.S_IMODE(target.lstat().st_mode) == 0o777


def test_fixed_controller_installer_refuses_existing_data_without_deleting_it(tmp_path, monkeypatch):
    installer = _load_module("managed_appdata_installer_reject_test", INSTALLER_PATH)
    assert installer is not None
    if os.name != "posix" or not hasattr(os, "geteuid") or os.geteuid() != 0:
        pytest.skip("requires POSIX root ownership semantics")

    target = tmp_path / "volume-root"
    target.mkdir(mode=0o755)
    target.chmod(0o755)
    monkeypatch.setattr(installer, "MANAGED_APPDATA_ROOT", target)
    monkeypatch.setattr(installer, "validate_initialization_prerequisites", lambda: True)
    monkeypatch.setattr(installer, "configured_appdata_root", lambda: target)
    monkeypatch.setattr(installer, "_open_fixed_directory", lambda _path: os.open(target, os.O_RDONLY | os.O_DIRECTORY))

    (target / "existing-data").write_text("keep", encoding="utf-8")
    with pytest.raises(installer.UnsafeManagedAppdataRoot):
        installer.ensure_managed_appdata_root()
    assert (target / "existing-data").read_text(encoding="utf-8") == "keep"
    assert not (target / installer.MANAGED_ROOT_MARKER).exists()


def test_initialized_volume_is_idempotent_without_enumerating_application_data(tmp_path, monkeypatch):
    installer = _load_module("managed_appdata_installer_idempotent_test", INSTALLER_PATH)
    assert installer is not None
    if os.name != "posix" or not hasattr(os, "geteuid") or os.geteuid() != 0:
        pytest.skip("requires POSIX root ownership semantics")

    target = tmp_path / "volume-root"
    target.mkdir(mode=0o700)
    (target / installer.MANAGED_ROOT_MARKER).write_bytes(installer.MANAGED_ROOT_MARKER_BYTES)
    (target / installer.MANAGED_ROOT_MARKER).chmod(0o400)
    (target / "app-data").mkdir()
    monkeypatch.setattr(installer, "MANAGED_APPDATA_ROOT", target)
    monkeypatch.setattr(installer, "validate_initialization_prerequisites", lambda: True)
    monkeypatch.setattr(installer, "configured_appdata_root", lambda: target)
    monkeypatch.setattr(installer, "_open_fixed_directory", lambda _path: os.open(target, os.O_RDONLY | os.O_DIRECTORY))
    monkeypatch.setattr(installer.os, "listdir", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("data must not be enumerated")))

    assert installer.ensure_managed_appdata_root() == "already_initialized"


def _valid_mounts(paths):
    return [
        {
            "Type": "volume",
            "Name": paths.VOLUME_NAME,
            "Source": paths.VOLUME_HOST_ROOT.as_posix(),
            "Destination": paths.MANAGED_APPDATA_ROOT.as_posix(),
            "Driver": "local",
            "RW": True,
        },
        {
            "Type": "bind",
            "Name": "",
            "Source": paths.DOCKER_ROOT.as_posix(),
            "Destination": paths.DOCKER_ROOT.as_posix(),
            "Driver": "",
            "RW": False,
        },
    ]


def test_controller_mount_inventory_accepts_only_exact_volume_and_readonly_docker_root():
    paths = _load_module("managed_appdata_mounts_test", BROKER_ROOT / "managed_appdata_paths.py")
    assert paths is not None
    mounts = _valid_mounts(paths)

    paths.validate_controller_mounts(
        mounts,
        controller_id_before="a" * 64,
        controller_id_after="a" * 64,
    )


@pytest.mark.parametrize(
    "mutate",
    [
        lambda m: m[0].update(Name="other-volume"),
        lambda m: m[0].update(Type="bind"),
        lambda m: m[0].update(Driver="nfs"),
        lambda m: m[0].update(Source="/var/lib/docker/volumes/other/_data"),
        lambda m: m[0].update(Destination="/managed-appdata"),
        lambda m: m[0].update(RW=False),
        lambda m: m[1].update(RW=True),
        lambda m: m.append(dict(m[0])),
        lambda m: m.append({"Type": "bind", "Source": "/var/lib/docker/volumes", "Destination": "/var/lib/docker/volumes", "RW": True}),
        lambda m: m.append({"Type": "bind", "Source": "/var/lib/docker/volumes/itworx-managed-appdata/_data/app", "Destination": "/var/lib/docker/volumes/itworx-managed-appdata/_data/app", "RW": True}),
        lambda m: m.append({"Type": "bind", "Source": "/var/lib/docker", "Destination": "/var/lib/docker", "RW": False}),
    ],
)
def test_controller_mount_inventory_fails_closed_on_mount_drift(mutate):
    paths = _load_module("managed_appdata_mounts_reject_test", BROKER_ROOT / "managed_appdata_paths.py")
    assert paths is not None
    mounts = _valid_mounts(paths)
    mutate(mounts)

    with pytest.raises(paths.UnsafeManagedAppdataRoot):
        paths.validate_controller_mounts(
            mounts,
            controller_id_before="b" * 64,
            controller_id_after="b" * 64,
        )


def test_controller_mount_inventory_rejects_container_replacement_during_inspection():
    paths = _load_module("managed_appdata_mounts_race_test", BROKER_ROOT / "managed_appdata_paths.py")
    assert paths is not None

    with pytest.raises(paths.UnsafeManagedAppdataRoot):
        paths.validate_controller_mounts(
            _valid_mounts(paths),
            controller_id_before="c" * 64,
            controller_id_after="d" * 64,
        )


@pytest.mark.parametrize(
    "override",
    [
        {"Name": "other"},
        {"Driver": "nfs"},
        {"Scope": "global"},
        {"Mountpoint": "/var/lib/docker/volumes/other/_data"},
        {"Options": {"type": "nfs"}},
        {"Labels": {"itworx.managed-by": "other"}},
    ],
)
def test_volume_metadata_rejects_drift(override):
    paths = _load_module("managed_appdata_volume_reject_test", BROKER_ROOT / "managed_appdata_paths.py")
    assert paths is not None
    document = {
        "Name": paths.VOLUME_NAME,
        "Driver": "local",
        "Scope": "local",
        "Mountpoint": paths.VOLUME_HOST_ROOT.as_posix(),
        "Options": None,
        "Labels": dict(paths.VOLUME_LABELS),
    }
    document.update(override)

    with pytest.raises(paths.UnsafeManagedAppdataRoot):
        paths.validate_volume_metadata(document, paths.DOCKER_ROOT.as_posix())


def test_controller_rejects_world_writable_legacy_root_without_mutating_it(tmp_path):
    paths = _load_module("managed_appdata_paths_test", BROKER_ROOT / "managed_appdata_paths.py")
    assert paths is not None, "the controller must centralize fixed managed-appdata path checks"
    if os.name != "posix":
        pytest.skip("requires POSIX permission bits")

    root = tmp_path / "legacy"
    root.mkdir()
    paths.LEGACY_APPDATA_ROOT = root
    root.chmod(0o777)
    assert paths.appdata_root_is_safe(root) is False
    assert stat.S_IMODE(root.lstat().st_mode) == 0o777

    root.chmod(0o750)
    assert paths.appdata_root_is_safe(root) is True