gendesign/tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py

"""Tests for newbuilding_enrich_backfill task (#972 / 950-E1).

No real network, no real Postgres. The Cian fetch is mocked; the DB session is a
lightweight in-memory fake that models just enough of the three target tables to
assert (a) rows land and (b) an idempotent re-run does not duplicate them.
"""

from __future__ import annotations

import os
import sys
from unittest.mock import AsyncMock, MagicMock, patch

# DATABASE_URL required by config before any app import.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

# WeasyPrint stub — not installed in CI without GTK.
_wp_mock = MagicMock()
sys.modules.setdefault("weasyprint", _wp_mock)

import pytest  # noqa: E402
from scraper_kit.providers.cian.newbuilding import NewbuildingEnrichment  # noqa: E402

from app.tasks.newbuilding_enrich_backfill import (  # noqa: E402
    NewbuildingEnrichBackfillResult,
    _save_cian_reviews,
    backfill_newbuilding_enrichment,
    run_newbuilding_enrich,
)

# ---------------------------------------------------------------------------
# In-memory fake Session — models houses_price_dynamics / house_reliability_checks
# / house_reviews well enough for count + idempotency assertions.
# ---------------------------------------------------------------------------


class _FakeResult:
    def __init__(self, *, scalar=None, rows=None):
        self._scalar = scalar
        self._rows = rows or []

    def scalar_one(self):
        return self._scalar

    def mappings(self):
        return self

    def all(self):
        return self._rows


class FakeDB:
    """Minimal SQLAlchemy-Session stand-in keyed off SQL text fragments.

    Stores price_dynamics / reliability / reviews as sets/lists keyed by their
    natural dedup key so re-inserts are no-ops (mirrors the real ON CONFLICT /
    dedup-guard behaviour the task relies on).
    """

    def __init__(self, pending_rows):
        # list[dict house_id, cian_zhk_url, (optional) ext_id]; default ext_id=None so
        # existing tests that omit it still model "url already present".
        self._pending_rows = [{"ext_id": None, **r} for r in pending_rows]
        # price_dynamics keyed by (house_id, month_date, source, room_count, prices_type, period)
        self.price_dynamics: set[tuple] = set()
        # reviews keyed by (source, ext_review_id)
        self.reviews: dict[tuple, dict] = {}
        # reliability: list of (house_id, source)
        self.reliability: list[tuple] = []
        # houses.cian_zhk_url written by the resolve step: {house_id: (url, nb_id)}
        self.resolved_urls: dict[int, tuple] = {}

    # -- query router ------------------------------------------------------
    def execute(self, statement, params=None):
        sql = str(statement)
        params = params or {}

        if "COUNT(DISTINCT h.id)" in sql and "houses_price_dynamics pd" in sql:
            # _COUNT_PENDING
            return _FakeResult(scalar=len(self._pending_rows))
        if "COUNT(DISTINCT h.id)" in sql and "cian_zhk_url IS NOT NULL" in sql:
            # _COUNT_FETCHABLE
            return _FakeResult(scalar=len(self._pending_rows))
        if "COUNT(DISTINCT h.id)" in sql:
            # _COUNT_TOTAL
            return _FakeResult(scalar=len(self._pending_rows))

        if "SELECT DISTINCT ON (h.id) h.id AS house_id" in sql:
            lim = params.get("lim", len(self._pending_rows))
            # When a row was resolved on a prior pass, surface the persisted url (mirrors
            # the real houses.cian_zhk_url UPDATE so a re-run sees it set).
            out = []
            for r in self._pending_rows[:lim]:
                resolved = self.resolved_urls.get(r["house_id"])
                if resolved is not None and not r.get("cian_zhk_url"):
                    r = {**r, "cian_zhk_url": resolved[0]}
                out.append(r)
            return _FakeResult(rows=out)

        if "UPDATE houses SET" in sql and "cian_zhk_url" in sql and "cian_internal_house_id" in sql:
            self.resolved_urls[params["hid"]] = (params["url"], params["nb_id"])
            return _FakeResult()

        if "COUNT(*) FROM houses_price_dynamics" in sql:
            h = params["h"]
            return _FakeResult(scalar=sum(1 for k in self.price_dynamics if k[0] == h))
        if "COUNT(*) FROM house_reliability_checks" in sql:
            h = params["h"]
            return _FakeResult(scalar=sum(1 for k in self.reliability if k[0] == h))
        if "COUNT(*) FROM house_reviews" in sql:
            h = params["h"]
            return _FakeResult(scalar=sum(1 for k, v in self.reviews.items() if v["house_id"] == h))

        if "INSERT INTO house_reviews" in sql:
            key = (params["source"], params["ext_review_id"])
            self.reviews[key] = {"house_id": params["house_id"], **params}  # UPSERT
            return _FakeResult()

        if "DELETE FROM house_reliability_checks" in sql:
            # dedup guard — keep newest only (model: keep 1 per (house, source))
            h, src = params["h"], params["src"]
            kept = [r for r in self.reliability if not (r[0] == h and r[1] == src)]
            kept.append((h, src))
            self.reliability = kept
            return _FakeResult()

        # Unhandled SQL → inert result (keeps the fake permissive).
        return _FakeResult()

    # -- transaction shims -------------------------------------------------
    def begin_nested(self):
        return MagicMock()

    def commit(self):
        pass

    def rollback(self):
        pass


def _enrichment_with_everything(seed: int = 0) -> NewbuildingEnrichment:
    """Build a full enrichment. `seed` shifts review ids + month so distinct ЖК pages
    produce distinct rows (house_reviews dedups on (source, ext_review_id) globally,
    not per-house — real pages have unique ids; the seed mimics that)."""
    month = f"2025-{(seed % 12) + 1:02d}-01"
    base = 9000 + seed * 10
    return NewbuildingEnrichment(
        cian_internal_house_id=12345 + seed,
        cian_zhk_url=f"https://zhk-test-{seed}-ekb-i.cian.ru/",
        name="ЖК Тест",
        realty_valuation_chart=[
            {
                "month_date": month,
                "room_count": "all",
                "prices_type": "price",
                "period": "halfYear",
                "price_per_sqm": 13800000.0,
            }
        ],
        reliability_checks=[
            {"check_name": "Надёжный застройщик", "check_status": "reliable", "details": []}
        ],
        reviews=[
            {"id": base + 1, "title": "Отличный ЖК", "rate": 5, "text": "Очень доволен"},
            {"id": base + 2, "title": "Норм", "rate": 4, "text": "Есть нюансы"},
        ],
    )


def _fake_save_newbuilding_enrichment(db, house_id, enrichment):
    """Stand-in for the real saver: lands price_dynamics + reliability into FakeDB."""
    for p in enrichment.realty_valuation_chart:
        if p.get("price_per_sqm") is None:
            continue
        db.price_dynamics.add(
            (
                house_id,
                p["month_date"],
                "cian_realty_valuation",
                p.get("room_count", "all"),
                p.get("prices_type", "price"),
                p.get("period", "halfYear"),
            )
        )
    for c in enrichment.reliability_checks:
        if c.get("check_name") or c.get("check_status"):
            db.reliability.append((house_id, "cian_nashdom"))
    db.commit()


# ---------------------------------------------------------------------------
# Pure-unit: _save_cian_reviews mapping (the load-bearing new logic).
# ---------------------------------------------------------------------------


def test_save_cian_reviews_maps_and_dedups() -> None:
    db = FakeDB(pending_rows=[])
    reviews = [
        {"id": 1, "title": "Хорошо", "rate": 5, "text": "ok"},
        {"id": 2, "title": "Плохо", "rate": 2, "text": "bad"},
    ]
    n = _save_cian_reviews(db, house_id=7, reviews=reviews)
    assert n == 2
    assert db.reviews[("cian", 1)]["house_id"] == 7
    assert db.reviews[("cian", 1)]["review_title"] == "Хорошо"

    # Re-run with same ids → UPSERT, no growth (idempotent on (source, ext_review_id)).
    n2 = _save_cian_reviews(db, house_id=7, reviews=reviews)
    assert n2 == 2
    assert len(db.reviews) == 2


def test_save_cian_reviews_skips_entries_without_id() -> None:
    db = FakeDB(pending_rows=[])
    reviews = [{"title": "no id", "text": "x"}, {"id": "not-int", "text": "y"}]
    n = _save_cian_reviews(db, house_id=7, reviews=reviews)
    assert n == 0
    assert len(db.reviews) == 0


def test_save_cian_reviews_clamps_bad_score() -> None:
    db = FakeDB(pending_rows=[])
    _save_cian_reviews(db, house_id=7, reviews=[{"id": 5, "rate": 99}])
    assert db.reviews[("cian", 5)]["score"] is None  # out of 1..5 → dropped


# ---------------------------------------------------------------------------
# Task-level: end-to-end populate + idempotent re-run.
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_backfill_dry_run_counts_only() -> None:
    db = FakeDB(pending_rows=[{"house_id": 1, "cian_zhk_url": "https://zhk-a.cian.ru/"}])
    with patch(
        "scraper_kit.providers.cian.newbuilding.fetch_newbuilding", new_callable=AsyncMock
    ) as mock_fetch:
        result = await backfill_newbuilding_enrichment(db, limit=5, dry_run=True)
    mock_fetch.assert_not_called()
    assert isinstance(result, NewbuildingEnrichBackfillResult)
    assert result.cian_houses_total == 1
    assert result.processed == 0


@pytest.mark.asyncio
async def test_backfill_populates_all_three_tables() -> None:
    db = FakeDB(
        pending_rows=[
            {"house_id": 1, "cian_zhk_url": "https://zhk-a.cian.ru/"},
            {"house_id": 2, "cian_zhk_url": "https://zhk-b.cian.ru/"},
        ]
    )
    # Distinct enrichment per house (real ЖК pages have unique review ids / months).
    by_url = {
        "https://zhk-a.cian.ru/": _enrichment_with_everything(seed=1),
        "https://zhk-b.cian.ru/": _enrichment_with_everything(seed=2),
    }

    async def _fetch(url, **kwargs):
        return by_url[url]

    with (
        patch(
            "scraper_kit.providers.cian.newbuilding.fetch_newbuilding",
            side_effect=_fetch,
        ),
        patch(
            "scraper_kit.providers.cian.newbuilding.save_newbuilding_enrichment",
            side_effect=_fake_save_newbuilding_enrichment,
        ),
    ):
        result = await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)

    assert result.processed == 2
    assert result.succeeded == 2
    assert result.failed_fetch == 0
    assert result.failed_save == 0
    # All three tables non-zero.
    assert len(db.price_dynamics) == 2  # 1 chart point × 2 houses
    assert len(db.reliability) == 2
    assert len(db.reviews) == 4  # 2 reviews × 2 houses
    assert result.price_dynamics_rows == 2
    assert result.reliability_rows == 2
    assert result.review_rows == 4


@pytest.mark.asyncio
async def test_backfill_force_rerun_no_dup() -> None:
    """force=True re-run over the same house must not duplicate any of the 3 tables."""
    rows = [{"house_id": 1, "cian_zhk_url": "https://zhk-a.cian.ru/"}]
    db = FakeDB(pending_rows=rows)
    enr = _enrichment_with_everything()

    with (
        patch(
            "scraper_kit.providers.cian.newbuilding.fetch_newbuilding",
            new_callable=AsyncMock,
            return_value=enr,
        ),
        patch(
            "scraper_kit.providers.cian.newbuilding.save_newbuilding_enrichment",
            side_effect=_fake_save_newbuilding_enrichment,
        ),
    ):
        await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)
        pd_after_first = len(db.price_dynamics)
        rv_after_first = len(db.reviews)

        # Re-run with force=True → re-process the same house.
        result2 = await backfill_newbuilding_enrichment(
            db, limit=5, force=True, request_delay_sec=0
        )

    assert len(db.price_dynamics) == pd_after_first  # ON CONFLICT dedup
    assert len(db.reviews) == rv_after_first  # (source, ext_review_id) dedup
    assert len(db.reliability) == 1  # dedup guard collapsed to newest
    assert result2.processed == 1


@pytest.mark.asyncio
async def test_backfill_default_rerun_skips_and_no_reliability_dup() -> None:
    """force=False resumability: an already-enriched house (pd+reliability present, NO
    reviews — the common Cian case) is skipped on the next run, so a plain re-run neither
    re-fetches nor accumulates duplicate reliability rows.
    """
    rows = [{"house_id": 1, "cian_zhk_url": "https://zhk-a.cian.ru/"}]
    db = FakeDB(pending_rows=rows)
    enr = _enrichment_with_everything()
    enr.reviews = []  # mimic a ЖК whose initialState carries no reviews

    fetch_calls = {"n": 0}

    async def _fetch(url, **kwargs):
        fetch_calls["n"] += 1
        return enr

    with (
        patch("scraper_kit.providers.cian.newbuilding.fetch_newbuilding", side_effect=_fetch),
        patch(
            "scraper_kit.providers.cian.newbuilding.save_newbuilding_enrichment",
            side_effect=_fake_save_newbuilding_enrichment,
        ),
    ):
        # First run enriches.
        r1 = await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)
        # FakeDB._pending_rows is static, so the SELECT still returns the house; the task's
        # per-house guard must skip it now that pd+reliability exist.
        r2 = await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)

    assert r1.succeeded == 1
    assert fetch_calls["n"] == 1  # second run skipped → no extra fetch
    assert r2.skipped_already_enriched == 1
    assert r2.succeeded == 0
    assert len(db.reliability) == 1  # no duplicate appended on re-run


@pytest.mark.asyncio
async def test_backfill_fetch_failure_does_not_abort_batch() -> None:
    """One house returns None (captcha) → counted as fetch-fail, batch continues."""
    db = FakeDB(
        pending_rows=[
            {"house_id": 1, "cian_zhk_url": "https://zhk-a.cian.ru/"},
            {"house_id": 2, "cian_zhk_url": "https://zhk-b.cian.ru/"},
        ]
    )
    enr = _enrichment_with_everything()

    async def _fetch(url, **kwargs):
        return None if "zhk-a" in url else enr

    with (
        patch(
            "scraper_kit.providers.cian.newbuilding.fetch_newbuilding",
            side_effect=_fetch,
        ),
        patch(
            "scraper_kit.providers.cian.newbuilding.save_newbuilding_enrichment",
            side_effect=_fake_save_newbuilding_enrichment,
        ),
    ):
        result = await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)

    assert result.processed == 2
    assert result.failed_fetch == 1
    assert result.succeeded == 1
    assert len(db.reviews) == 2  # only house 2 enriched


# ---------------------------------------------------------------------------
# #972 ЖК-url resolution wiring: ext_id (NULL cian_zhk_url) → resolve → persist → enrich.
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_backfill_resolves_zhk_url_from_ext_id_then_enriches() -> None:
    """A house with ext_id but NULL cian_zhk_url is resolved FIRST (cat.php SERP), the
    url is persisted, and only THEN is the enrichment fetched + saved.
    """
    db = FakeDB(pending_rows=[{"house_id": 1, "cian_zhk_url": None, "ext_id": "48853"}])
    enr = _enrichment_with_everything()

    async def _resolve(nb_id, **kwargs):
        assert nb_id == 48853  # ext_id parsed to int
        return "https://zhk-parkovyy-kvartal-ekb-i.cian.ru"

    fetched_urls: list[str] = []

    async def _fetch(url, **kwargs):
        fetched_urls.append(url)
        return enr

    with (
        patch(
            "scraper_kit.providers.cian.newbuilding.resolve_cian_zhk_url_via_search",
            side_effect=_resolve,
        ),
        patch(
            "scraper_kit.providers.cian.newbuilding.fetch_newbuilding",
            side_effect=_fetch,
        ),
        patch(
            "scraper_kit.providers.cian.newbuilding.save_newbuilding_enrichment",
            side_effect=_fake_save_newbuilding_enrichment,
        ),
    ):
        result = await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)

    # url resolved + persisted to houses (CAST/UPDATE), with the nb_id carried through.
    assert db.resolved_urls[1] == ("https://zhk-parkovyy-kvartal-ekb-i.cian.ru", 48853)
    # enrich fetched the RESOLVED url (not the NULL original).
    assert fetched_urls == ["https://zhk-parkovyy-kvartal-ekb-i.cian.ru"]
    assert result.resolved_zhk_url == 1
    assert result.failed_resolve == 0
    assert result.succeeded == 1
    # enrichment landed.
    assert len(db.price_dynamics) == 1
    assert len(db.reliability) == 1


@pytest.mark.asyncio
async def test_backfill_unresolved_ext_id_counts_fail_and_skips_fetch() -> None:
    """Resolver returns None (404 / empty SERP / block) → failed_resolve, no fetch."""
    db = FakeDB(pending_rows=[{"house_id": 1, "cian_zhk_url": None, "ext_id": "999"}])

    fetch_calls = {"n": 0}

    async def _fetch(url, **kwargs):
        fetch_calls["n"] += 1
        return _enrichment_with_everything()

    with (
        patch(
            "scraper_kit.providers.cian.newbuilding.resolve_cian_zhk_url_via_search",
            new_callable=AsyncMock,
            return_value=None,
        ),
        patch(
            "scraper_kit.providers.cian.newbuilding.fetch_newbuilding",
            side_effect=_fetch,
        ),
        patch(
            "scraper_kit.providers.cian.newbuilding.save_newbuilding_enrichment",
            side_effect=_fake_save_newbuilding_enrichment,
        ),
    ):
        result = await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)

    assert result.failed_resolve == 1
    assert result.resolved_zhk_url == 0
    assert result.succeeded == 0
    assert fetch_calls["n"] == 0  # never fetched without a url
    assert 1 not in db.resolved_urls


@pytest.mark.asyncio
async def test_backfill_resolve_then_idempotent_rerun_no_reresolve() -> None:
    """After a resolve+enrich pass the house is enriched; a plain re-run skips it — no
    second resolve, no second fetch, no duplicate rows.
    """
    db = FakeDB(pending_rows=[{"house_id": 1, "cian_zhk_url": None, "ext_id": "48853"}])
    enr = _enrichment_with_everything()
    enr.reviews = []  # the common Cian case (initialState carries no reviews)

    resolve_calls = {"n": 0}
    fetch_calls = {"n": 0}

    async def _resolve(nb_id, **kwargs):
        resolve_calls["n"] += 1
        return "https://zhk-parkovyy-kvartal-ekb-i.cian.ru"

    async def _fetch(url, **kwargs):
        fetch_calls["n"] += 1
        return enr

    with (
        patch(
            "scraper_kit.providers.cian.newbuilding.resolve_cian_zhk_url_via_search",
            side_effect=_resolve,
        ),
        patch(
            "scraper_kit.providers.cian.newbuilding.fetch_newbuilding",
            side_effect=_fetch,
        ),
        patch(
            "scraper_kit.providers.cian.newbuilding.save_newbuilding_enrichment",
            side_effect=_fake_save_newbuilding_enrichment,
        ),
    ):
        r1 = await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)
        # Second pass: the house now has pd+reliability, so the per-house guard skips it.
        r2 = await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)

    assert r1.resolved_zhk_url == 1
    assert r1.succeeded == 1
    assert resolve_calls["n"] == 1  # not re-resolved on the second pass
    assert fetch_calls["n"] == 1  # not re-fetched on the second pass
    assert r2.skipped_already_enriched == 1
    assert r2.succeeded == 0
    assert len(db.price_dynamics) == 1  # no duplicate
    assert len(db.reliability) == 1


# ---------------------------------------------------------------------------
# #973 (950-E2): nightly schedule — run_newbuilding_enrich wrapper + seed migration 103.
# ---------------------------------------------------------------------------

import re  # noqa: E402
from pathlib import Path  # noqa: E402

from app.tasks import newbuilding_enrich_backfill as task_mod  # noqa: E402

_SQL_DIR = Path(__file__).resolve().parents[2] / "data" / "sql"
_MIGRATION_103 = _SQL_DIR / "103_scrape_schedules_seed_newbuilding_enrich.sql"


# ── run_newbuilding_enrich wrapper: lifecycle + param plumbing ────────────────


@pytest.mark.asyncio
async def test_run_wrapper_marks_done_and_passes_params(monkeypatch: pytest.MonkeyPatch) -> None:
    """run_newbuilding_enrich emits heartbeat → delegates with parsed params → mark_done."""
    seen: dict = {}

    async def _fake_backfill(_db, *, limit, force, request_delay_sec):
        seen.update(limit=limit, force=force, request_delay_sec=request_delay_sec)
        return NewbuildingEnrichBackfillResult(processed=3, succeeded=2, price_dynamics_rows=2)

    monkeypatch.setattr(task_mod, "backfill_newbuilding_enrichment", _fake_backfill)
    monkeypatch.setattr(task_mod.runs_mod, "update_heartbeat", lambda *a, **k: None)
    marked: dict = {}
    monkeypatch.setattr(
        task_mod.runs_mod,
        "mark_done",
        lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
    )
    monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)

    out = await run_newbuilding_enrich(
        object(),  # type: ignore[arg-type]
        run_id=55,
        params={"limit": 7, "force": True, "request_delay_sec": 0},
    )

    assert out.processed == 3
    # default_params plumbed through to the backfill.
    assert seen == {"limit": 7, "force": True, "request_delay_sec": 0.0}
    # run finalised via mark_done with the result counters.
    assert marked["run_id"] == 55
    assert marked["counters"]["processed"] == 3
    assert marked["counters"]["succeeded"] == 2


@pytest.mark.asyncio
async def test_run_wrapper_defaults_when_params_empty(monkeypatch: pytest.MonkeyPatch) -> None:
    """Empty default_params → limit=25, force=False, request_delay_sec=None (→ scraper delay)."""
    seen: dict = {}

    async def _fake_backfill(_db, *, limit, force, request_delay_sec):
        seen.update(limit=limit, force=force, request_delay_sec=request_delay_sec)
        return NewbuildingEnrichBackfillResult()

    monkeypatch.setattr(task_mod, "backfill_newbuilding_enrichment", _fake_backfill)
    monkeypatch.setattr(task_mod.runs_mod, "update_heartbeat", lambda *a, **k: None)
    monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
    monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)

    await run_newbuilding_enrich(object(), run_id=1, params={})  # type: ignore[arg-type]

    assert seen == {"limit": 25, "force": False, "request_delay_sec": None}


@pytest.mark.asyncio
async def test_run_wrapper_marks_failed_and_reraises(monkeypatch: pytest.MonkeyPatch) -> None:
    """Backfill raising → mark_failed + re-raise (no silent swallow)."""

    async def _boom(_db, **kwargs):
        raise RuntimeError("cian exploded")

    monkeypatch.setattr(task_mod, "backfill_newbuilding_enrichment", _boom)
    monkeypatch.setattr(task_mod.runs_mod, "update_heartbeat", lambda *a, **k: None)
    monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
    failed: dict = {}
    monkeypatch.setattr(
        task_mod.runs_mod,
        "mark_failed",
        lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err),
    )

    with pytest.raises(RuntimeError, match="cian exploded"):
        await run_newbuilding_enrich(object(), run_id=9, params={})  # type: ignore[arg-type]

    assert failed["run_id"] == 9
    assert "cian exploded" in failed["err"]


# ── Migration 103 ─────────────────────────────────────────────────────────────


def test_migration_103_exists() -> None:
    assert _MIGRATION_103.is_file(), f"missing migration: {_MIGRATION_103}"


def test_migration_103_seeds_correct_source_and_window() -> None:
    sql = _MIGRATION_103.read_text("utf-8")
    assert "'newbuilding_enrich'" in sql
    insert_block = sql.split("INSERT INTO scrape_schedules")[1]
    # Window 00:00-01:00 UTC = 03:00-04:00 МСК (start=0, end=1).
    # Values may carry a trailing inline comment, so match "<n>," at line start only.
    assert re.search(r"^\s*0,", insert_block, re.MULTILINE), "window_start_hour 0 missing"
    assert re.search(r"^\s*1,", insert_block, re.MULTILINE), "window_end_hour 1 missing"


def test_migration_103_is_idempotent() -> None:
    sql = _MIGRATION_103.read_text("utf-8")
    assert "ON CONFLICT (source) DO NOTHING" in sql


def test_migration_103_is_transactional() -> None:
    sql = _MIGRATION_103.read_text("utf-8")
    assert "BEGIN;" in sql
    assert "COMMIT;" in sql


def test_migration_103_no_psycopg_trap() -> None:
    """Plain seed (no bind params), but guard the :x::type trap anyway (psycopg v3)."""
    sql = _MIGRATION_103.read_text("utf-8")
    assert not re.search(r":\w+::", sql)


def test_migration_103_bootstraps_next_run_at() -> None:
    """next_run_at bootstrapped to tomorrow so the scheduler does not fire on deploy."""
    sql = _MIGRATION_103.read_text("utf-8")
    assert "next_run_at" in sql
    assert "CURRENT_DATE + INTERVAL '1 day'" in sql