"""Watermark resolution in the contract-fallback path (issue 02).

When live information_schema is unavailable, the extract manifest is built from the
generated source contracts. Those contracts store lowercase column identifiers, while
``extensions.watermark_columns`` keeps source case (e.g. ``SystemModstamp``). An
exact-match filter dropped every watermark in that path, silently — the live path
emitted watermarks, the fallback path emitted none.

Runnable without a test harness::

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

from __future__ import annotations

import pathlib
import sys
import tempfile

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

from ingestion.common.models import Column, Table
from ingestion.extract.tidbcloud.engine.config import SourceConfig
from ingestion.extract.tidbcloud.engine.manifest import load_tables_from_contracts


def test_property_matches_candidates_case_insensitively() -> None:
    """A lowercased column ('systemmodstamp') resolves a source-cased candidate
    ('SystemModstamp'), returning the column's own casing."""
    table = Table(
        database="nexuscrm",
        name="account",
        rows_estimate=None,
        columns=[
            Column(name="id", data_type="bigint", column_type="bigint", nullable=False, ordinal_position=1),
            Column(name="systemmodstamp", data_type="timestamp", column_type="timestamp", nullable=True, ordinal_position=2),
        ],
        primary_key=["id"],
        watermark_candidates=["SystemModstamp"],
    )
    assert table.watermark_columns == ["systemmodstamp"], table.watermark_columns


def test_unknown_candidate_is_still_dropped() -> None:
    """The safety filter still drops a candidate that is not a column at all."""
    table = Table(
        database="nexuscrm",
        name="account",
        rows_estimate=None,
        columns=[Column(name="id", data_type="bigint", column_type="bigint", nullable=False, ordinal_position=1)],
        primary_key=["id"],
        watermark_candidates=["NotAColumn"],
    )
    assert table.watermark_columns == []


CONTRACT_YAML = """\
source:
  identifiers:
    database: nexuscrm
    table: account
schema:
  columns:
    - name: id
      type: bigint
      nullable: false
    - name: systemmodstamp
      type: timestamp(6)
      nullable: true
extensions:
  primary_key: [id]
  watermark_columns: [SystemModstamp]
governance:
  classifications: [internal]
"""


def test_load_tables_from_contracts_yields_watermarks() -> None:
    """Building a Table from a generated source contract yields a non-empty
    watermark_columns despite the contract's lowercase column names."""
    with tempfile.TemporaryDirectory() as tmp:
        root = pathlib.Path(tmp)
        generated = root / "contracts" / "generated"
        generated.mkdir(parents=True)
        (generated / "nexuscrm.account.s3-snapshot.source.contract.yaml").write_text(CONTRACT_YAML)

        config = SourceConfig(
            path=root / "source.yaml",
            data={"source": {"name": "nexuscrm", "database": "nexuscrm"}},
        )
        tables = load_tables_from_contracts(config)

    assert len(tables) == 1, tables
    assert tables[0].watermark_columns == ["systemmodstamp"], tables[0].watermark_columns


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