"""Re-run safety + COPY shape for the GCS bulk loader.

The GCS loader shares the raw-snapshot idempotency contract with the S3 loader (clear this
run's rows before a FORCE=TRUE COPY; scope the delete to the run_id plus crash-orphaned
NULL rows), and adds GCS specifics worth pinning: the COPY reads the per-table ``gcs://``
prefix from the manifest (Google's ``gs://`` translated to Databend's ``gcs://`` scheme),
and a failed COPY's error is sanitised so the base64 credential never reaches a log.

Runnable without a test harness::

    .venv/bin/python tests/test_gcs_load_rerun_safety.py
"""

from __future__ import annotations

import os
import pathlib
import sys
from types import SimpleNamespace

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

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


class FakeCursor:
    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(f"simulated warehouse failure for credential {os.environ.get('GCS_CRED','')}")

    def fetchone(self):
        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-27",
        "runtime_log_uri": "gs://bucket/logs/run-123/events.jsonl",
        "source_name": "metadb",
        "dataset": "metadb_new",
        "tables": [
            {
                "table_name": "invoices",
                "source_database": "metadb_new",
                "primary_key": ["id"],
                "classifications": [],
                "watermark_columns": [],
                "file_count": 2,
                "total_bytes": 100,
                "gcs_uri": "gs://bucket/snap/run_id=run-123/table=invoices/",
                "columns": [
                    {"name": "id", "data_type": "string", "column_type": "STRING",
                     "nullable": True, "ordinal_position": 1},
                    {"name": "amount", "data_type": "bignumeric", "column_type": "BIGNUMERIC",
                     "nullable": True, "ordinal_position": 2},
                ],
            }
        ],
    }


def _config() -> SimpleNamespace:
    return SimpleNamespace(
        gcs_credential="SUPER-SECRET-BASE64",
        gcs_endpoint_url=None,
        gcs_credential_env_names=["GCS_CRED"],
    )


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


class _Patched:
    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", "gcs_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_databend_gcs_uri_translates_scheme() -> None:
    assert load_mod.databend_gcs_uri("gs://b/p/") == "gcs://b/p/"
    assert load_mod.databend_gcs_uri("gcs://b/p/") == "gcs://b/p/"


def test_rerun_clears_run_rows_before_copy() -> None:
    cursor = FakeCursor()
    with _Patched(cursor, []):
        load_mod.load_table(_config(), _manifest(), "invoices", _targets())
        load_mod.load_table(_config(), _manifest(), "invoices", _targets())
    deletes, copies = _delete_indices(cursor), _copy_indices(cursor)
    assert len(deletes) == 2 and len(copies) == 2
    assert deletes[0] < copies[0] and deletes[1] < copies[1], "DELETE must precede COPY each run"


def test_cleanup_scoped_to_run_and_orphans() -> None:
    cursor = FakeCursor()
    with _Patched(cursor, []):
        load_mod.load_table(_config(), _manifest(), "invoices", _targets())
    delete_sql = cursor.executed[_delete_indices(cursor)[0]]
    assert "'run-123'" in delete_sql and "IS NULL" in delete_sql


def test_copy_reads_per_table_gcs_prefix_with_gcs_scheme() -> None:
    cursor = FakeCursor()
    with _Patched(cursor, []):
        load_mod.load_table(_config(), _manifest(), "invoices", _targets())
    copy_sql = cursor.executed[_copy_indices(cursor)[0]]
    assert "FROM 'gcs://bucket/snap/run_id=run-123/table=invoices/'" in copy_sql, copy_sql
    assert "CREDENTIAL = 'SUPER-SECRET-BASE64'" in copy_sql
    assert "FILE_FORMAT = (TYPE = PARQUET)" in copy_sql and "FORCE = TRUE" in copy_sql


def test_failed_copy_emits_event_and_sanitises_credential() -> None:
    os.environ["GCS_CRED"] = "SUPER-SECRET-BASE64"
    cursor = FakeCursor(fail_on="COPY INTO")
    events: list[dict] = []
    raised_msg = ""
    try:
        with _Patched(cursor, events):
            load_mod.load_table(_config(), _manifest(), "invoices", _targets())
    except RuntimeError as error:
        raised_msg = str(error)
    finally:
        del os.environ["GCS_CRED"]
    assert raised_msg, "load_table should re-raise after a COPY failure"
    assert "SUPER-SECRET-BASE64" not in raised_msg, "credential must be sanitised from the error"
    assert "<GCS_CRED>" in raised_msg
    assert any(e.get("status") == "failed" for e in events)


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")