gendesign/tradein-mvp/backend/tests/test_city_sweep.py

"""Offline smoke tests for city sweep utilities + endpoints.

Тесты НЕ требуют реального DB — мокируем SessionLocal и scrape_runs модуль.
"""

from __future__ import annotations

import os

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

from datetime import UTC
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

# ── EKB_ANCHORS ────────────────────────────────────────────────────────────


def test_ekb_anchors_count() -> None:
    from scraper_kit.orchestration.pipeline import EKB_ANCHORS

    assert len(EKB_ANCHORS) == 5
    for lat, lon, name in EKB_ANCHORS:
        assert 56.5 < lat < 57.0, f"lat={lat} out of EKB range for anchor {name}"
        assert 60.3 < lon < 60.8, f"lon={lon} out of EKB range for anchor {name}"
        assert isinstance(name, str) and name


# ── resolve_city_name (#2594) ────────────────────────────────────────────────


def test_resolve_city_name_known_oblast_slugs() -> None:
    """Каждый city_slug из CITY_LOCATIONS резолвится в человекочитаемое имя.

    #262 wave 2: CITY_LOCATIONS выросла с 5 (wave 1) до 45 (wave 1 + 40 wave-2
    городов) — вместо хардкода полного списка (дублировал бы CITY_DISPLAY_NAMES и
    ломался при каждом новом городе) проверяем структурный инвариант: CITY_DISPLAY_
    NAMES обязан покрывать РОВНО те же slug'и, что CITY_LOCATIONS (иначе oblast-город
    бы тихо получил ЕКБ-дефолт вместо своего имени) + spot-check wave-1 (не тронуты
    этим PR) и по одному wave-2 city из каждой tier-группы (все три id / только
    cian+yandex / только cian).
    """
    from scraper_kit.orchestration.pipeline import (
        CITY_DISPLAY_NAMES,
        CITY_LOCATIONS,
        resolve_city_name,
    )

    wave1_expected = {
        "nizhniy_tagil": "Нижний Тагил",
        "kamensk_uralskiy": "Каменск-Уральский",
        "pervouralsk": "Первоуральск",
        "verkhnyaya_pyshma": "Верхняя Пышма",
        "serov": "Серов",
    }
    assert set(CITY_DISPLAY_NAMES) == set(CITY_LOCATIONS)
    assert wave1_expected.items() <= CITY_DISPLAY_NAMES.items()
    for slug, name in wave1_expected.items():
        assert resolve_city_name(slug) == name
    # wave-2 spot-check: novouralsk (avito+cian+yandex), revda (cian+yandex, avito
    # НЕ подтверждён), mikhaylovsk (только cian, yandex отсутствует у источника).
    for slug, name in {
        "novouralsk": "Новоуральск",
        "revda": "Ревда",
        "mikhaylovsk": "Михайловск",
    }.items():
        assert slug in CITY_LOCATIONS
        assert resolve_city_name(slug) == name


def test_resolve_city_name_none_defaults_to_ekaterinburg() -> None:
    """city_slug=None — ЕКБ-развёртка той же функции, НЕ «город неизвестен» (#2594 симметрия)."""
    from scraper_kit.orchestration.pipeline import EKATERINBURG_CITY_NAME, resolve_city_name

    assert resolve_city_name(None) == EKATERINBURG_CITY_NAME == "Екатеринбург"


def test_resolve_city_name_unknown_slug_defaults_to_ekaterinburg() -> None:
    """Неизвестный slug — тот же ЕКБ-дефолт, что и get_city_location/get_city_anchors."""
    from scraper_kit.orchestration.pipeline import resolve_city_name

    assert resolve_city_name("nonexistent_city") == "Екатеринбург"


# ── #262: явный fail на известный город БЕЗ подтверждённого provider-id (НЕ силентный
#    ЕКБ-fallback) ───────────────────────────────────────────────────────────────────


async def test_run_avito_city_sweep_raises_on_known_city_without_avito_slug() -> None:
    """revda — известный CITY_LOCATIONS город, но avito_slug=None (не подтверждён).

    run_avito_city_sweep обязан упасть ДО любого сетевого/DB похода — иначе
    `_city_seg()` молча взяла бы 'ekaterinburg' и sweep собрал бы ЕКБ под меткой
    'revda'. Raise происходит в самом начале функции (до await) — MagicMock() для
    config/matcher/enrichment безопасен, до них не доходит."""
    from scraper_kit.orchestration.pipeline import run_avito_city_sweep

    with pytest.raises(ValueError, match="revda"):
        await run_avito_city_sweep(
            MagicMock(),
            run_id=1,
            config=MagicMock(),
            matcher=MagicMock(),
            enrichment=MagicMock(),
            city_slug="revda",
        )


async def test_run_yandex_city_sweep_raises_on_known_city_without_yandex_rgid() -> None:
    """mikhaylovsk — известный CITY_LOCATIONS город, но yandex_rgid=None (город
    отсутствует в гео-базе Яндекса вообще). run_yandex_city_sweep обязан упасть ДО
    любого сетевого/DB похода — иначе YandexRealtyScraper(city_rgid=None) молча
    взял бы ЕКБ rgid."""
    from scraper_kit.orchestration.pipeline import run_yandex_city_sweep

    with pytest.raises(ValueError, match="mikhaylovsk"):
        await run_yandex_city_sweep(
            MagicMock(),
            run_id=1,
            config=MagicMock(),
            matcher=MagicMock(),
            enrichment=MagicMock(),
            city_slug="mikhaylovsk",
        )


async def test_run_avito_city_sweep_does_not_raise_for_ekb_none_city_slug() -> None:
    """city_slug=None (ЕКБ-путь, back-compat) НЕ должен затронуть новый guard — guard
    условие `_loc is not None`, а get_city_location(None) возвращает None."""
    from scraper_kit.orchestration.pipeline import get_city_location

    assert get_city_location(None) is None


# ── CitySweepCounters ───────────────────────────────────────────────────────


def test_city_sweep_counters_defaults() -> None:
    from scraper_kit.orchestration.pipeline import CitySweepCounters

    c = CitySweepCounters()
    assert c.anchors_total == 0
    assert c.lots_fetched == 0
    assert c.errors_count == 0


def test_city_sweep_counters_to_dict() -> None:
    from scraper_kit.orchestration.pipeline import CitySweepCounters

    c = CitySweepCounters(anchors_total=5, anchors_done=2, lots_fetched=150)
    d = c.to_dict()
    assert d["anchors_total"] == 5
    assert d["anchors_done"] == 2
    assert d["lots_fetched"] == 150
    assert "errors_count" in d
    assert "houses_enriched" in d
    assert "detail_enriched" in d


def test_city_sweep_counters_to_dict_all_keys() -> None:
    from dataclasses import fields

    from scraper_kit.orchestration.pipeline import CitySweepCounters

    c = CitySweepCounters()
    d = c.to_dict()
    expected_keys = {f.name for f in fields(c)}
    assert set(d.keys()) == expected_keys


# ── scrape_runs utilities (unit, no DB) ────────────────────────────────────


def test_scrape_runs_create_run_returns_id() -> None:
    from app.services.scrape_runs import create_run

    mock_db = MagicMock()
    mock_row = MagicMock()
    mock_row.id = 42
    mock_db.execute.return_value.fetchone.return_value = mock_row

    run_id = create_run(mock_db, source="avito_city_sweep", params={"pages_per_anchor": 3})

    assert run_id == 42
    mock_db.commit.assert_called()


def test_scrape_runs_is_cancelled_true() -> None:
    from app.services.scrape_runs import is_cancelled

    mock_db = MagicMock()
    mock_row = MagicMock()
    mock_row.status = "cancelled"
    mock_db.execute.return_value.fetchone.return_value = mock_row

    assert is_cancelled(mock_db, 99) is True


def test_scrape_runs_is_cancelled_false_when_running() -> None:
    from app.services.scrape_runs import is_cancelled

    mock_db = MagicMock()
    mock_row = MagicMock()
    mock_row.status = "running"
    mock_db.execute.return_value.fetchone.return_value = mock_row

    assert is_cancelled(mock_db, 99) is False


def test_scrape_runs_is_cancelled_false_when_not_found() -> None:
    from app.services.scrape_runs import is_cancelled

    mock_db = MagicMock()
    mock_db.execute.return_value.fetchone.return_value = None

    assert is_cancelled(mock_db, 999) is False


def test_scrape_runs_mark_cancelled_returns_bool() -> None:
    from app.services.scrape_runs import mark_cancelled

    mock_db = MagicMock()
    # source обязателен: #2674 добавил гейт honors_cancel — отменять можно только то,
    # что отмену опрашивает (иначе 'cancelled' у живой задачи + обход has_running_run).
    mock_db.execute.return_value.fetchone.return_value = MagicMock(source="avito_city_sweep")

    result = mark_cancelled(mock_db, 10)
    assert result is True
    mock_db.commit.assert_called()


def test_scrape_runs_mark_cancelled_returns_false_when_not_running() -> None:
    from app.services.scrape_runs import mark_cancelled

    mock_db = MagicMock()
    mock_db.execute.return_value.fetchone.return_value = None  # no row (not running)

    result = mark_cancelled(mock_db, 10)
    assert result is False


# ── total_seen / new_count column population (audit #1926) ──────────────────


def _captured_params(mock_db: MagicMock) -> dict:
    """Извлечь dict bind-параметров из ПЕРВОГО db.execute(text(...), params).

    Первый вызов — всегда основной UPDATE (mark_done/update_heartbeat). mark_done
    (#2625) может выполнить дополнительные db.execute() ПОСЛЕ него для
    zero-result-алерта (_alert_on_run_id → SELECT source / SELECT streak) —
    call_args_list[0] остаётся стабильным независимо от этого хвоста.
    """
    assert mock_db.execute.call_args_list, "db.execute was not called"
    return mock_db.execute.call_args_list[0].args[1]


def test_column_counts_maps_lots_fetched_inserted() -> None:
    """_column_counts: lots_fetched→total_seen, lots_inserted→new_count."""
    from app.services.scrape_runs import _column_counts

    total_seen, new_count = _column_counts(
        {"lots_fetched": 150, "lots_inserted": 12, "anchors_done": 5}
    )
    assert total_seen == 150
    assert new_count == 12


def test_column_counts_prefers_explicit_total_seen_new_count() -> None:
    """Если в counters уже есть total_seen/new_count — они приоритетнее lots_*."""
    from app.services.scrape_runs import _column_counts

    total_seen, new_count = _column_counts(
        {"total_seen": 99, "new_count": 7, "lots_fetched": 1, "lots_inserted": 1}
    )
    assert total_seen == 99
    assert new_count == 7


def test_column_counts_returns_none_when_absent() -> None:
    """Нет ни одного из ключей → None (COALESCE сохранит старое значение колонки)."""
    from app.services.scrape_runs import _column_counts

    total_seen, new_count = _column_counts({"anchors_done": 3})
    assert total_seen is None
    assert new_count is None


def test_mark_done_persists_total_seen_new_count_columns() -> None:
    """mark_done пишет total_seen/new_count из lots_fetched/lots_inserted (audit #1926)."""
    from app.services.scrape_runs import mark_done

    mock_db = MagicMock()
    mock_db.execute.return_value.first.return_value = MagicMock(id=1)  # row updated

    mark_done(mock_db, 1, {"lots_fetched": 200, "lots_inserted": 18, "anchors_done": 5})

    params = _captured_params(mock_db)
    assert params["total_seen"] == 200
    assert params["new_count"] == 18
    # SQL must SET the dedicated columns, not only the jsonb blob (first call = UPDATE)
    sql = str(mock_db.execute.call_args_list[0].args[0])
    assert "total_seen" in sql
    assert "new_count" in sql
    mock_db.commit.assert_called()


def test_update_heartbeat_persists_total_seen_new_count_columns() -> None:
    """update_heartbeat также обновляет колонки (live progress, не только финал)."""
    from app.services.scrape_runs import update_heartbeat

    mock_db = MagicMock()

    update_heartbeat(mock_db, 1, {"lots_fetched": 75, "lots_inserted": 4})

    params = _captured_params(mock_db)
    assert params["total_seen"] == 75
    assert params["new_count"] == 4
    mock_db.commit.assert_called()


def test_mark_done_yandex_counters_populate_columns() -> None:
    """End-to-end shape: YandexCitySweepCounters.to_dict() → mark_done → columns set."""
    from scraper_kit.orchestration.pipeline import YandexCitySweepCounters

    from app.services.scrape_runs import mark_done

    counters = YandexCitySweepCounters(
        anchors_total=1, anchors_done=1, lots_fetched=540, lots_inserted=42
    )
    mock_db = MagicMock()
    mock_db.execute.return_value.first.return_value = MagicMock(id=1)

    mark_done(mock_db, 1, counters.to_dict())

    params = _captured_params(mock_db)
    assert params["total_seen"] == 540
    assert params["new_count"] == 42


def test_mark_done_avito_cian_counters_populate_columns() -> None:
    """End-to-end shape: CitySweepCounters.to_dict() (avito/cian) → mark_done → columns set.

    CitySweepCounters обслуживает avito + cian city-sweep'ы и проходит тот же путь
    counters.to_dict() → mark_done, что и Yandex — фикс #1926 должен заполнять
    total_seen/new_count и для них (issue: «то же касается CitySweepCounters»).
    """
    from scraper_kit.orchestration.pipeline import CitySweepCounters

    from app.services.scrape_runs import mark_done

    counters = CitySweepCounters(
        anchors_total=4, anchors_done=4, lots_fetched=1234, lots_inserted=56
    )
    mock_db = MagicMock()
    mock_db.execute.return_value.first.return_value = MagicMock(id=1)

    mark_done(mock_db, 1, counters.to_dict())

    params = _captured_params(mock_db)
    assert params["total_seen"] == 1234
    assert params["new_count"] == 56


def test_mark_failed_persists_total_seen_new_count_columns() -> None:
    """Даже при mark_failed колонки заполняются из частично собранных counters (#1926).

    Sweep может упасть на середине, успев собрать тысячи лотов — observability должна
    видеть total_seen>0, а не 0 для failed-прогона.
    """
    from app.services.scrape_runs import mark_failed

    mock_db = MagicMock()
    mock_db.execute.return_value.first.return_value = MagicMock(id=1)

    mark_failed(mock_db, 1, "boom", {"lots_fetched": 300, "lots_inserted": 9})

    # mark_failed после UPDATE дёргает _alert_on_run_id (SELECT source ...), поэтому
    # последний execute — это alert-SELECT, а не наш UPDATE. Ищем UPDATE-вызов явно.
    update_call = next(
        c for c in mock_db.execute.call_args_list if "UPDATE scrape_runs" in str(c.args[0])
    )
    params = update_call.args[1]
    assert params["total_seen"] == 300
    assert params["new_count"] == 9
    sql = str(update_call.args[0])
    assert "total_seen" in sql
    assert "new_count" in sql


# ── FastAPI endpoints (offline) ─────────────────────────────────────────────


@pytest.fixture
def app_with_admin():
    from fastapi import FastAPI
    from fastapi.testclient import TestClient

    # Lazy import to avoid side effects before env is set
    from app.api.v1 import admin as admin_module
    from app.core.db import get_db

    app = FastAPI()
    app.include_router(admin_module.router, prefix="/api/v1/admin")

    def fake_db():
        yield MagicMock()

    app.dependency_overrides[get_db] = fake_db
    return TestClient(app)


def test_start_endpoint_validates_pages_per_anchor(app_with_admin) -> None:
    """pages_per_anchor=99 превышает max=10 → 422."""
    r = app_with_admin.post(
        "/api/v1/admin/scrape/avito-city-sweep",
        json={"pages_per_anchor": 99},
    )
    assert r.status_code == 422


def test_start_endpoint_validates_request_delay_too_low(app_with_admin) -> None:
    """request_delay_sec=1.0 меньше min=3.0 → 422."""
    r = app_with_admin.post(
        "/api/v1/admin/scrape/avito-city-sweep",
        json={"request_delay_sec": 1.0},
    )
    assert r.status_code == 422


def test_start_endpoint_ok(app_with_admin) -> None:
    """Valid request → 200, run_id в ответе; kit sweep вызван с DI (config/matcher/enrichment)."""
    from app.services.scraper_adapters import (
        RealEnrichmentJobs,
        RealMatcherAdapter,
        RealScraperConfig,
    )

    with (
        patch("app.api.v1.admin.has_running_run", return_value=False),
        patch("app.services.scrape_runs.create_run", return_value=7),
        patch("app.api.v1.admin.run_avito_city_sweep", new_callable=AsyncMock) as mock_sweep,
    ):
        r = app_with_admin.post(
            "/api/v1/admin/scrape/avito-city-sweep",
            json={"pages_per_anchor": 2, "detail_top_n": 10},
        )
    assert r.status_code == 200
    body = r.json()
    assert body["run_id"] == 7
    assert body["status"] == "running"
    assert body["pages_per_anchor"] == 2

    mock_sweep.assert_called_once()
    kw = mock_sweep.call_args.kwargs
    assert isinstance(kw["config"], RealScraperConfig)
    assert isinstance(kw["matcher"], RealMatcherAdapter)
    assert isinstance(kw["enrichment"], RealEnrichmentJobs)
    assert "proxy_provider" in kw


def test_cancel_endpoint_exists(app_with_admin) -> None:
    """Cancel endpoint returns run_id + cancelled flag."""
    with patch("app.services.scrape_runs.mark_cancelled", return_value=True):
        r = app_with_admin.post("/api/v1/admin/scrape/avito-city-sweep/123/cancel")
    assert r.status_code == 200
    body = r.json()
    assert body["run_id"] == 123
    assert body["cancelled"] is True
    assert body["ok"] is True


def test_cancel_endpoint_not_running(app_with_admin) -> None:
    """Cancel на уже завершённый run → cancelled=False."""
    with patch("app.services.scrape_runs.mark_cancelled", return_value=False):
        r = app_with_admin.post("/api/v1/admin/scrape/avito-city-sweep/456/cancel")
    assert r.status_code == 200
    assert r.json()["cancelled"] is False


def test_list_runs_endpoint(app_with_admin) -> None:
    """List runs endpoint возвращает список."""
    from datetime import datetime

    fake_rows = [
        {
            "run_id": 1,
            "source": "avito_city_sweep",
            "status": "done",
            "params": {"pages_per_anchor": 3},
            "counters": {"lots_fetched": 300},
            "error": None,
            "started_at": datetime(2025, 1, 1, tzinfo=UTC),
            "finished_at": datetime(2025, 1, 1, 1, tzinfo=UTC),
            "heartbeat_at": datetime(2025, 1, 1, 1, tzinfo=UTC),
        }
    ]
    with patch("app.services.scrape_runs.list_recent", return_value=fake_rows):
        r = app_with_admin.get("/api/v1/admin/scrape/avito-city-sweep/runs")
    assert r.status_code == 200
    body = r.json()
    assert len(body) == 1
    assert body[0]["run_id"] == 1
    assert body[0]["status"] == "done"
    assert "2025-01-01" in body[0]["started_at"]