import os

import asyncpg
import pytest

from ktrade.db.pool import create_pool, dsn, migrate
from ktrade.ingestion.backfill import detect_gaps, upsert_candles
from ktrade.ingestion.stream import _parse_ts
from ktrade.kraken.models import normalize_symbol

pytestmark = pytest.mark.skipif(os.environ.get("KTRADE_SKIP_DB") == "1", reason="db tests disabled")


def test_symbol_normalisation():
    assert normalize_symbol("XBT/USD") == "BTC/USD"
    assert normalize_symbol("XDG/EUR") == "DOGE/EUR"
    assert normalize_symbol("ETH/USD") == "ETH/USD"


def test_rfc3339_nanoseconds():
    assert abs(_parse_ts("2026-10-22T03:11:19.123456789Z") - 1792638679.123456) < 1e-3


async def test_closed_flag_never_regresses_and_gaps_detected():
    url = os.environ["KTRADE_DATABASE_URL"]
    conn = await asyncpg.connect(dsn(url))
    await conn.execute("DROP SCHEMA public CASCADE; CREATE SCHEMA public")
    await conn.close()
    await migrate(url)
    pool = await create_pool(url)
    try:
        async with pool.acquire() as c:
            await upsert_candles(c, "T/USD", 1, [(60, 1, 1, 1, 1, 1, 1, 1, True), (120, 1, 1, 1, 2, 1, 1, 1, False)], "rest")
            # a later poll reports bar 120 as closed with the final close → overwritten
            await upsert_candles(c, "T/USD", 1, [(120, 1, 1, 1, 3, 1, 1, 1, True)], "ws")
            # a stale 'forming' snapshot must NOT overwrite a closed bar
            await upsert_candles(c, "T/USD", 1, [(120, 1, 1, 1, 9, 1, 1, 1, False)], "rest")
            # gap: bars 180, 240 missing
            await upsert_candles(c, "T/USD", 1, [(300, 1, 1, 1, 1, 1, 1, 1, True)], "rest")
        rows = await pool.fetch("SELECT ts, close, closed FROM candles WHERE symbol='T/USD' ORDER BY ts")
        assert [(r["ts"], r["close"], r["closed"]) for r in rows] == [(60, 1.0, True), (120, 3.0, True), (300, 1.0, True)]
        assert await detect_gaps(pool, "T/USD", 1) == 1
        g = await pool.fetchrow("SELECT gap_start, gap_end, missing_bars FROM data_gaps")
        assert (g["gap_start"], g["gap_end"], g["missing_bars"]) == (180, 240, 2)
    finally:
        await pool.close()