import io
import os
import socket
import stat
import tarfile
from collections.abc import Sequence
from pathlib import Path

import pyrage
import pytest
from rich.console import Console

from osx_backup.context import Context
from osx_backup.modules import secrets as secrets_module
from osx_backup.modules.base import NullProgress
from osx_backup.modules.secrets import (
    SECRETS_FILE,
    SecretsModule,
    collect_secret_paths,
    extract_archive,
    get_passphrase,
)
from osx_backup.runner import RecordingRunner

not_root = pytest.mark.skipif(os.geteuid() == 0, reason="root contourne les permissions de lecture")


def _seed(home: Path) -> None:
    (home / ".ssh").mkdir()
    (home / ".ssh" / "id_ed25519").write_text("PRIVATE")
    (home / ".config" / "gh").mkdir(parents=True)
    (home / ".config" / "gh" / "hosts.yml").write_text("token")
    proj = home / "Github" / "proj"
    proj.mkdir(parents=True)
    (proj / ".env").write_text("A=1")
    (proj / ".env.local").write_text("B=2")
    (proj / "node_modules" / "x").mkdir(parents=True)
    (proj / "node_modules" / "x" / ".env").write_text("ignored")
    (proj / ".git").mkdir()
    (proj / ".git" / ".env").write_text("ignored")


def _fresh_home(tmp_path: Path, dest: Path, map_path: Path, name: str = "home2") -> Context:
    home = tmp_path / name
    home.mkdir()
    return Context(
        home=home,
        dest=dest,
        runner=RecordingRunner(),
        console=Console(record=True, width=400),
        passphrase="pw",
        map_path=map_path,
    )


def _tar_bytes(members: Sequence[tuple[str, bytes]]) -> bytes:
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w:gz") as tar:
        for name, payload in members:
            info = tarfile.TarInfo(name)
            info.size = len(payload)
            info.mode = 0o600
            tar.addfile(info, io.BytesIO(payload))
    return buf.getvalue()


def test_collect_secret_paths(ctx: Context, map_path: Path) -> None:
    ctx.map_path = map_path
    _seed(ctx.home)
    rels = sorted(p.as_posix() for p in collect_secret_paths(ctx))
    assert rels == [".config/gh", ".ssh", "Github/proj/.env", "Github/proj/.env.local"]


def test_get_passphrase_sources(ctx: Context, monkeypatch: pytest.MonkeyPatch) -> None:
    ctx.passphrase = "ctx"
    assert get_passphrase(ctx, confirm=True) == "ctx"
    ctx.passphrase = None
    monkeypatch.setenv("OSX_BACKUP_PASSPHRASE", "env")
    assert get_passphrase(ctx, confirm=False) == "env"
    monkeypatch.delenv("OSX_BACKUP_PASSPHRASE")
    ctx.yes = True
    with pytest.raises(RuntimeError):
        get_passphrase(ctx, confirm=False)


def test_backup_restore_roundtrip(ctx: Context, map_path: Path, tmp_path: Path) -> None:
    ctx.map_path = map_path
    ctx.passphrase = "pw"
    _seed(ctx.home)
    m = SecretsModule()
    inv = m.inventory(ctx)
    assert inv.items == 4 and inv.available
    res = m.backup(ctx, NullProgress())
    assert res.status == "ok", res.message
    archive = ctx.dest / SECRETS_FILE
    assert archive.exists()
    assert b"PRIVATE" not in archive.read_bytes()
    assert not list(ctx.dest.glob("*.tar"))

    home2 = tmp_path / "home2"
    home2.mkdir()
    (home2 / ".ssh").mkdir()
    (home2 / ".ssh" / "id_ed25519").write_text("OLD")
    ctx2 = Context(
        home=home2,
        dest=ctx.dest,
        runner=RecordingRunner(),
        console=Console(record=True),
        passphrase="pw",
        map_path=map_path,
    )
    res2 = m.restore(ctx2, NullProgress())
    assert res2.status == "ok", res2.message
    assert (home2 / ".ssh" / "id_ed25519").read_text() == "PRIVATE"
    assert (home2 / "Github" / "proj" / ".env.local").read_text() == "B=2"
    assert (ctx2.pre_restore_dir / ".ssh" / "id_ed25519").read_text() == "OLD"
    assert stat.S_IMODE(os.stat(home2 / ".ssh").st_mode) == 0o700
    assert stat.S_IMODE(os.stat(home2 / ".ssh" / "id_ed25519").st_mode) == 0o600
    assert res.items == res2.items


def test_wrong_passphrase_fails_module(ctx: Context, map_path: Path) -> None:
    ctx.map_path = map_path
    ctx.passphrase = "pw"
    _seed(ctx.home)
    SecretsModule().backup(ctx, NullProgress())
    ctx.passphrase = "nope"
    res = SecretsModule().restore(ctx, NullProgress())
    assert res.status == "failed" and "passphrase" in res.message


def test_restore_without_archive_skipped(ctx: Context) -> None:
    assert SecretsModule().restore(ctx, NullProgress()).status == "skipped"


def test_extract_archive_refuses_paths_outside_home(ctx: Context, tmp_path: Path) -> None:
    """Une archive forgée ne doit toucher à rien hors de $HOME (noms absolus, ``..``)."""
    ctx.console = Console(record=True, width=400)
    outside = tmp_path / "outside"
    outside.mkdir()
    victim = outside / "victim.txt"
    victim.write_text("DO NOT TOUCH")
    data = _tar_bytes(
        [
            (str(victim), b"pwned"),
            ("../escape.txt", b"pwned"),
            (".ssh/id_ed25519", b"PRIVATE"),
        ]
    )
    assert extract_archive(ctx, data) == 1
    assert victim.read_text() == "DO NOT TOUCH"
    assert not (tmp_path / "escape.txt").exists()
    assert (ctx.home / ".ssh" / "id_ed25519").read_text() == "PRIVATE"
    out = ctx.console.export_text()
    assert "victim.txt" in out and "escape.txt" in out


def test_restore_with_unreadable_archive_fails_cleanly(ctx: Context) -> None:
    ctx.passphrase = "pw"
    (ctx.dest / SECRETS_FILE).write_bytes(pyrage.passphrase.encrypt(b"pas un tar", "pw"))
    res = SecretsModule().restore(ctx, NullProgress())
    assert res.status == "failed" and "archive" in res.message


def test_absolute_symlink_is_dereferenced(ctx: Context, map_path: Path, tmp_path: Path) -> None:
    """Un lien vers un dépôt de dotfiles redevient un vrai fichier ; un lien cassé est ignoré."""
    ctx.map_path = map_path
    ctx.passphrase = "pw"
    ctx.console = Console(record=True, width=400)
    _seed(ctx.home)
    external = tmp_path / "dotfiles"
    external.mkdir()
    (external / "ssh_config").write_text("Host *")
    (ctx.home / ".ssh" / "config").symlink_to(external / "ssh_config")
    (ctx.home / ".ssh" / "dangling").symlink_to(tmp_path / "nulle-part")

    m = SecretsModule()
    res = m.backup(ctx, NullProgress())
    assert res.status == "ok", res.message
    assert "dangling" in ctx.console.export_text()

    ctx2 = _fresh_home(tmp_path, ctx.dest, map_path)
    res2 = m.restore(ctx2, NullProgress())
    assert res2.status == "ok", res2.message
    config = ctx2.home / ".ssh" / "config"
    assert not config.is_symlink()
    assert config.read_text() == "Host *"
    assert (ctx2.home / ".ssh" / "id_ed25519").read_text() == "PRIVATE"
    assert not (ctx2.home / ".ssh" / "dangling").exists()


@not_root
def test_unreadable_file_keeps_the_rest_of_the_entry(
    ctx: Context, map_path: Path, tmp_path: Path
) -> None:
    ctx.map_path = map_path
    ctx.passphrase = "pw"
    _seed(ctx.home)
    gh = ctx.home / ".config" / "gh"
    (gh / "a.yml").write_text("A")
    (gh / "b.yml").write_text("B")
    (gh / "c.yml").write_text("C")
    (gh / "b.yml").chmod(0o000)
    m = SecretsModule()
    try:
        res = m.backup(ctx, NullProgress())
    finally:
        (gh / "b.yml").chmod(0o600)
    assert res.status == "ok", res.message
    out = ctx.console.export_text()
    assert "b.yml" in out and "a.yml" not in out
    assert (ctx.dest / SECRETS_FILE).exists()

    ctx2 = _fresh_home(tmp_path, ctx.dest, map_path)
    res2 = m.restore(ctx2, NullProgress())
    assert res2.status == "ok", res2.message
    assert (ctx2.home / ".config" / "gh" / "a.yml").read_text() == "A"
    assert (ctx2.home / ".config" / "gh" / "c.yml").read_text() == "C"
    assert not (ctx2.home / ".config" / "gh" / "b.yml").exists()
    assert res.items == res2.items


@not_root
def test_backup_fails_when_nothing_can_be_archived(ctx: Context, map_path: Path) -> None:
    ctx.map_path = map_path
    ctx.passphrase = "pw"
    (ctx.home / ".ssh").mkdir()
    unreadable = ctx.home / ".ssh" / "id_ed25519"
    unreadable.write_text("PRIVATE")
    unreadable.chmod(0o000)
    try:
        res = SecretsModule().backup(ctx, NullProgress())
    finally:
        unreadable.chmod(0o600)
    assert res.status == "failed"
    assert not (ctx.dest / SECRETS_FILE).exists()


def test_special_files_are_skipped_and_not_counted(
    ctx: Context, map_path: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Socket (agent ssh) et fifo ne rentrent pas dans le tar : ne pas les compter non plus.

    `tarfile.gettarinfo` rend None pour un socket et `add` ne fait alors rien, sans lever ;
    un fifo entre bien dans l'archive mais l'extraction ne repose que fichiers et liens. Sans
    garde explicite, la sauvegarde annonçait plus de fichiers chiffrés que l'archive n'en
    contient. Les sockets d'agent ssh réels (`s.r1R6siXHN7.agent.78xB2Nf9uK`) n'ont pas de
    suffixe `.sock` : ils passent au travers de l'exclusion globale `**/*.sock`.
    """
    ctx.map_path = map_path
    ctx.passphrase = "pw"
    _seed(ctx.home)
    agent = ctx.home / ".ssh" / "agent"
    agent.mkdir()
    os.mkfifo(agent / "s.fifo")
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    # AF_UNIX plafonne le chemin à ~104 octets : on se place dans le dossier pour rester court.
    monkeypatch.chdir(agent)
    try:
        sock.bind("s.agent")
        res = SecretsModule().backup(ctx, NullProgress())
    finally:
        sock.close()
    assert res.status == "ok", res.message
    out = ctx.console.export_text()
    assert "~/.ssh/agent/s.agent" in out and "~/.ssh/agent/s.fifo" in out
    # Les avertissements de l'archivage remontent dans le résultat, donc dans le manifeste.
    assert any("s.agent" in w for w in res.warnings)
    assert any("s.fifo" in w for w in res.warnings)

    ctx2 = _fresh_home(tmp_path, ctx.dest, map_path)
    res2 = SecretsModule().restore(ctx2, NullProgress())
    assert res2.status == "ok", res2.message
    assert res.items == res2.items
    assert not (ctx2.home / ".ssh" / "agent" / "s.agent").exists()
    assert not (ctx2.home / ".ssh" / "agent" / "s.fifo").exists()
    assert (ctx2.home / ".ssh" / "id_ed25519").read_text() == "PRIVATE"


def test_archive_write_failure_keeps_the_previous_archive(
    ctx: Context, map_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Écriture atomique : une panne en cours d'écriture ne doit pas tronquer l'archive."""
    ctx.map_path = map_path
    ctx.passphrase = "pw"
    _seed(ctx.home)
    archive = ctx.dest / SECRETS_FILE
    archive.write_bytes(b"archive precedente")
    real_write = Path.write_bytes

    def boom(self: Path, data: bytes) -> int:
        if self.name.endswith(".tmp"):
            raise OSError(28, "No space left on device")
        return real_write(self, data)

    monkeypatch.setattr(Path, "write_bytes", boom)
    with pytest.raises(OSError):
        SecretsModule().backup(ctx, NullProgress())
    monkeypatch.undo()
    assert archive.read_bytes() == b"archive precedente"
    assert list(ctx.dest.glob("*.tmp")) == []


def test_backup_dry_run_plans_without_passphrase(
    ctx: Context, map_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """En dry-run, rien n'est chiffré : demander la passphrase n'aurait aucun objet."""
    ctx.map_path = map_path
    ctx.dry_run = True
    _seed(ctx.home)

    def boom(*_a: object, **_kw: object) -> str:
        raise AssertionError("get_passphrase ne doit pas être appelée en dry-run")

    monkeypatch.setattr(secrets_module, "get_passphrase", boom)
    res = SecretsModule().backup(ctx, NullProgress())
    assert res.status == "ok" and "dry-run" in res.message
    assert not (ctx.dest / SECRETS_FILE).exists()