"""Tests for record_yandex_price_history (gate price.previous/trend → offer_price_history).

Behavioural tests run against an in-memory SQLite DB with a minimal listings +
offer_price_history schema. The service keeps all timestamp arithmetic in Python and
uses only CAST(:x AS type) (no :x::type, no Postgres-only interval), so the same SQL
exercises correctly on SQLite — letting us assert the stateful idempotent logic
(seed previous-point, current-point, no-op on unchanged price) without a live Postgres.

Plus static checks that the emitted SQL keeps psycopg-v3 discipline.
"""

from __future__ import annotations

import inspect
import os
import re

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import pytest
from scraper_kit.base import ScrapedLot
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session, sessionmaker

from app.services import yandex_price_history as yph

_SERVICE_SRC = inspect.getsource(yph)
# Concatenated text of every actual SQL statement the service emits (not the docstring,
# which may mention the :x::type anti-pattern in prose).
_SQL_TEXT = "\n".join(
    str(getattr(stmt, "text", "")) for stmt in (yph._RESOLVE_SQL, yph._LATEST_SQL, yph._INSERT_SQL)
)


# ── Static: psycopg-v3 discipline ─────────────────────────────────────────────


def test_service_uses_cast_not_colon_colon() -> None:
    """No :x::type shorthand in emitted SQL — only CAST(:x AS type) (psycopg-v3 rule)."""
    # No bind-param cast shorthand (:param::type) in the real SQL statements.
    assert not re.search(r":\w+::", _SQL_TEXT)
    assert "CAST(:listing_id AS bigint)" in _SQL_TEXT
    assert "CAST(:price_rub AS numeric)" in _SQL_TEXT


def test_service_inserts_with_yandex_source_tag() -> None:
    assert "INSERT INTO offer_price_history" in _SERVICE_SRC
    assert "'yandex'" in _SERVICE_SRC


# ── Behavioural: in-memory SQLite ─────────────────────────────────────────────


@pytest.fixture()
def db() -> Session:
    """In-memory SQLite session with minimal listings + offer_price_history schema."""
    engine = create_engine("sqlite://", future=True)
    with engine.begin() as conn:
        conn.execute(
            text(
                """
                CREATE TABLE listings (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    source TEXT NOT NULL,
                    source_id TEXT
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE offer_price_history (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    listing_id INTEGER NOT NULL,
                    change_time TEXT NOT NULL,
                    price_rub NUMERIC NOT NULL,
                    source TEXT
                )
                """
            )
        )
    sess = sessionmaker(bind=engine, future=True)()
    try:
        yield sess
    finally:
        sess.close()


def _seed_listing(db: Session, *, source: str, source_id: str) -> int:
    db.execute(
        text("INSERT INTO listings (source, source_id) VALUES (:s, :sid)"),
        {"s": source, "sid": source_id},
    )
    db.commit()
    row = db.execute(
        text("SELECT id FROM listings WHERE source = :s AND source_id = :sid"),
        {"s": source, "sid": source_id},
    ).fetchone()
    assert row is not None
    return int(row[0])


def _lot(
    *,
    source_id: str,
    price_rub: int,
    price_previous_rub: int | None = None,
    source: str = "yandex",
) -> ScrapedLot:
    return ScrapedLot(
        source=source,
        source_url=f"https://realty.yandex.ru/offer/{source_id}/",
        source_id=source_id,
        price_rub=price_rub,
        price_previous_rub=price_previous_rub,
    )


def _history(db: Session, listing_id: int) -> list[tuple[float, str]]:
    rows = db.execute(
        text(
            "SELECT price_rub, change_time FROM offer_price_history "
            "WHERE listing_id = :lid ORDER BY change_time ASC"
        ),
        {"lid": listing_id},
    ).fetchall()
    return [(float(r[0]), str(r[1])) for r in rows]


def test_first_capture_with_previous_inserts_two_points(db: Session) -> None:
    """price_previous задан и != price_rub → previous@-1d + current@now (2 строки)."""
    lid = _seed_listing(db, source="yandex", source_id="111")
    n = yph.record_yandex_price_history(
        db, [_lot(source_id="111", price_rub=5_000_000, price_previous_rub=5_500_000)]
    )
    assert n == 2
    hist = _history(db, lid)
    assert len(hist) == 2
    # Chronological: previous-point first (earlier change_time), current second.
    assert hist[0][0] == 5_500_000.0  # previous
    assert hist[1][0] == 5_000_000.0  # current
    assert hist[0][1] < hist[1][1]  # previous change_time strictly earlier


def test_first_capture_without_previous_inserts_one_point(db: Session) -> None:
    """price_previous отсутствует → только current-точка (1 строка)."""
    lid = _seed_listing(db, source="yandex", source_id="222")
    n = yph.record_yandex_price_history(db, [_lot(source_id="222", price_rub=4_000_000)])
    assert n == 1
    hist = _history(db, lid)
    assert len(hist) == 1
    assert hist[0][0] == 4_000_000.0


def test_first_capture_previous_equals_current_inserts_one_point(db: Session) -> None:
    """price_previous == price_rub → previous-seed пропущена, только current (1 строка)."""
    lid = _seed_listing(db, source="yandex", source_id="223")
    n = yph.record_yandex_price_history(
        db, [_lot(source_id="223", price_rub=4_000_000, price_previous_rub=4_000_000)]
    )
    assert n == 1
    assert len(_history(db, lid)) == 1


def test_recapture_same_price_is_noop(db: Session) -> None:
    """Повторный захват с той же ценой → 0 строк (idempotent)."""
    _seed_listing(db, source="yandex", source_id="333")
    first = yph.record_yandex_price_history(db, [_lot(source_id="333", price_rub=6_000_000)])
    assert first == 1
    second = yph.record_yandex_price_history(db, [_lot(source_id="333", price_rub=6_000_000)])
    assert second == 0


def test_recapture_new_price_inserts_one_point(db: Session) -> None:
    """Повторный захват с новой ценой → 1 строка (зафиксировали смену)."""
    lid = _seed_listing(db, source="yandex", source_id="444")
    yph.record_yandex_price_history(db, [_lot(source_id="444", price_rub=6_000_000)])
    n = yph.record_yandex_price_history(db, [_lot(source_id="444", price_rub=5_800_000)])
    assert n == 1
    hist = _history(db, lid)
    assert len(hist) == 2
    assert {h[0] for h in hist} == {6_000_000.0, 5_800_000.0}


def test_non_yandex_lot_skipped(db: Session) -> None:
    """Не-yandex lot пропускается (даже если листинг с таким source_id есть)."""
    _seed_listing(db, source="cian", source_id="555")
    n = yph.record_yandex_price_history(
        db, [_lot(source_id="555", price_rub=3_000_000, source="cian")]
    )
    assert n == 0


def test_unknown_source_id_skipped(db: Session) -> None:
    """source_id без листинга в БД → skip (листинг не найден)."""
    n = yph.record_yandex_price_history(db, [_lot(source_id="does-not-exist", price_rub=1_000_000)])
    assert n == 0


def test_lot_without_source_id_skipped(db: Session) -> None:
    lot = ScrapedLot(
        source="yandex",
        source_url="https://realty.yandex.ru/offer/x/",
        source_id=None,
        price_rub=2_000_000,
    )
    assert yph.record_yandex_price_history(db, [lot]) == 0


def test_empty_lots_returns_zero(db: Session) -> None:
    assert yph.record_yandex_price_history(db, []) == 0


def test_broken_lot_does_not_abort_batch(db: Session, monkeypatch: pytest.MonkeyPatch) -> None:
    """SAVEPOINT: один битый lot (INSERT падает) не валит остальные."""
    good_lid = _seed_listing(db, source="yandex", source_id="aaa")
    _seed_listing(db, source="yandex", source_id="bbb")

    real_insert = yph._insert_point

    def _flaky(db_: Session, **kw: object) -> None:
        # Ломаем только для листинга 'bbb' (его listing_id заранее не знаем —
        # рвём по price_rub-значению, привязанному к bbb).
        if kw.get("price_rub") == 999:
            raise RuntimeError("boom")
        real_insert(db_, **kw)  # type: ignore[arg-type]

    monkeypatch.setattr(yph, "_insert_point", _flaky)

    lots = [
        _lot(source_id="aaa", price_rub=7_000_000),
        _lot(source_id="bbb", price_rub=999),
    ]
    n = yph.record_yandex_price_history(db, lots)
    # 'aaa' inserted (1), 'bbb' rolled back via SAVEPOINT → batch survives.
    assert n == 1
    assert len(_history(db, good_lid)) == 1