"""Single source of truth for the extract table set + schema-drift surfacing (issue 07).

Two divergent queries decided "which tables": one drove contracts/manifest, one drove the
export filter. They were meant to be identical but nothing enforced it. And the manifest
silently dropped tables with metadata-but-no-files / files-but-no-metadata. This pins:

1. inspect_tables and selected_table_names derive the same set from kept_base_table_names.
2. schema_drift surfaces both mismatch directions.

Runnable without a test harness::

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

from __future__ import annotations

import pathlib
import sys

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

from ingestion.extract.tidbcloud.engine import source_db
from ingestion.extract.tidbcloud.engine.config import SourceConfig
from ingestion.extract.tidbcloud.engine.manifest import schema_drift

# Fake information_schema: account/contact/_System_Config have PKs; eventlog has none.
_CATALOG_NAMES = ["account", "contact", "_System_Config", "eventlog"]
_WITH_PK = {"account", "contact", "_System_Config"}


class FakeCursor:
    """Answers the information_schema queries source_db issues, by SQL shape."""

    def __init__(self) -> None:
        self._rows: list[dict] = []

    def execute(self, sql: str, params: tuple = ()) -> None:
        s = " ".join(sql.split())
        if "FROM information_schema.TABLES" in s and "TABLE_ROWS" in s:
            wanted = set(params[1:])
            self._rows = [{"TABLE_NAME": n, "TABLE_ROWS": 10} for n in _CATALOG_NAMES if n in wanted]
        elif "FROM information_schema.TABLES" in s:
            self._rows = [{"TABLE_NAME": n} for n in _CATALOG_NAMES]
        elif "TABLE_CONSTRAINTS" in s and "DISTINCT" in s:
            wanted = set(params[1:])
            self._rows = [{"TABLE_NAME": n} for n in _WITH_PK if n in wanted]
        elif "information_schema.COLUMNS" in s:
            wanted = set(params[1:])
            self._rows = [
                {"TABLE_NAME": n, "COLUMN_NAME": "Id", "DATA_TYPE": "bigint",
                 "COLUMN_TYPE": "bigint", "IS_NULLABLE": "NO", "ORDINAL_POSITION": 1}
                for n in _CATALOG_NAMES if n in wanted
            ]
        elif "KEY_COLUMN_USAGE" in s:
            wanted = set(params[1:])
            self._rows = [{"TABLE_NAME": n, "COLUMN_NAME": "Id"} for n in _WITH_PK if n in wanted]
        else:
            self._rows = []

    def fetchall(self) -> list[dict]:
        return self._rows

    def __enter__(self) -> "FakeCursor":
        return self

    def __exit__(self, *exc) -> None:
        return None


class FakeConn:
    def cursor(self) -> FakeCursor:
        return FakeCursor()

    def __enter__(self) -> "FakeConn":
        return self

    def __exit__(self, *exc) -> None:
        return None


def _config() -> SourceConfig:
    return SourceConfig(
        path=pathlib.Path("source.yaml"),
        data={
            "source": {"name": "nexuscrm", "database": "nexuscrm", "connection_env": "X"},
            "tables": {"require_primary_key": True},
        },
    )


def _with_fake_connect(fn):
    original = source_db.connect
    source_db.connect = lambda config: FakeConn()
    try:
        return fn()
    finally:
        source_db.connect = original


def test_inspect_and_filter_share_the_same_set() -> None:
    config = _config()
    inspected = _with_fake_connect(lambda: [t.name for t in source_db.inspect_tables(config)])
    filtered = _with_fake_connect(lambda: source_db.selected_table_names(config))
    # eventlog has no PK and require_primary_key is on, so both drop it — identically.
    assert filtered == ["_System_Config", "account", "contact"]
    assert sorted(inspected) == filtered, "contract set and export-filter set must match"


def test_schema_drift_reports_both_directions() -> None:
    drift = schema_drift(exported={"account", "contact", "orphan"}, expected={"account", "contact", "ghost"})
    assert drift == {"files_without_metadata": ["orphan"], "metadata_without_files": ["ghost"]}


def test_schema_drift_clean_when_aligned() -> None:
    drift = schema_drift(exported={"account", "contact"}, expected={"account", "contact"})
    assert drift == {"files_without_metadata": [], "metadata_without_files": []}


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