"""The agent host's vault and SilverBullet (deploy/agent-host; L10, M5.4):
the operator writes the vault, never the agents' memory; nothing but the
vault and SilverBullet's own data is mounted; the image changes only in a
PR; only the operator's Caddy reaches it."""

import re
import shutil
import subprocess
from pathlib import Path
from typing import Any

import pytest
import yaml

HOST = Path(__file__).resolve().parents[2] / "deploy" / "agent-host"
VAULT = "/srv/nisse/vault"


def silverbullet() -> dict[str, Any]:
    compose = yaml.safe_load((HOST / "silverbullet" / "compose.yaml").read_text())
    return compose["services"]["silverbullet"]


def mounts() -> list[tuple[str, str, str]]:
    out = []
    for volume in silverbullet()["volumes"]:
        source, target, *mode = volume.split(":")
        out.append((source, target, mode[0] if mode else "rw"))
    return out


def setting(name: str) -> str:
    match = re.search(rf"^{name}=(\S+)", (HOST / "setup.sh").read_text(), re.MULTILINE)
    assert match, name
    return match.group(1)


def test_the_agents_memory_is_read_only_and_the_rest_of_the_vault_is_not() -> None:
    by_source = {source: (target, mode) for source, target, mode in mounts()}
    vault_target, vault_mode = by_source[VAULT]
    assert vault_mode == "rw"
    # memory/ is mounted over the same place in the space, read-only.
    assert by_source[f"{VAULT}/memory"] == (f"{vault_target}/memory", "ro")
    # Nothing else of the vault is mounted, so nothing gets around that.
    others = [s for s in by_source if s.startswith(VAULT)]
    assert sorted(others) == [VAULT, f"{VAULT}/memory"]


def test_it_sees_nothing_of_the_host_but_the_vault_and_its_own_data() -> None:
    assert sorted(source for source, _, _ in mounts()) == [
        "/srv/nisse/silverbullet",
        VAULT,
        f"{VAULT}/memory",
    ]


def test_the_image_is_pinned_to_a_release() -> None:
    # The docker role updates images nightly; a pinned release changes only here.
    image = silverbullet()["image"]
    assert re.fullmatch(
        r"ghcr\.io/silverbulletmd/silverbullet:\d+\.\d+\.\d+(-slim)?", image
    )


def test_no_secrets_in_the_compose_file() -> None:
    environment = silverbullet().get("environment", {})
    assert not [k for k in environment if re.search(r"USER|TOKEN|PASS|SECRET|KEY", k)]


def test_its_port_caddys_and_the_firewalls_agree() -> None:
    port = setting("SB_PORT")
    assert silverbullet()["ports"] == [f"0.0.0.0:{port}:3000"]  # IPv4 only
    caddy = (HOST / "caddy" / "vault.dongar.net.caddy").read_text()
    assert f"reverse_proxy nisse.garrett-hall.com:{port}" in caddy
    assert "@outside not remote_ip private_ranges" in caddy  # the LAN only


def test_logs_go_to_the_hosts_journal() -> None:
    assert silverbullet()["logging"] == {"driver": "journald"}


def _can_use_a_network_namespace() -> bool:
    if not (shutil.which("unshare") and shutil.which("iptables")):
        return False
    probe = subprocess.run(
        ["unshare", "-n", "iptables", "-S"], capture_output=True, check=False
    )
    return probe.returncode == 0


@pytest.mark.skipif(
    not _can_use_a_network_namespace(),
    reason="needs root, unshare and iptables to try the rule in a namespace",
)
def test_the_firewall_lets_only_one_address_in_and_replaces_its_own_rule() -> None:
    script = HOST / "nisse-firewall"
    steps = (
        "iptables -N DOCKER-USER; iptables -A DOCKER-USER -j RETURN; "
        f"{script} 192.168.6.13 3000; {script} 192.168.6.13 3000; "
        f"{script} 192.168.6.14 3000; iptables -S DOCKER-USER"
    )
    run = subprocess.run(
        ["unshare", "-n", "sh", "-ec", steps],
        capture_output=True,
        text=True,
        check=True,
    )
    rules = [r for r in run.stdout.splitlines() if "nisse-firewall" in r]
    assert len(rules) == 1, run.stdout  # run again: still one; changed: replaced
    [rule] = rules
    assert "! -s 192.168.6.14/32" in rule
    assert "--ctorigdstport 3000" in rule
    assert rule.endswith("-j DROP")
    # Ahead of Docker's own return, so it's checked first.
    assert run.stdout.splitlines().index(rule) < run.stdout.splitlines().index(
        "-A DOCKER-USER -j RETURN"
    )