"""
Tests for the postgres dump executor.

The point of this executor is a backup that is actually restorable, so most
of these cover the failure paths: an empty TOC behind a successful dump, one
database failing while the others are kept, retention touching only its own
files, and a per-database timeout actually firing. Every subprocess call is
mocked — this suite must pass with postgres-shared unreachable (D-26).
"""
import asyncio
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from src.config import Settings
from src.executors import pg_dump_executor as pde


def _config(tmp_path, **over):
    cfg = {
        "container": "postgres-shared",
        "databases": "all",
        "exclude": [],
        "globals": False,  # most tests don't exercise the globals path; see TestGlobals
        "output_dir": str(tmp_path),
        "retention_days": 14,
        "timeout_per_db": 900,
    }
    cfg.update(over)
    return cfg


async def _writing_dump(container, db, destination, timeout):
    """A fake _dump_database that behaves like the real one: it writes bytes
    to the destination it is given."""
    destination.write_bytes(f"dump-of-{db}".encode())


@pytest.mark.executor
@pytest.mark.unit
class TestConfigValidation:

    @pytest.mark.asyncio
    async def test_missing_container_rejected(self, test_settings: Settings, tmp_path):
        with pytest.raises(ValueError, match="container"):
            await pde.execute(_config(tmp_path, container=""), test_settings)

    @pytest.mark.asyncio
    @pytest.mark.parametrize("bad", [0, -1, "14", None, True])
    async def test_bad_retention_rejected(self, bad, test_settings: Settings, tmp_path):
        with pytest.raises(ValueError, match="retention_days"):
            await pde.execute(_config(tmp_path, retention_days=bad), test_settings)

    @pytest.mark.asyncio
    @pytest.mark.parametrize("bad", [0, -1, "900", None, True])
    async def test_bad_timeout_per_db_rejected(self, bad, test_settings: Settings, tmp_path):
        with pytest.raises(ValueError, match="timeout_per_db"):
            await pde.execute(_config(tmp_path, timeout_per_db=bad), test_settings)

    @pytest.mark.asyncio
    async def test_bad_databases_value_rejected(self, test_settings: Settings, tmp_path):
        with pytest.raises(ValueError, match="databases"):
            await pde.execute(_config(tmp_path, databases={"not": "valid"}), test_settings)

    @pytest.mark.asyncio
    async def test_everything_excluded_and_no_globals_rejected(self, test_settings: Settings, tmp_path):
        with patch.object(pde, "_discover_databases", AsyncMock(return_value=["a", "b"])):
            with pytest.raises(ValueError, match="nothing to do"):
                await pde.execute(
                    _config(tmp_path, exclude=["a", "b"], globals=False), test_settings
                )


@pytest.mark.executor
@pytest.mark.unit
class TestDiscovery:

    @pytest.mark.asyncio
    async def test_discover_databases_parses_rows(self):
        proc = MagicMock()
        proc.communicate = AsyncMock(return_value=(b"alpha\nbeta\n\n", b""))
        proc.returncode = 0
        with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)) as mock_exec:
            result = await pde._discover_databases("postgres-shared")
        assert result == ["alpha", "beta"]
        args = mock_exec.call_args.args
        assert args[:3] == ("docker", "exec", "postgres-shared")
        assert "psql" in args

    @pytest.mark.asyncio
    async def test_discover_databases_failure_raises(self):
        proc = MagicMock()
        proc.communicate = AsyncMock(return_value=(b"", b"connection refused"))
        proc.returncode = 1
        with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
            with pytest.raises(Exception, match="database discovery failed"):
                await pde._discover_databases("postgres-shared")

    @pytest.mark.asyncio
    async def test_all_minus_excludes_selects_correct_targets(self, test_settings: Settings, tmp_path):
        attempted = []

        async def fake_dump(container, db, destination, timeout):
            attempted.append(db)
            destination.write_bytes(b"x")

        with patch.object(pde, "_discover_databases", AsyncMock(
                return_value=["keep", "template0", "test_scheduler"])), \
             patch.object(pde, "_dump_database", fake_dump), \
             patch.object(pde, "_verify_archive", AsyncMock(return_value=1)):
            await pde.execute(
                _config(tmp_path, exclude=["template0", "test_scheduler"]), test_settings
            )

        assert attempted == ["keep"]

    @pytest.mark.asyncio
    async def test_explicit_database_list_also_respects_exclude(self, test_settings: Settings, tmp_path):
        attempted = []

        async def fake_dump(container, db, destination, timeout):
            attempted.append(db)
            destination.write_bytes(b"x")

        with patch.object(pde, "_discover_databases", AsyncMock()) as discover, \
             patch.object(pde, "_dump_database", fake_dump), \
             patch.object(pde, "_verify_archive", AsyncMock(return_value=1)):
            await pde.execute(
                _config(tmp_path, databases=["one", "two"], exclude=["two"]), test_settings
            )

        discover.assert_not_called()  # explicit list means no discovery call at all
        assert attempted == ["one"]


@pytest.mark.executor
@pytest.mark.unit
class TestDumpBehaviour:

    @pytest.mark.asyncio
    async def test_writes_verified_archive_per_database_mode_0600(self, test_settings: Settings, tmp_path):
        with patch.object(pde, "_discover_databases", AsyncMock(return_value=["alpha", "beta"])), \
             patch.object(pde, "_dump_database", _writing_dump), \
             patch.object(pde, "_verify_archive", AsyncMock(return_value=5)):
            result = await pde.execute(_config(tmp_path), test_settings)

        files = sorted(tmp_path.glob("*.dump"))
        assert [f.name.split("-")[0] for f in files] == ["alpha", "beta"]
        assert "5 entries" in result
        for f in files:
            assert oct(f.stat().st_mode)[-3:] == "600"
        # no .partial left behind
        assert list(tmp_path.glob("*.partial")) == []

    @pytest.mark.asyncio
    async def test_one_failing_database_fails_run_but_keeps_others(self, test_settings: Settings, tmp_path):
        async def fake_dump(container, db, destination, timeout):
            if db == "bad":
                raise Exception("pg_dump exploded")
            destination.write_bytes(b"x")

        with patch.object(pde, "_discover_databases", AsyncMock(return_value=["good", "bad"])), \
             patch.object(pde, "_dump_database", fake_dump), \
             patch.object(pde, "_verify_archive", AsyncMock(return_value=3)):
            with pytest.raises(Exception, match="FAILED bad: pg_dump exploded"):
                await pde.execute(_config(tmp_path), test_settings)

        files = list(tmp_path.glob("*.dump"))
        assert len(files) == 1
        assert files[0].name.startswith("good-")
        assert list(tmp_path.glob("*.partial")) == []

    @pytest.mark.asyncio
    async def test_empty_toc_deletes_partial_and_counts_as_failure(self, test_settings: Settings, tmp_path):
        async def fake_verify(path):
            raise Exception("archive has zero TOC entries")

        with patch.object(pde, "_discover_databases", AsyncMock(return_value=["alpha"])), \
             patch.object(pde, "_dump_database", _writing_dump), \
             patch.object(pde, "_verify_archive", fake_verify):
            with pytest.raises(Exception, match="FAILED alpha: archive has zero TOC entries"):
                await pde.execute(_config(tmp_path), test_settings)

        assert list(tmp_path.glob("*.dump")) == []
        assert list(tmp_path.glob("*.partial")) == []


@pytest.mark.executor
@pytest.mark.unit
class TestGlobals:

    @pytest.mark.asyncio
    async def test_globals_dump_written_and_mode_0600(self, test_settings: Settings, tmp_path):
        async def fake_globals(container, destination, timeout):
            destination.write_bytes(b"CREATE ROLE postgres;")

        with patch.object(pde, "_discover_databases", AsyncMock(return_value=["alpha"])), \
             patch.object(pde, "_dump_database", _writing_dump), \
             patch.object(pde, "_verify_archive", AsyncMock(return_value=2)), \
             patch.object(pde, "_dump_globals", fake_globals):
            result = await pde.execute(_config(tmp_path, globals=True), test_settings)

        files = list(tmp_path.glob("globals-*.sql"))
        assert len(files) == 1
        assert oct(files[0].stat().st_mode)[-3:] == "600"
        assert "globals" in result

    @pytest.mark.asyncio
    async def test_empty_globals_dump_is_a_failure_but_keeps_db_dumps(self, test_settings: Settings, tmp_path):
        async def fake_globals(container, destination, timeout):
            destination.write_bytes(b"")  # empty

        with patch.object(pde, "_discover_databases", AsyncMock(return_value=["alpha"])), \
             patch.object(pde, "_dump_database", _writing_dump), \
             patch.object(pde, "_verify_archive", AsyncMock(return_value=2)), \
             patch.object(pde, "_dump_globals", fake_globals):
            with pytest.raises(Exception, match="FAILED globals: globals dump is empty"):
                await pde.execute(_config(tmp_path, globals=True), test_settings)

        assert len(list(tmp_path.glob("*.dump"))) == 1   # alpha kept
        assert list(tmp_path.glob("globals-*.sql")) == []
        assert list(tmp_path.glob("*.partial")) == []


@pytest.mark.executor
@pytest.mark.unit
class TestVerifyArchive:

    def test_toc_entry_count_counts_non_comment_lines(self):
        output = (
            ";\n"
            "; Archive created at 2026-10-19 09:45:07 UTC\n"
            ";     TOC Entries: 2\n"
            ";\n"
            "216; 1259 44268 TABLE public scheduled_tasks test_scheduler_user\n"
            "215; 1259 44267 SEQUENCE public scheduled_tasks_id_seq test_scheduler_user\n"
        )
        assert pde._toc_entry_count(output) == 2

    def test_toc_entry_count_zero_for_header_only(self):
        output = (
            ";\n"
            "; Archive created at 2026-10-19 09:45:07 UTC\n"
            ";     TOC Entries: 0\n"
            ";\n"
        )
        assert pde._toc_entry_count(output) == 0

    @pytest.mark.asyncio
    async def test_verify_archive_raises_on_nonzero_exit(self, tmp_path):
        proc = MagicMock()
        proc.communicate = AsyncMock(return_value=(b"", b"not an archive"))
        proc.returncode = 1
        with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
            with pytest.raises(Exception, match="pg_restore --list failed"):
                await pde._verify_archive(tmp_path / "broken.dump")

    @pytest.mark.asyncio
    async def test_verify_archive_raises_on_zero_entries(self, tmp_path):
        proc = MagicMock()
        proc.communicate = AsyncMock(return_value=(b";\n;     TOC Entries: 0\n;\n", b""))
        proc.returncode = 0
        with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
            with pytest.raises(Exception, match="zero TOC entries"):
                await pde._verify_archive(tmp_path / "empty.dump")


@pytest.mark.executor
@pytest.mark.unit
class TestTimeout:

    @pytest.mark.asyncio
    async def test_dump_times_out_per_database(self, tmp_path):
        """The actual asyncio.wait_for/kill mechanism, not a stand-in for it."""

        class HangingStdout:
            async def read(self, n):
                await asyncio.sleep(10)
                return b""

        class EmptyStderr:
            async def read(self):
                return b""

        fake_proc = MagicMock()
        fake_proc.stdout = HangingStdout()
        fake_proc.stderr = EmptyStderr()
        fake_proc.kill = MagicMock()
        fake_proc.wait = AsyncMock(return_value=0)

        with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=fake_proc)):
            with pytest.raises(Exception, match="timed out after"):
                await pde._dump_database(
                    "postgres-shared", "alpha", tmp_path / "alpha.partial", timeout=0.05
                )

        fake_proc.kill.assert_called_once()

    @pytest.mark.asyncio
    async def test_timeout_failure_handled_like_any_other_db_failure(self, test_settings: Settings, tmp_path):
        async def fake_dump(container, db, destination, timeout):
            if db == "slow":
                raise Exception(f"timed out after {timeout}s: pg_dump")
            destination.write_bytes(b"x")

        with patch.object(pde, "_discover_databases", AsyncMock(return_value=["fast", "slow"])), \
             patch.object(pde, "_dump_database", fake_dump), \
             patch.object(pde, "_verify_archive", AsyncMock(return_value=1)):
            with pytest.raises(Exception, match="FAILED slow: timed out after"):
                await pde.execute(_config(tmp_path, timeout_per_db=1), test_settings)

        files = list(tmp_path.glob("*.dump"))
        assert len(files) == 1 and files[0].name.startswith("fast-")


@pytest.mark.executor
@pytest.mark.unit
class TestRetention:

    def _age(self, path, days):
        import os
        old = (datetime.now(timezone.utc) - timedelta(days=days)).timestamp()
        os.utime(path, (old, old))

    def test_prunes_only_past_the_window(self, tmp_path):
        fresh = tmp_path / "alpha-20261019T094500Z.dump"
        stale = tmp_path / "alpha-20260101T120000Z.dump"
        for f in (fresh, stale):
            f.write_bytes(b"x")
        self._age(stale, 45)

        assert pde._prune(tmp_path, 14) == 1
        assert fresh.exists() and not stale.exists()

    def test_leaves_unrelated_files_alone(self, tmp_path):
        """Retention must not touch anything it did not write."""
        other = tmp_path / "important-database-dump.tar.gz"
        named_alike = tmp_path / "alpha-backup-manual.dump"
        for f in (other, named_alike):
            f.write_bytes(b"x")
            self._age(f, 400)

        assert pde._prune(tmp_path, 14) == 0
        assert other.exists() and named_alike.exists()

    def test_prune_leaves_partial_files_alone(self, tmp_path):
        """A .partial is never this executor's finished output; retention
        must never be the thing that deletes one out from under a dump that
        is still writing it."""
        partial = tmp_path / "alpha-20260101T120000Z.dump.partial"
        partial.write_bytes(b"x")
        self._age(partial, 400)

        assert pde._prune(tmp_path, 14) == 0
        assert partial.exists()

    def test_prunes_globals_files_too(self, tmp_path):
        stale = tmp_path / "globals-20260101T120000Z.sql"
        stale.write_bytes(b"x")
        self._age(stale, 400)

        assert pde._prune(tmp_path, 14) == 1
        assert not stale.exists()