import json
import shutil
import subprocess

import pytest

from osx_backup.context import Context
from osx_backup.modules import postgres
from osx_backup.modules.base import NullProgress, get_module
from osx_backup.modules.postgres import (
    EXISTS_SQL,
    LIST_SQL,
    Database,
    PostgresModule,
    parse_databases,
    postgres_formula,
)
from osx_backup.runner import RealRunner, RecordingRunner

LIST_OUT = "advmfr_hr|advmfr\npostgres|humperdoo\nseoinsight_data|seoinsight\n"


def test_registered() -> None:
    assert isinstance(get_module("postgres"), PostgresModule)


def test_parse_databases() -> None:
    assert parse_databases(LIST_OUT) == [
        Database("advmfr_hr", "advmfr"),
        Database("seoinsight_data", "seoinsight"),
    ]


def test_postgres_formula(ctx: Context) -> None:
    assert postgres_formula(ctx.dest) == "postgresql@18"
    (ctx.dest / "brew").mkdir()
    (ctx.dest / "brew" / "services.json").write_text(
        json.dumps(
            [{"name": "redis", "status": "none"}, {"name": "postgresql@17", "status": "started"}]
        )
    )
    assert postgres_formula(ctx.dest) == "postgresql@17"


def _runner() -> RecordingRunner:
    r = RecordingRunner()
    r.add(("psql", "-X", "-At", "-d", "postgres", "-c", LIST_SQL), stdout=LIST_OUT)
    r.add(("pg_dumpall", "--globals-only"), stdout="CREATE ROLE advmfr;\n")
    return r


def test_inventory(ctx: Context) -> None:
    ctx.runner = _runner()
    inv = PostgresModule().inventory(ctx)
    assert inv.available and inv.items == 2
    assert inv.details["databases"] == ["advmfr_hr", "seoinsight_data"]


def test_inventory_unreachable(ctx: Context) -> None:
    r = RecordingRunner()
    r.add(("psql",), returncode=2, stderr="connection refused")
    ctx.runner = r
    inv = PostgresModule().inventory(ctx)
    assert not inv.available and "injoignable" in inv.reason


def test_inventory_without_pg_dump(ctx: Context) -> None:
    ctx.runner = RecordingRunner(which_map={})
    assert not PostgresModule().inventory(ctx).available


def test_backup_dumps_globals_and_each_db(ctx: Context) -> None:
    r = _runner()
    ctx.runner = r
    p = NullProgress()
    res = PostgresModule().backup(ctx, p)
    assert res.status == "ok" and res.items == 2
    out = ctx.dest / "postgres"
    assert (out / "globals.sql").read_text() == "CREATE ROLE advmfr;\n"
    assert json.loads((out / "databases.json").read_text()) == [
        {"name": "advmfr_hr", "owner": "advmfr"},
        {"name": "seoinsight_data", "owner": "seoinsight"},
    ]
    assert ["pg_dump", "-Fc", "-f", str(out / "advmfr_hr.dump"), "advmfr_hr"] in r.mutations
    assert p.tasks[0].completed == 2


def test_backup_unreachable_fails(ctx: Context) -> None:
    r = RecordingRunner()
    r.add(("psql",), returncode=2, stderr="connection refused")
    ctx.runner = r
    res = PostgresModule().backup(ctx, NullProgress())
    assert res.status == "failed" and "injoignable" in res.message
    assert not r.has_call("pg_dumpall")


def test_backup_globals_failure_fails(ctx: Context) -> None:
    r = _runner()
    r.add(("pg_dumpall", "--globals-only"), returncode=1, stderr="ligne 1\nfatal: pas de rôle\n")
    ctx.runner = r
    res = PostgresModule().backup(ctx, NullProgress())
    assert res.status == "failed" and "fatal: pas de rôle" in res.message
    assert not r.has_call("pg_dump", "-Fc")


def test_backup_continues_when_one_db_fails(ctx: Context) -> None:
    out = ctx.dest / "postgres"
    r = _runner()
    r.add(
        ("pg_dump", "-Fc", "-f", str(out / "advmfr_hr.dump")),
        returncode=1,
        stderr="pg_dump: disque plein",
    )
    ctx.runner = r
    res = PostgresModule().backup(ctx, NullProgress())
    assert res.status == "ok" and res.items == 1
    assert any("advmfr_hr" in w and "disque plein" in w for w in res.warnings)
    assert ["pg_dump", "-Fc", "-f", str(out / "seoinsight_data.dump"), "seoinsight_data"] in (
        r.mutations
    )
    # databases.json ne liste que les bases réellement dumpées.
    assert json.loads((out / "databases.json").read_text()) == [
        {"name": "seoinsight_data", "owner": "seoinsight"}
    ]


def test_backup_fails_when_no_db_dumped(ctx: Context) -> None:
    r = _runner()
    r.add(("pg_dump", "-Fc"), returncode=1, stderr="pg_dump: disque plein")
    ctx.runner = r
    res = PostgresModule().backup(ctx, NullProgress())
    assert res.status == "failed" and "aucune base" in res.message
    assert not (ctx.dest / "postgres" / "databases.json").exists()


def test_backup_clears_stale_metadata(ctx: Context) -> None:
    """Une sauvegarde ratée ne doit pas laisser les métadonnées du run précédent."""
    out = ctx.dest / "postgres"
    out.mkdir()
    (out / "globals.sql").write_text("CREATE ROLE ancien;\n")
    (out / "databases.json").write_text(json.dumps([{"name": "ancienne", "owner": "moi"}]))
    r = RecordingRunner()
    r.add(("psql",), returncode=2, stderr="connection refused")
    ctx.runner = r
    assert PostgresModule().backup(ctx, NullProgress()).status == "failed"
    assert not (out / "globals.sql").exists()
    assert not (out / "databases.json").exists()


def _seed_backup(ctx: Context) -> None:
    out = ctx.dest / "postgres"
    out.mkdir()
    (out / "globals.sql").write_text("CREATE ROLE x;\n")
    (out / "databases.json").write_text(
        json.dumps([{"name": "advmfr_hr", "owner": "advmfr"}, {"name": "seo", "owner": "seo"}])
    )
    (out / "advmfr_hr.dump").write_bytes(b"x")
    (out / "seo.dump").write_bytes(b"x")


def _exists_cmd(name: str) -> tuple[str, ...]:
    # La requête EXISTS_SQL est passée sur stdin, pas en argument.
    return ("psql", "-v", f"name={name}", "-X", "-At", "-d", "postgres")


def test_restore_sequence(ctx: Context) -> None:
    _seed_backup(ctx)
    r = RecordingRunner()
    r.add(("pg_isready",), returncode=0)
    r.add(_exists_cmd("advmfr_hr"), stdout="1\n")
    r.add(_exists_cmd("seo"), stdout="")
    ctx.runner = r
    m = PostgresModule()
    res = m.restore(ctx, NullProgress())
    assert res.status == "ok", res.message
    out = ctx.dest / "postgres"
    assert list(_exists_cmd("advmfr_hr")) in r.calls
    assert r.mutations == [
        ["psql", "-X", "-d", "postgres", "-v", "ON_ERROR_STOP=0", "-f", str(out / "globals.sql")],
        [
            "pg_restore",
            "--no-owner",
            "--role=advmfr",
            "-d",
            "advmfr_hr",
            str(out / "advmfr_hr.dump"),
        ],
        ["createdb", "-O", "seo", "seo"],
        ["pg_restore", "--no-owner", "--role=seo", "-d", "seo", str(out / "seo.dump")],
    ]
    assert not r.has_call("brew", "services", "start")


def test_restore_starts_service_when_down(ctx: Context) -> None:
    _seed_backup(ctx)
    r = RecordingRunner()
    r.add(("pg_isready",), returncode=2)
    ctx.runner = r
    m = PostgresModule()
    m.ready_attempts = 2
    m.ready_delay = 0
    res = m.restore(ctx, NullProgress())
    assert res.status == "failed" and "injoignable" in res.message
    assert ["brew", "services", "start", "postgresql@18"] in r.mutations
    assert not r.has_call("pg_restore")


def test_restore_continues_when_one_db_fails(ctx: Context) -> None:
    _seed_backup(ctx)
    (ctx.dest / "postgres" / "seo.dump").unlink()
    r = RecordingRunner()
    r.add(("pg_isready",), returncode=0)
    ctx.runner = r
    res = PostgresModule().restore(ctx, NullProgress())
    assert res.status == "ok" and res.items == 1
    assert any("seo" in w for w in res.warnings)
    assert r.has_call("pg_restore", "--no-owner", "--role=advmfr")


def test_restore_ok_when_pg_restore_reports_errors(ctx: Context) -> None:
    """pg_restore sort en 1 dès qu'une erreur est ignorée : les données sont restaurées."""
    out = ctx.dest / "postgres"
    out.mkdir()
    (out / "databases.json").write_text(json.dumps([{"name": "seo", "owner": "seo"}]))
    (out / "seo.dump").write_bytes(b"x")
    r = RecordingRunner()
    r.add(("pg_isready",), returncode=0)
    r.add(("pg_restore",), returncode=1, stderr="pg_restore: errors ignored on restore: 3")
    ctx.runner = r
    res = PostgresModule().restore(ctx, NullProgress())
    assert res.status == "ok" and res.items == 1
    assert len(res.warnings) == 1
    assert "errors ignored on restore: 3" in res.warnings[0]


def test_restore_fails_when_nothing_restored(ctx: Context) -> None:
    _seed_backup(ctx)
    r = RecordingRunner()
    r.add(("pg_isready",), returncode=0)
    r.add(("createdb",), returncode=1, stderr="createdb: droits insuffisants")
    ctx.runner = r
    res = PostgresModule().restore(ctx, NullProgress())
    assert res.status == "failed" and "aucune base" in res.message
    assert not r.has_call("pg_restore")


def test_restore_dry_run_does_not_wait_for_service(
    ctx: Context, monkeypatch: pytest.MonkeyPatch
) -> None:
    _seed_backup(ctx)
    sleeps: list[float] = []
    monkeypatch.setattr(postgres.time, "sleep", lambda s: sleeps.append(s))
    ctx.dry_run = True
    r = RecordingRunner()
    r.add(("pg_isready",), returncode=2)
    ctx.runner = r
    res = PostgresModule().restore(ctx, NullProgress())
    assert res.status == "failed" and "injoignable" in res.message
    assert sleeps == []
    assert not r.has_call("pg_restore")


def test_restore_warns_when_globals_fail(ctx: Context) -> None:
    _seed_backup(ctx)
    r = RecordingRunner()
    r.add(("pg_isready",), returncode=0)
    r.add(("psql", "-X", "-d", "postgres"), returncode=1, stderr="ERROR: rôle déjà existant")
    ctx.runner = r
    res = PostgresModule().restore(ctx, NullProgress())
    assert res.status == "ok"
    assert any("globals.sql" in w for w in res.warnings)


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


class _InputRecordingRunner(RecordingRunner):
    """RecordingRunner n'enregistre pas `input` : on le capture ici pour le vérifier."""

    def __init__(self) -> None:
        super().__init__()
        self.inputs: list[tuple[list[str], str | None]] = []

    def run(self, cmd, **kw):  # type: ignore[no-untyped-def]
        self.inputs.append((list(cmd), kw.get("input")))
        return super().run(cmd, **kw)


def test_exists_sends_sql_on_stdin(ctx: Context) -> None:
    """psql n'interpole pas :'name' dans un -c : la requête doit passer par stdin."""
    r = _InputRecordingRunner()
    r.add(_exists_cmd("seo"), stdout="1\n")
    ctx.runner = r
    assert PostgresModule()._exists(ctx, "seo")
    assert (list(_exists_cmd("seo")), EXISTS_SQL) in r.inputs


def _postgres_is_live() -> bool:
    if shutil.which("psql") is None or shutil.which("pg_isready") is None:
        return False
    return subprocess.run(["pg_isready"], capture_output=True).returncode == 0


@pytest.mark.skipif(not _postgres_is_live(), reason="PostgreSQL local indisponible")
def test_exists_against_live_cluster(ctx: Context) -> None:
    """Smoke test lecture seule sur le cluster local : la requête doit vraiment marcher."""
    ctx.runner = RealRunner()
    module = PostgresModule()
    assert module._exists(ctx, "postgres")
    assert not module._exists(ctx, "definitely_missing_db_xyz")