fxhnt/tests/integration/test_warehouse_migrate.py

"""Phase 2: memory-bounded, idempotent, verified DuckDB -> TimescaleDB migration.

The migration STREAMS the DuckDB `features` table in bounded chunks (per symbol, capped at `chunk_rows`
via a server-side cursor) and bulk-upserts into a `TimescaleFeatureStore`, so RAM stays bounded for the
~5-10M-row warehouse (a prior naive whole-table load OOM'd a 2Gi node). The destination here is the store
pointed at in-memory SQLite (no Postgres needed in CI) — the chunked DuckDB read + ON CONFLICT upsert run
for real; only the hypertable/compression block no-ops off-Postgres.

Asserts: row-count + per-feature checksum verification passes; the destination's panels equal the source
DuckDB's (end-to-end parity post-migration); re-running is idempotent (no duplication); and the chunking
path is actually used (multiple chunks at a small chunk_rows)."""
from __future__ import annotations

import duckdb
import pytest

from fxhnt.adapters.warehouse import DuckDbFeatureStore, TimescaleFeatureStore
from fxhnt.application.warehouse_migrate import migrate_duckdb_to_timescale
from fxhnt.ports.warehouse import FeatureRow

# A few symbols x several features x several days. ts is epoch-SECONDS aligned to day starts so
# epoch_day = ts // 86400 is exact. Non-USDT AAPL must round-trip too (full table, not just crypto).
_FIXTURE: list[tuple[str, list[FeatureRow]]] = [
    ("BTCUSDT", [
        (19000 * 86400, {"close": 100.0, "funding": 0.0001, "spot_close": 99.5, "volume": 10.0}),
        (19001 * 86400, {"close": 110.0, "funding": 0.0002, "spot_close": 108.0, "volume": 12.0}),
        (19002 * 86400, {"close": 120.0, "funding": 0.0003, "spot_close": 118.0, "volume": 14.0}),
    ]),
    ("ETHUSDT", [
        (19000 * 86400, {"close": 50.0, "funding": -0.0001, "spot_close": 49.0, "volume": 20.0}),
        (19001 * 86400, {"close": 55.0, "funding": 0.0003, "spot_close": 54.5, "volume": 22.0}),
    ]),
    ("AAPL", [
        (19000 * 86400, {"close": 200.0, "volume": 5.0}),
        (19001 * 86400, {"close": 205.0, "volume": 6.0}),
    ]),
]

_MEMBERSHIP = [
    ("BTCUSDT", "BINANCE", "2020-01-01", ""),
    ("AAPL", "NASDAQ", "1980-12-12", "2026-01-01"),
]


@pytest.fixture()
def duckdb_path(tmp_path) -> str:
    path = str(tmp_path / "warehouse.duckdb")
    src = DuckDbFeatureStore(path)
    src.write_features_bulk(_FIXTURE)
    src.upsert_membership(_MEMBERSHIP)
    src.close()
    return path


def _total_rows_duckdb(path: str) -> int:
    con = duckdb.connect(path, read_only=True)
    try:
        return con.execute("SELECT count(*) FROM features").fetchone()[0]
    finally:
        con.close()


def test_migration_verification_passes(duckdb_path) -> None:
    dest = TimescaleFeatureStore("sqlite://")
    report = migrate_duckdb_to_timescale(duckdb_path, dest, chunk_rows=200_000)
    assert report["verified"] is True
    assert report["features_count_duckdb"] == report["features_count_timescale"]
    assert report["features_count_duckdb"] == _total_rows_duckdb(duckdb_path)
    assert report["membership_rows"] == len(_MEMBERSHIP)
    # per-feature checksum table present and every feature reconciles
    assert report["per_feature"], "per-feature checksum report missing"
    for feat, block in report["per_feature"].items():
        assert block["count_duckdb"] == block["count_timescale"], feat
        assert block["sum_duckdb"] == pytest.approx(block["sum_timescale"], abs=1e-6), feat
    dest.close()


def test_post_migration_panel_parity(duckdb_path) -> None:
    """End-to-end: after migration the destination's panels/reads equal the source DuckDB's."""
    dest = TimescaleFeatureStore("sqlite://")
    migrate_duckdb_to_timescale(duckdb_path, dest)  # opens DuckDB read-only, then closes it
    src = DuckDbFeatureStore(duckdb_path)           # open the source for comparison AFTER migration

    assert dest.crypto_close_panel() == src.crypto_close_panel()
    assert dest.crypto_funding_panel() == src.crypto_funding_panel()
    assert dest.crypto_spot_panel() == src.crypto_spot_panel()
    syms = ["BTCUSDT", "ETHUSDT", "AAPL"]
    assert dest.read_panel(syms, "close") == src.read_panel(syms, "close")
    assert dest.read_panel(syms, "volume") == src.read_panel(syms, "volume")
    assert dest.read_features("BTCUSDT") == src.read_features("BTCUSDT")
    assert dest.catalog() == src.catalog()
    assert dest.read_membership() == src.read_membership()
    src.close()
    dest.close()


def test_migration_is_idempotent(duckdb_path) -> None:
    """Re-running yields the same destination state — no duplicated rows, same verification pass."""
    dest = TimescaleFeatureStore("sqlite://")
    r1 = migrate_duckdb_to_timescale(duckdb_path, dest)
    r2 = migrate_duckdb_to_timescale(duckdb_path, dest)
    assert r1["features_count_timescale"] == r2["features_count_timescale"]
    assert r2["verified"] is True
    # row count on the destination unchanged after the second run (ON CONFLICT overwrite, no dupes)
    assert r2["features_count_timescale"] == _total_rows_duckdb(duckdb_path)
    dest.close()


def test_chunking_path_is_used(duckdb_path) -> None:
    """With a tiny chunk_rows the stream must process MULTIPLE bounded chunks (never a whole-table load)."""
    dest = TimescaleFeatureStore("sqlite://")
    report = migrate_duckdb_to_timescale(duckdb_path, dest, chunk_rows=2)
    # fixture has 9+9+4 = far more than 2 rows -> several chunks; never one giant chunk
    assert report["chunks"] > 1
    assert report["max_chunk_rows"] <= 2
    # still correct + verified despite the tiny chunks
    assert report["verified"] is True
    assert report["features_count_timescale"] == _total_rows_duckdb(duckdb_path)
    dest.close()


def test_migration_accepts_dsn_string(duckdb_path) -> None:
    """The function also accepts a DSN string (cluster path), constructing the store itself."""
    report = migrate_duckdb_to_timescale(duckdb_path, "sqlite://")
    assert report["verified"] is True
    assert report["features_count_timescale"] == _total_rows_duckdb(duckdb_path)