"""Re-run safety tests for the S3 bulk loader (issue 01).

The loader must be idempotent: loading a ``run_id`` converges on exactly one copy of
that run's rows however many times it runs, and a crash between COPY and stamp on a
prior run must not let the next run mislabel orphaned rows.

These tests drive ``load_table`` against a fake Databend cursor that records the SQL it
is asked to run, so we can assert the COPY → stamp sequence is preceded by the right
cleanup without a live warehouse.

Runnable without a test harness::

    .venv/bin/python tests/test_load_rerun_safety.py

The ``test_*`` functions are also pytest-discoverable if pytest is added later.
"""

from __future__ import annotations

import pathlib
import sys
from types import SimpleNamespace

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))

from ingestion.load.s3_bulk_load.engine import load as load_mod
from ingestion.load.s3_bulk_load.engine.config import TableTargets


class FakeCursor:
    """Records executed SQL; optionally raises when a statement matches ``fail_on``."""

    def __init__(self, fail_on: str | None = None) -> None:
        self.executed: list[str] = []
        self._fail_on = fail_on

    def execute(self, sql: str) -> None:
        self.executed.append(sql)
        if self._fail_on and self._fail_on in sql:
            raise RuntimeError("simulated warehouse failure")

    def fetchone(self):  # noqa: ANN201 - count_rows / count_duplicate_keys read [0]
        return (0,)


class FakeClient:
    def __init__(self, cursor: FakeCursor) -> None:
        self._cursor = cursor

    def cursor(self) -> FakeCursor:
        return self._cursor


def _manifest() -> dict:
    return {
        "run_id": "run-123",
        "run_date": "2026-05-26",
        "s3_uri": "s3://bucket/export/",
        "runtime_log_uri": "s3://bucket/runs/2026-05-26/run-123/events.jsonl",
        "source_name": "nexuscrm",
        "tables": [
            {
                "table_name": "account",
                "source_database": "nexuscrm",
                "primary_key": ["id"],
                "classifications": [],
                "watermark_columns": [],
                "file_count": 1,
                "total_bytes": 100,
                "columns": [
                    {
                        "name": "id",
                        "data_type": "bigint",
                        "column_type": "bigint",
                        "nullable": False,
                        "ordinal_position": 1,
                    }
                ],
            }
        ],
    }


def _config() -> SimpleNamespace:
    # Only the attributes load_table touches (COPY credentials + error sanitisation).
    return SimpleNamespace(
        s3_access_key_id="AKIAEXAMPLE",
        s3_secret_access_key="secret",
        aws_region="us-west-2",
        s3_credential_env_names=["S3_ACCESS_KEY_ID", "S3_SECRET_ACCESS_KEY"],
    )


def _targets() -> dict[str, TableTargets]:
    return {
        "account": TableTargets(
            table="account",
            raw_table="nexuscrm_raw.account",
            classifications=[],
            product_ids=[],
        )
    }


class _Patched:
    """Point the loader at a fake cursor and capture runtime events instead of S3 writes."""

    def __init__(self, cursor: FakeCursor, events: list[dict]) -> None:
        self._cursor = cursor
        self._events = events

    def __enter__(self) -> "_Patched":
        self._orig_client = load_mod.client
        self._orig_write_event = load_mod.write_event
        load_mod.client = lambda config: FakeClient(self._cursor)
        load_mod.write_event = lambda config, manifest, event: (
            self._events.append(event) or {"local_path": "x", "s3_uri": "y"}
        )
        return self

    def __exit__(self, *exc) -> None:
        load_mod.client = self._orig_client
        load_mod.write_event = self._orig_write_event


def _delete_indices(cursor: FakeCursor) -> list[int]:
    return [i for i, sql in enumerate(cursor.executed) if sql.strip().startswith("DELETE")]


def _copy_indices(cursor: FakeCursor) -> list[int]:
    return [i for i, sql in enumerate(cursor.executed) if "COPY INTO" in sql]


def test_default_rerun_clears_run_rows_before_copy() -> None:
    """Re-running the same run_id always deletes that run's rows before COPY, so
    FORCE = TRUE cannot double rows. No --replace-run flag is needed."""
    cursor = FakeCursor()
    events: list[dict] = []
    with _Patched(cursor, events):
        manifest, config, targets = _manifest(), _config(), _targets()
        load_mod.load_table(config, manifest, "account", targets)  # initial load
        load_mod.load_table(config, manifest, "account", targets)  # re-run

    deletes, copies = _delete_indices(cursor), _copy_indices(cursor)
    assert len(deletes) == 2, f"expected a cleanup DELETE on every run, got {deletes}"
    assert len(copies) == 2, f"expected a COPY on every run, got {copies}"
    assert deletes[0] < copies[0] and deletes[1] < copies[1], "DELETE must precede COPY each run"


def test_cleanup_is_scoped_to_run_and_orphaned_rows() -> None:
    """The cleanup deletes only this run_id's rows plus orphaned NULL-run_id rows left by
    a crash between COPY and stamp on a prior run — never the whole table (other runs)."""
    cursor = FakeCursor()
    events: list[dict] = []
    with _Patched(cursor, events):
        load_mod.load_table(_config(), _manifest(), "account", _targets())

    delete_sql = cursor.executed[_delete_indices(cursor)[0]]
    assert "'run-123'" in delete_sql, "cleanup must target this run_id"
    assert "IS NULL" in delete_sql, "cleanup must also remove crash-orphaned unstamped rows"


def test_failed_table_emits_failure_event_and_reraises() -> None:
    """A failure mid-load emits a 'failed' runtime event (so a partial run is observable)
    and re-raises to abort the run."""
    cursor = FakeCursor(fail_on="COPY INTO")
    events: list[dict] = []
    raised = False
    with _Patched(cursor, events):
        try:
            load_mod.load_table(_config(), _manifest(), "account", _targets())
        except RuntimeError:
            raised = True

    assert raised, "load_table should re-raise after a COPY failure"
    assert any(e.get("status") == "failed" for e in events), "expected a 'failed' runtime event"


if __name__ == "__main__":
    cases = [value for name, value in sorted(globals().items()) if name.startswith("test_") and callable(value)]
    for case in cases:
        case()
        print(f"ok  {case.__name__}")
    print(f"\n{len(cases)} passed")