"""
Regression tests for automatic live schema synchronization.

Covers: INFORMATION_SCHEMA introspection, structural change detection
(new/removed tables and columns, type changes, FK changes), no-op refresh
when nothing changed, failure keeping the last known good schema, on-disk
cache restore at startup, non-blocking periodic refresh, and atomic
snapshot swaps for concurrent readers.
"""
from __future__ import annotations

import asyncio
import json
import threading
import time
from unittest.mock import MagicMock

from app.core.config import Settings, get_settings
from app.schema import introspection as introspection_module
from app.schema import sync as sync_module
from app.schema.introspection import introspect_schema
from app.schema.loader import ColumnInfo, ForeignKeyInfo, TableInfo, load_schema_contract
from app.schema.registry import SchemaRegistry, build_snapshot, build_snapshot_from_tables
from app.schema.sync import CHANGED, FAILED, UNCHANGED, SchemaSyncManager, _fingerprint


def _make_tables() -> dict[str, TableInfo]:
    invoices = TableInfo(name="invoices")
    invoices.columns["id"] = ColumnInfo(name="id", type="int(11)", nullable=False, primary_key=True)
    invoices.columns["total"] = ColumnInfo(name="total", type="decimal(10,2)", nullable=True)
    invoices.columns["customer_id"] = ColumnInfo(name="customer_id", type="int(11)", nullable=True)
    invoices.primary_keys = ["id"]
    invoices.foreign_keys.append(
        ForeignKeyInfo(
            constrained_columns=["customer_id"],
            referred_table="customers",
            referred_columns=["id"],
        )
    )
    customers = TableInfo(name="customers")
    customers.columns["id"] = ColumnInfo(name="id", type="int(11)", nullable=False, primary_key=True)
    customers.columns["name"] = ColumnInfo(name="name", type="varchar(255)", nullable=True)
    customers.primary_keys = ["id"]
    return {"invoices": invoices, "customers": customers}


def _extended_tables() -> dict[str, TableInfo]:
    tables = _make_tables()
    expense_entries = TableInfo(name="expense_entries")
    expense_entries.columns["id"] = ColumnInfo(name="id", type="int(11)", nullable=False, primary_key=True)
    expense_entries.columns["amount"] = ColumnInfo(name="amount", type="decimal(10,2)", nullable=True)
    expense_entries.primary_keys = ["id"]
    tables["expense_entries"] = expense_entries
    return tables


def _make_manager(tmp_path) -> SchemaSyncManager:
    settings = Settings(
        schema_customizations_path=str(tmp_path / "missing_customizations.json"),
        schema_cache_path=str(tmp_path / "cache" / "live_schema.json"),
    )
    return SchemaSyncManager(
        settings=settings,
        registry=SchemaRegistry(settings=settings),
        engine_factory=lambda: MagicMock(),
        cache_path=tmp_path / "cache" / "live_schema.json",
    )


# ---------------------------------------------------------------------------
# Introspection
# ---------------------------------------------------------------------------


def test_introspection_builds_tableinfo_from_information_schema_rows(monkeypatch):
    monkeypatch.setattr(introspection_module, "_fetch_table_names", lambda conn: [("invoices",), ("customers",)])
    monkeypatch.setattr(
        introspection_module,
        "_fetch_column_rows",
        lambda conn: [
            ("invoices", "id", "int(11)", "NO", "PRI", None),
            ("invoices", "total", "decimal(10,2)", "YES", "", None),
            ("customers", "id", "int(11)", "NO", "PRI", None),
        ],
    )
    monkeypatch.setattr(
        introspection_module,
        "_fetch_primary_key_rows",
        lambda conn: [("invoices", "id"), ("customers", "id")],
    )
    monkeypatch.setattr(
        introspection_module,
        "_fetch_foreign_key_rows",
        lambda conn: [("invoices", "customer_id", "customers", "id", "fk_invoices_customers")],
    )

    raw = introspect_schema(MagicMock())

    assert set(raw) == {"invoices", "customers"}
    invoices = raw["invoices"]
    assert invoices.columns["total"].type == "decimal(10,2)"
    assert invoices.columns["total"].nullable is True
    assert invoices.primary_keys == ["id"]
    assert invoices.columns["id"].primary_key is True
    assert len(invoices.foreign_keys) == 1
    assert invoices.foreign_keys[0].referred_table == "customers"
    assert invoices.foreign_keys[0].constrained_columns == ["customer_id"]
    assert invoices.foreign_keys[0].referred_columns == ["id"]


def test_introspection_ignores_views_and_non_base_tables(monkeypatch):
    monkeypatch.setattr(introspection_module, "_fetch_table_names", lambda conn: [("invoices",)])
    monkeypatch.setattr(introspection_module, "_fetch_column_rows", lambda conn: [("invoices", "id", "int(11)", "NO", "PRI", None)])
    monkeypatch.setattr(introspection_module, "_fetch_primary_key_rows", lambda conn: [("invoices", "id")])
    monkeypatch.setattr(introspection_module, "_fetch_foreign_key_rows", lambda conn: [])

    raw = introspect_schema(MagicMock())
    assert set(raw) == {"invoices"}


# ---------------------------------------------------------------------------
# Change detection
# ---------------------------------------------------------------------------


def test_fingerprint_detects_new_table():
    assert _fingerprint(_make_tables()) != _fingerprint(_extended_tables())


def test_fingerprint_detects_new_column():
    changed = _make_tables()
    changed["invoices"].columns["discount"] = ColumnInfo(name="discount", type="decimal(10,2)", nullable=True)
    assert _fingerprint(_make_tables()) != _fingerprint(changed)


def test_fingerprint_detects_removed_column():
    changed = _make_tables()
    del changed["invoices"].columns["total"]
    assert _fingerprint(_make_tables()) != _fingerprint(changed)


def test_fingerprint_detects_changed_column_type():
    changed = _make_tables()
    changed["invoices"].columns["total"].type = "decimal(12,2)"
    assert _fingerprint(_make_tables()) != _fingerprint(changed)


def test_fingerprint_detects_primary_key_change():
    changed = _make_tables()
    changed["invoices"].primary_keys = []
    changed["invoices"].columns["id"].primary_key = False
    assert _fingerprint(_make_tables()) != _fingerprint(changed)


def test_fingerprint_detects_foreign_key_change():
    changed = _make_tables()
    changed["invoices"].foreign_keys = []
    assert _fingerprint(_make_tables()) != _fingerprint(changed)


def test_fingerprint_ignores_column_defaults():
    changed = _make_tables()
    changed["invoices"].columns["total"].default = "0.00"
    assert _fingerprint(_make_tables()) == _fingerprint(changed)


# ---------------------------------------------------------------------------
# Manager behavior
# ---------------------------------------------------------------------------


def test_refresh_installs_new_table_into_registry(tmp_path, monkeypatch):
    manager = _make_manager(tmp_path)
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _make_tables())
    assert manager.refresh() == CHANGED
    assert not manager._registry.snapshot.has_table("expense_entries")

    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _extended_tables())
    assert manager.refresh() == CHANGED
    assert manager._registry.snapshot.has_table("expense_entries")
    assert manager._registry.snapshot.has_column("expense_entries", "amount")


def test_unchanged_schema_does_not_rebuild_snapshot(tmp_path, monkeypatch):
    manager = _make_manager(tmp_path)
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _make_tables())
    assert manager.refresh() == CHANGED
    first = manager._registry.snapshot

    assert manager.refresh() == UNCHANGED
    assert manager._registry.snapshot is first


def test_failed_refresh_keeps_previous_valid_schema(tmp_path, monkeypatch):
    manager = _make_manager(tmp_path)
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _make_tables())
    assert manager.refresh() == CHANGED
    good = manager._registry.snapshot

    def unreachable(engine):
        raise RuntimeError("database unreachable")

    monkeypatch.setattr(sync_module, "introspect_schema", unreachable)
    assert manager.refresh() == FAILED
    assert manager._registry.snapshot is good

    # Recovery on the next successful cycle does not rebuild unnecessarily.
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _make_tables())
    assert manager.refresh() == UNCHANGED
    assert manager._registry.snapshot is good


def test_cache_is_written_and_restored_at_startup(tmp_path, monkeypatch):
    monkeypatch.setattr(sync_module, "_STARTUP_REFRESH_BACKOFF_SECONDS", 0.0)
    cache_path = tmp_path / "cache" / "live_schema.json"
    first = _make_manager(tmp_path)
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _make_tables())
    assert first.refresh() == CHANGED
    assert cache_path.is_file()
    payload = json.loads(cache_path.read_text(encoding="utf-8"))
    assert set(payload["tables"]) == {"invoices", "customers"}
    assert payload["fingerprint"]

    # A fresh process starts with the database unreachable: the cached
    # last-known-good live schema must be served instead of the contract.
    def unreachable(engine):
        raise RuntimeError("database unreachable")

    monkeypatch.setattr(sync_module, "introspect_schema", unreachable)
    second = _make_manager(tmp_path)
    assert asyncio.run(second.startup_refresh()) == FAILED
    snapshot = second._registry.snapshot
    assert snapshot.has_table("invoices")
    assert snapshot.has_column("invoices", "total")

    # Restored fingerprint means the next live check (unchanged) skips rebuild.
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _make_tables())
    assert second.refresh() == UNCHANGED
    assert second._registry.snapshot is snapshot


def test_startup_without_cache_keeps_contract_when_db_down(tmp_path, monkeypatch):
    monkeypatch.setattr(sync_module, "_STARTUP_REFRESH_BACKOFF_SECONDS", 0.0)
    registry = SchemaRegistry()
    contract_snapshot = registry.refresh(get_settings().schema_source_path)

    def unreachable(engine):
        raise RuntimeError("database unreachable")

    monkeypatch.setattr(sync_module, "introspect_schema", unreachable)
    manager = SchemaSyncManager(
        settings=Settings(
            schema_customizations_path=str(tmp_path / "missing.json"),
            schema_cache_path=str(tmp_path / "cache" / "live_schema.json"),
        ),
        registry=registry,
        engine_factory=lambda: MagicMock(),
        cache_path=tmp_path / "cache" / "live_schema.json",
    )
    assert asyncio.run(manager.startup_refresh()) == FAILED
    assert manager._registry.snapshot is contract_snapshot


def test_corrupt_cache_is_ignored_and_contract_stays(tmp_path, monkeypatch):
    monkeypatch.setattr(sync_module, "_STARTUP_REFRESH_BACKOFF_SECONDS", 0.0)
    registry = SchemaRegistry()
    contract_snapshot = registry.refresh(get_settings().schema_source_path)
    cache_path = tmp_path / "cache" / "live_schema.json"
    cache_path.parent.mkdir(parents=True, exist_ok=True)
    cache_path.write_text("{not valid json", encoding="utf-8")

    def unreachable(engine):
        raise RuntimeError("database unreachable")

    monkeypatch.setattr(sync_module, "introspect_schema", unreachable)
    manager = SchemaSyncManager(
        settings=Settings(
            schema_customizations_path=str(tmp_path / "missing.json"),
            schema_cache_path=str(cache_path),
        ),
        registry=registry,
        engine_factory=lambda: MagicMock(),
        cache_path=cache_path,
    )
    assert asyncio.run(manager.startup_refresh()) == FAILED
    assert manager._registry.snapshot is contract_snapshot


# ---------------------------------------------------------------------------
# Non-blocking + lifecycle
# ---------------------------------------------------------------------------


async def test_refresh_runs_off_the_event_loop(tmp_path, monkeypatch):
    manager = _make_manager(tmp_path)

    def slow_introspect(engine):
        time.sleep(0.4)
        return _make_tables()

    monkeypatch.setattr(sync_module, "introspect_schema", slow_introspect)
    task = asyncio.create_task(asyncio.to_thread(manager.refresh))
    # If the refresh ran on the event loop, this sleep could not return until
    # the 0.4s refresh finished. Returning while it is still in flight proves
    # the loop stays free for API requests during a refresh.
    await asyncio.sleep(0.1)
    assert not task.done()
    await task
    assert task.result() == CHANGED


async def test_start_periodic_is_idempotent_and_stops_cleanly(tmp_path, monkeypatch):
    manager = _make_manager(tmp_path)
    manager.start_periodic()
    task = manager._task
    assert task is not None and not task.done()
    manager.start_periodic()
    assert manager._task is task
    await manager.stop_periodic()
    assert manager._task is None


# ---------------------------------------------------------------------------
# Snapshot consistency
# ---------------------------------------------------------------------------


def test_old_snapshot_reference_remains_complete_after_refresh(tmp_path, monkeypatch):
    manager = _make_manager(tmp_path)
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _make_tables())
    assert manager.refresh() == CHANGED
    first = manager._registry.snapshot

    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _extended_tables())
    assert manager.refresh() == CHANGED
    second = manager._registry.snapshot

    assert first is not second
    # Readers holding the old reference still see a complete, valid schema.
    assert first.has_table("invoices")
    assert first.has_column("invoices", "total")
    assert first.find_join_path("invoices", "customers") is not None
    assert second.has_table("expense_entries")


def test_concurrent_readers_never_observe_partial_schema(tmp_path, monkeypatch):
    manager = _make_manager(tmp_path)
    registry = manager._registry
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _make_tables())
    assert manager.refresh() == CHANGED

    variants = [_make_tables(), _extended_tables()]
    stop = threading.Event()
    anomalies: list[str] = []

    def reader() -> None:
        valid_shapes = {("customers", "invoices"), ("customers", "expense_entries", "invoices")}
        while not stop.is_set():
            snapshot = registry.snapshot
            shape = tuple(sorted(snapshot.tables))
            if shape not in valid_shapes:
                anomalies.append(f"partial or invalid snapshot shape: {shape}")

    def writer() -> None:
        for i in range(60):
            registry.refresh_from_tables(variants[i % len(variants)])

    reader_thread = threading.Thread(target=reader)
    writer_thread = threading.Thread(target=writer)
    reader_thread.start()
    writer_thread.start()
    writer_thread.join()
    stop.set()
    reader_thread.join()

    assert anomalies == []


# ---------------------------------------------------------------------------
# Pipeline parity + configuration
# ---------------------------------------------------------------------------


def test_build_snapshot_from_tables_matches_contract_pipeline():
    settings = get_settings()
    raw_tables = load_schema_contract(settings.schema_source_path)

    from_contract = build_snapshot(settings.schema_source_path, settings.schema_customizations_path)
    from_tables = build_snapshot_from_tables(raw_tables, settings.schema_customizations_path)

    assert set(from_tables.tables) == set(from_contract.tables)
    assert len(from_tables.relationships) == len(from_contract.relationships)
    assert from_tables.has_column("vendors", "company_name")


def test_schema_sync_config_defaults():
    from app.core.config import _PROJECT_ROOT

    settings = get_settings()
    assert settings.schema_refresh_interval_seconds == 1800
    # The default cache path is rooted at the project root (never the CWD),
    # so the sync manager writes/reads the same file from any working dir.
    assert settings.schema_cache_path == str((_PROJECT_ROOT / "schema_cache" / "live_schema.json").resolve())


# ---------------------------------------------------------------------------
# Stale schema customizations (live path must degrade, not break)
# ---------------------------------------------------------------------------


def _stale_customizations_file(tmp_path) -> str:
    """An overlay whose entries reference tables/columns that do not exist in
    `_make_tables()`, alongside valid entries that must still be applied."""
    path = tmp_path / "stale_customizations.json"
    path.write_text(
        json.dumps(
            {
                "tables": {
                    "ghost_table": {"description": "Ghost table"},
                    "invoices": {
                        "description": "Invoice records (custom)",
                        "platform_column": "ghost_platform_col",
                        "columns": {
                            "total": {"description": "Invoice total (custom)"},
                            "ghost_col": {"description": "Ghost column"},
                        },
                    },
                },
                "computed_fields": {
                    "profit": {
                        "table": "ghost_table",
                        "description": "Profit",
                        "aliases": ["profit"],
                        "builder": "net_sum",
                        "components": [{"table": "invoices", "value_column": "total", "sign": "add"}],
                    }
                },
            }
        ),
        encoding="utf-8",
    )
    return str(path)


def test_live_refresh_degrades_stale_customizations_and_keeps_syncing(tmp_path, monkeypatch):
    """A customization referencing a table/column the live DB no longer has
    must be skipped (with a log) rather than failing the whole refresh, and
    synchronization must keep working on subsequent cycles."""
    manager = _make_manager(tmp_path)
    manager._settings.schema_customizations_path = _stale_customizations_file(tmp_path)
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _make_tables())

    assert manager.refresh() == CHANGED
    snapshot = manager._registry.snapshot

    # Valid entries were applied...
    assert snapshot.tables["invoices"].description == "Invoice records (custom)"
    assert snapshot.tables["invoices"].columns["total"].description == "Invoice total (custom)"
    # ...stale ones were skipped, never invented.
    assert "ghost_table" not in snapshot.tables
    assert "ghost_col" not in snapshot.tables["invoices"].columns
    assert snapshot.tables["invoices"].platform_column is None
    assert all("profit" not in table.columns for table in snapshot.tables.values())

    # The same live structure now compares equal -> sync continues normally
    # (no fingerprint lock-out, no repeated failures).
    assert manager.refresh() == UNCHANGED


def test_malformed_customizations_fail_refresh_and_keep_last_known_good(tmp_path, monkeypatch):
    """Malformed configuration (invalid JSON) is not staleness: it must fail
    the cycle and keep the previously installed schema."""
    manager = _make_manager(tmp_path)
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _make_tables())
    assert manager.refresh() == CHANGED
    good = manager._registry.snapshot

    # A refresh only rebuilds when the live structure actually changed, so
    # simulate a real schema change (new table) to force the rebuild that
    # would hit the malformed config.
    bad_path = tmp_path / "bad_customizations.json"
    bad_path.write_text("{not valid json", encoding="utf-8")
    manager._settings.schema_customizations_path = str(bad_path)
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _extended_tables())
    assert manager.refresh() == FAILED
    assert manager._registry.snapshot is good

    # Fixing the file lets the next cycle succeed again.
    manager._settings.schema_customizations_path = _stale_customizations_file(tmp_path)
    assert manager.refresh() == CHANGED


# ---------------------------------------------------------------------------
# Retry/backoff after a failed refresh
# ---------------------------------------------------------------------------


def test_next_check_delay_backoff_sequence():
    assert sync_module._next_check_delay(1800, 0) == 1800.0
    assert sync_module._next_check_delay(1800, 1) == 30.0
    assert sync_module._next_check_delay(1800, 2) == 60.0
    assert sync_module._next_check_delay(1800, 3) == 120.0
    assert sync_module._next_check_delay(1800, 6) == 960.0
    # Doubling is capped at the configured interval.
    assert sync_module._next_check_delay(1800, 7) == 1800.0
    assert sync_module._next_check_delay(1800, 100) == 1800.0
    # The minimum-interval clamp wins over the backoff base.
    assert sync_module._next_check_delay(10, 1) == 10.0


async def test_periodic_loop_backs_off_after_failure_and_resets_after_success(tmp_path, monkeypatch):
    """The loop must retry quickly (30s, 60s, ...) after a failed refresh and
    return to the normal interval once a refresh succeeds - including when
    startup itself failed (`retry_soon=True`)."""
    manager = _make_manager(tmp_path)
    sleeps: list[float] = []
    real_sleep = asyncio.sleep  # captured before the patch below
    original_sleep = sync_module.asyncio.sleep

    async def _capture_sleep(delay):
        sleeps.append(delay)
        await original_sleep(0)

    monkeypatch.setattr(sync_module.asyncio, "sleep", _capture_sleep)

    calls = {"n": 0}

    def _refresh_sequence():
        calls["n"] += 1
        return FAILED if calls["n"] == 1 else CHANGED

    monkeypatch.setattr(manager, "refresh", _refresh_sequence)

    manager.start_periodic(retry_soon=True)
    for _ in range(20):
        await real_sleep(0.01)
    await manager.stop_periodic()

    # retry_soon -> 30s; one failure -> 60s; success -> back to interval.
    assert sleeps[:4] == [30.0, 60.0, 1800.0, 1800.0]


# ---------------------------------------------------------------------------
# Bounded quick-retry for the startup refresh (a DB starting up alongside
# the app - e.g. docker-compose - should not have to wait for the periodic
# loop's own backoff schedule to become reachable).
# ---------------------------------------------------------------------------


def test_startup_refresh_retries_and_succeeds_before_attempts_exhausted(tmp_path, monkeypatch):
    """A database that only becomes reachable on a later attempt must be
    picked up within `startup_refresh` itself, without falling through to
    the periodic loop."""
    monkeypatch.setattr(sync_module, "_STARTUP_REFRESH_BACKOFF_SECONDS", 0.0)
    manager = _make_manager(tmp_path)

    attempts = {"n": 0}

    def _flaky_introspect(engine):
        attempts["n"] += 1
        if attempts["n"] < sync_module._STARTUP_REFRESH_MAX_ATTEMPTS:
            raise RuntimeError("database unreachable")
        return _make_tables()

    monkeypatch.setattr(sync_module, "introspect_schema", _flaky_introspect)

    result = asyncio.run(manager.startup_refresh())

    assert result == CHANGED
    assert attempts["n"] == sync_module._STARTUP_REFRESH_MAX_ATTEMPTS
    assert manager._registry.snapshot.has_table("invoices")


def test_startup_refresh_gives_up_after_max_attempts(tmp_path, monkeypatch):
    """Once every bounded attempt has failed, `startup_refresh` must give up
    (never crash-loop) and let the caller's periodic loop take over."""
    monkeypatch.setattr(sync_module, "_STARTUP_REFRESH_BACKOFF_SECONDS", 0.0)
    manager = _make_manager(tmp_path)

    attempts = {"n": 0}

    def _always_fails(engine):
        attempts["n"] += 1
        raise RuntimeError("database unreachable")

    monkeypatch.setattr(sync_module, "introspect_schema", _always_fails)

    result = asyncio.run(manager.startup_refresh())

    assert result == FAILED
    assert attempts["n"] == sync_module._STARTUP_REFRESH_MAX_ATTEMPTS


def test_startup_refresh_backoff_doubles_between_attempts(tmp_path, monkeypatch):
    """Delay between bounded startup attempts must double each time,
    starting at `_STARTUP_REFRESH_BACKOFF_SECONDS`."""
    manager = _make_manager(tmp_path)
    sleeps: list[float] = []

    async def _capture_sleep(delay):
        sleeps.append(delay)

    monkeypatch.setattr(sync_module.asyncio, "sleep", _capture_sleep)
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: (_ for _ in ()).throw(RuntimeError("down")))

    assert asyncio.run(manager.startup_refresh()) == FAILED
    # One fewer sleep than attempts: no sleep needed after the last attempt.
    assert len(sleeps) == sync_module._STARTUP_REFRESH_MAX_ATTEMPTS - 1
    assert sleeps[0] == sync_module._STARTUP_REFRESH_BACKOFF_SECONDS
    for earlier, later in zip(sleeps, sleeps[1:]):
        assert later == earlier * 2


def test_startup_refresh_still_restores_cache_before_retrying(tmp_path, monkeypatch):
    """The cache-restore step must run exactly once regardless of how many
    live-refresh attempts follow it."""
    monkeypatch.setattr(sync_module, "_STARTUP_REFRESH_BACKOFF_SECONDS", 0.0)
    cache_path = tmp_path / "cache" / "live_schema.json"
    first = _make_manager(tmp_path)
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _make_tables())
    assert first.refresh() == CHANGED
    assert cache_path.is_file()

    def _always_fails(engine):
        raise RuntimeError("database unreachable")

    monkeypatch.setattr(sync_module, "introspect_schema", _always_fails)
    second = _make_manager(tmp_path)
    assert asyncio.run(second.startup_refresh()) == FAILED

    # The cache was restored once (not re-loaded per attempt) and stays in
    # effect after every bounded attempt fails.
    snapshot = second._registry.snapshot
    assert snapshot.has_table("invoices")
    assert snapshot.has_column("invoices", "total")


# ---------------------------------------------------------------------------
# ConfigurationError is distinguished from generic connectivity/parse
# failures in both the periodic and startup refresh paths.
# ---------------------------------------------------------------------------


def test_refresh_logs_configuration_error_reason(tmp_path, monkeypatch, caplog):
    """A malformed (not merely stale) customization must be logged with its
    specific reason, not just a generic exception class name."""
    import logging

    manager = _make_manager(tmp_path)
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _make_tables())
    assert manager.refresh() == CHANGED

    bad_path = tmp_path / "bad_customizations.json"
    bad_path.write_text("{not valid json", encoding="utf-8")
    manager._settings.schema_customizations_path = str(bad_path)
    monkeypatch.setattr(sync_module, "introspect_schema", lambda engine: _extended_tables())

    with caplog.at_level(logging.WARNING):
        assert manager.refresh() == FAILED

    assert any(record.message == "schema_refresh_invalid_customization" for record in caplog.records)

# ---------------------------------------------------------------------------
# The declared classifications ride the live-schema path, and degrade honestly
# ---------------------------------------------------------------------------

def test_live_refresh_installs_classifications_and_a_dropped_column_disables_its_class(tmp_path, monkeypatch):
    """A live refresh carries the `classifications` overlay exactly like the
    contract path does. When the live database drops the column a class
    depends on, that class is skipped (the refresh still succeeds) and a
    question about it is unsupported - never quietly answered as the whole
    figure; when the column returns, so does the class."""
    import copy

    from app.clarification.service import next_clarification
    from app.planning.semantic_guards import availability_reason
    from app.schema.loader import load_schema_contract
    from app.schema.registry import SchemaRegistry
    import app.schema.sync as sync_mod

    settings = get_settings()
    registry = SchemaRegistry()
    registry.refresh(settings.schema_source_path)
    contract_tables = load_schema_contract(settings.schema_source_path)
    live = {"tables": copy.deepcopy(contract_tables)}
    monkeypatch.setattr(sync_mod, "introspect_schema", lambda engine: live["tables"])
    manager = SchemaSyncManager(settings=settings, registry=registry, engine_factory=lambda: object(), cache_path=tmp_path / "live.json")

    assert manager.refresh() == CHANGED
    assert set(registry.snapshot.classifications) == {
        "recurring_revenue", "recurring_expenses", "billing_currency", "customer_status", "credit_limit_breach",
    }
    assert availability_reason("What is our recurring revenue this year?", registry.snapshot) is None

    live["tables"] = copy.deepcopy(contract_tables)
    del live["tables"]["invoices"].columns["recurring_invoice_id"]
    assert manager.refresh() == CHANGED
    # Only the class whose column was dropped goes; the expense class keeps its own column.
    assert set(registry.snapshot.classifications) == {
        "recurring_expenses", "billing_currency", "customer_status", "credit_limit_breach",
    }
    reason = availability_reason("What is our recurring revenue this year?", registry.snapshot)
    assert reason is not None and "Recurring revenue data is not available" in reason
    assert next_clarification("What is recurring revenue vs. one-time revenue split?", registry.snapshot) is None
    assert availability_reason("How much revenue was billed in foreign currency?", registry.snapshot) is None

    live["tables"] = copy.deepcopy(contract_tables)
    assert manager.refresh() == CHANGED
    assert "recurring_revenue" in registry.snapshot.classifications
    assert availability_reason("What is our recurring revenue this year?", registry.snapshot) is None