bot-backend
5373f5db63
All checks were successful
Deploy Trade-In / changes (push) Successful in 5s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / test (push) Successful in 28s
Deploy Trade-In / build-backend (push) Successful in 39s
Deploy Trade-In / deploy (push) Successful in 34s
Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
563 lines
19 KiB
Python
563 lines
19 KiB
Python
"""Unit tests for T7/T8a/T10 backfill services (wave-2).
|
|
|
|
All tests are offline — no DB, no network required.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# ============================================================================
|
|
# T10 — yandex_address_backfill: _extract_address_from_title
|
|
# ============================================================================
|
|
|
|
|
|
class TestYandexTitleExtract:
|
|
"""Tests for _extract_address_from_title helper."""
|
|
|
|
def _extract(self, html: str) -> str | None:
|
|
from app.services.yandex_address_backfill import _extract_address_from_title
|
|
|
|
return _extract_address_from_title(html)
|
|
|
|
def test_extracts_full_address_standard(self):
|
|
html = (
|
|
"<title>Продажа квартиры — Екатеринбург, улица Горького, 36 — id 7654321 "
|
|
"на Яндекс.Недвижимости</title>"
|
|
)
|
|
result = self._extract(html)
|
|
assert result == "Екатеринбург, улица Горького, 36"
|
|
|
|
def test_extracts_address_with_corpus(self):
|
|
html = (
|
|
"<title>Продажа квартиры — Екатеринбург, улица Ленина, 52/1 — id 9999 "
|
|
"на Яндекс.Недвижимости</title>"
|
|
)
|
|
result = self._extract(html)
|
|
assert result == "Екатеринбург, улица Ленина, 52/1"
|
|
|
|
def test_returns_none_when_no_id_marker(self):
|
|
html = "<title>Продажа квартиры — Екатеринбург, улица Ленина, 52/1</title>"
|
|
result = self._extract(html)
|
|
assert result is None
|
|
|
|
def test_returns_none_when_no_title(self):
|
|
html = "<html><body>No title here</body></html>"
|
|
result = self._extract(html)
|
|
assert result is None
|
|
|
|
def test_strips_trailing_punctuation(self):
|
|
html = (
|
|
"<title>Купить квартиру — Екатеринбург, улица Гагарина, 11, — id 111 "
|
|
"на Яндекс.Недвижимости</title>"
|
|
)
|
|
result = self._extract(html)
|
|
# should strip trailing comma
|
|
assert result is not None
|
|
assert not result.endswith(",")
|
|
|
|
def test_nbsp_replaced(self):
|
|
html = (
|
|
"<title>— Екатеринбург,\xa0улица Горького, 36 — id 12 " "на Яндекс.Недвижимости</title>"
|
|
)
|
|
result = self._extract(html)
|
|
assert result is not None
|
|
assert "\xa0" not in result
|
|
|
|
def test_case_insensitive_city(self):
|
|
html = (
|
|
"<title>— ЕКАТЕРИНБУРГ, улица Горького, 36 — id 777 " "на Яндекс.Недвижимости</title>"
|
|
)
|
|
# regex is case-insensitive; city name retains original case
|
|
result = self._extract(html)
|
|
assert result is not None
|
|
assert "Горького, 36" in result
|
|
|
|
|
|
class TestYandexHouseNumberRegex:
|
|
def test_no_house_number_pattern(self):
|
|
from app.services.yandex_address_backfill import _RE_HAS_HOUSE_NUMBER
|
|
|
|
assert not _RE_HAS_HOUSE_NUMBER.search("улица Горького")
|
|
assert not _RE_HAS_HOUSE_NUMBER.search("Екатеринбург, улица Горького")
|
|
|
|
def test_has_house_number_pattern(self):
|
|
from app.services.yandex_address_backfill import _RE_HAS_HOUSE_NUMBER
|
|
|
|
assert _RE_HAS_HOUSE_NUMBER.search("Екатеринбург, улица Горького, 36")
|
|
assert _RE_HAS_HOUSE_NUMBER.search("улица Ленина, 52/1")
|
|
|
|
|
|
# ============================================================================
|
|
# T10 — backfill_yandex_addresses: integration (mocked)
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_yandex_addresses_saves_enriched():
|
|
"""Happy path: fetch returns full address → saved=1."""
|
|
from app.services.yandex_address_backfill import backfill_yandex_addresses
|
|
|
|
html_with_addr = (
|
|
"<title>Продажа — Екатеринбург, улица Горького, 36 — id 11 "
|
|
"на Яндекс.Недвижимости</title>"
|
|
)
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.text = html_with_addr
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.get = AsyncMock(return_value=mock_resp)
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
mock_db = MagicMock()
|
|
mock_db.execute = MagicMock()
|
|
mock_db.commit = MagicMock()
|
|
|
|
items = [
|
|
{
|
|
"id": 42,
|
|
"source_id": "yandex-42",
|
|
"address": "улица Горького",
|
|
"source_url": "https://realty.yandex.ru/offer/11/",
|
|
}
|
|
]
|
|
|
|
with (
|
|
patch(
|
|
"app.services.yandex_address_backfill._eligible_listings",
|
|
return_value=items,
|
|
),
|
|
patch(
|
|
"app.services.yandex_address_backfill.AsyncSession",
|
|
return_value=mock_session,
|
|
),
|
|
):
|
|
result = await backfill_yandex_addresses(mock_db, limit=10)
|
|
|
|
assert result.checked == 1
|
|
assert result.saved == 1
|
|
assert result.errors == 0
|
|
assert result.skipped == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_yandex_addresses_skips_no_house_number():
|
|
"""Address extracted but still has no house number → skipped."""
|
|
from app.services.yandex_address_backfill import backfill_yandex_addresses
|
|
|
|
# Title without house number in extracted part
|
|
html = "<title>— Екатеринбург, улица Горького — id 22 на Яндекс.Недвижимости</title>"
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.text = html
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.get = AsyncMock(return_value=mock_resp)
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
items = [
|
|
{
|
|
"id": 99,
|
|
"source_id": "y-99",
|
|
"address": "улица Горького",
|
|
"source_url": "https://realty.yandex.ru/offer/22/",
|
|
}
|
|
]
|
|
|
|
with (
|
|
patch(
|
|
"app.services.yandex_address_backfill._eligible_listings",
|
|
return_value=items,
|
|
),
|
|
patch(
|
|
"app.services.yandex_address_backfill.AsyncSession",
|
|
return_value=mock_session,
|
|
),
|
|
):
|
|
result = await backfill_yandex_addresses(MagicMock(), limit=10)
|
|
|
|
assert result.skipped == 1
|
|
assert result.saved == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_yandex_addresses_error_on_non_200():
|
|
"""Non-200 response → errors counter incremented, no crash."""
|
|
from app.services.yandex_address_backfill import backfill_yandex_addresses
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 403
|
|
mock_resp.text = ""
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.get = AsyncMock(return_value=mock_resp)
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
items = [
|
|
{
|
|
"id": 55,
|
|
"source_id": "y-55",
|
|
"address": "улица Горького",
|
|
"source_url": "https://realty.yandex.ru/offer/55/",
|
|
}
|
|
]
|
|
|
|
with (
|
|
patch(
|
|
"app.services.yandex_address_backfill._eligible_listings",
|
|
return_value=items,
|
|
),
|
|
patch(
|
|
"app.services.yandex_address_backfill.AsyncSession",
|
|
return_value=mock_session,
|
|
),
|
|
):
|
|
result = await backfill_yandex_addresses(MagicMock(), limit=10)
|
|
|
|
assert result.errors == 1
|
|
assert result.saved == 0
|
|
|
|
|
|
# ============================================================================
|
|
# T8a — cian_price_history: backfill_cian_price_history
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cian_price_history_saves_changes():
|
|
"""Happy path: fetch returns DetailEnrichment with price_changes → saved."""
|
|
from app.services.cian_price_history import backfill_cian_price_history
|
|
from app.services.scrapers.cian_detail import DetailEnrichment
|
|
|
|
enrichment = DetailEnrichment(
|
|
cian_id=123456,
|
|
price_changes=[
|
|
{"change_time": "2026-04-01T00:00:00Z", "price_rub": 5000000, "diff_percent": None},
|
|
{"change_time": "2026-04-15T00:00:00Z", "price_rub": 4900000, "diff_percent": -2.0},
|
|
],
|
|
)
|
|
|
|
mock_db = MagicMock()
|
|
# Simulate listings rows returned by SELECT
|
|
rows = [{"id": 10, "source_url": "https://ekb.cian.ru/sale/flat/123456/"}]
|
|
|
|
with (
|
|
patch(
|
|
"app.services.cian_price_history.fetch_detail",
|
|
new_callable=AsyncMock,
|
|
return_value=enrichment,
|
|
) as mock_fetch,
|
|
patch("app.services.cian_price_history.save_detail_enrichment") as mock_save,
|
|
patch("app.services.cian_price_history.get_scraper_delay", return_value=0.0),
|
|
):
|
|
mock_mappings = MagicMock()
|
|
mock_mappings.all.return_value = rows
|
|
mock_db.execute.return_value.mappings.return_value = mock_mappings
|
|
|
|
result = await backfill_cian_price_history(mock_db, batch_size=10)
|
|
|
|
assert result.checked == 1
|
|
assert result.saved == 2 # 2 price_changes
|
|
assert result.errors == 0
|
|
mock_fetch.assert_called_once()
|
|
mock_save.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cian_price_history_skips_no_changes():
|
|
"""Fetch succeeds but no price_changes → skipped (not error)."""
|
|
from app.services.cian_price_history import backfill_cian_price_history
|
|
from app.services.scrapers.cian_detail import DetailEnrichment
|
|
|
|
enrichment = DetailEnrichment(cian_id=111, price_changes=[])
|
|
|
|
mock_db = MagicMock()
|
|
rows = [{"id": 20, "source_url": "https://ekb.cian.ru/sale/flat/111/"}]
|
|
|
|
with (
|
|
patch(
|
|
"app.services.cian_price_history.fetch_detail",
|
|
new_callable=AsyncMock,
|
|
return_value=enrichment,
|
|
),
|
|
patch("app.services.cian_price_history.save_detail_enrichment") as mock_save,
|
|
patch("app.services.cian_price_history.get_scraper_delay", return_value=0.0),
|
|
):
|
|
mock_mappings = MagicMock()
|
|
mock_mappings.all.return_value = rows
|
|
mock_db.execute.return_value.mappings.return_value = mock_mappings
|
|
|
|
result = await backfill_cian_price_history(mock_db, batch_size=10)
|
|
|
|
assert result.skipped == 1
|
|
assert result.saved == 0
|
|
mock_save.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cian_price_history_handles_fetch_none():
|
|
"""fetch_detail returns None → errors+1, no crash."""
|
|
from app.services.cian_price_history import backfill_cian_price_history
|
|
|
|
mock_db = MagicMock()
|
|
rows = [{"id": 30, "source_url": "https://ekb.cian.ru/sale/flat/999/"}]
|
|
|
|
with (
|
|
patch(
|
|
"app.services.cian_price_history.fetch_detail",
|
|
new_callable=AsyncMock,
|
|
return_value=None,
|
|
),
|
|
patch("app.services.cian_price_history.get_scraper_delay", return_value=0.0),
|
|
):
|
|
mock_mappings = MagicMock()
|
|
mock_mappings.all.return_value = rows
|
|
mock_db.execute.return_value.mappings.return_value = mock_mappings
|
|
|
|
result = await backfill_cian_price_history(mock_db, batch_size=10)
|
|
|
|
assert result.errors == 1
|
|
assert result.saved == 0
|
|
|
|
|
|
# ============================================================================
|
|
# T7 — house_imv_backfill: helpers
|
|
# ============================================================================
|
|
|
|
|
|
class TestHouseTypeMap:
|
|
def test_known_types(self):
|
|
from app.services.house_imv_backfill import _map_house_type
|
|
|
|
assert _map_house_type("panel") == "panel"
|
|
assert _map_house_type("brick") == "brick"
|
|
assert _map_house_type("monolith") == "monolithic"
|
|
assert _map_house_type("monolith_brick") == "monolithic"
|
|
assert _map_house_type("monolithic") == "monolithic"
|
|
assert _map_house_type("block") == "block"
|
|
assert _map_house_type("wood") == "wood"
|
|
|
|
def test_unknown_falls_back_to_panel(self):
|
|
from app.services.house_imv_backfill import _map_house_type
|
|
|
|
assert _map_house_type("unknown_type") == "panel"
|
|
assert _map_house_type(None) == "panel"
|
|
assert _map_house_type("") == "panel"
|
|
|
|
def test_case_insensitive(self):
|
|
from app.services.house_imv_backfill import _map_house_type
|
|
|
|
assert _map_house_type("PANEL") == "panel"
|
|
assert _map_house_type("Brick") == "brick"
|
|
assert _map_house_type("MONOLITH_BRICK") == "monolithic"
|
|
|
|
|
|
class TestRegionPrefix:
|
|
def test_ekb_coords(self):
|
|
from app.services.house_imv_backfill import _detect_region_prefix
|
|
|
|
prefix = _detect_region_prefix(56.8296, 60.6001)
|
|
assert prefix == "Свердловская область, Екатеринбург"
|
|
|
|
def test_none_coords_default(self):
|
|
from app.services.house_imv_backfill import _detect_region_prefix
|
|
|
|
prefix = _detect_region_prefix(None, None)
|
|
assert prefix == "Свердловская область, Екатеринбург"
|
|
|
|
def test_out_of_any_region_default(self):
|
|
from app.services.house_imv_backfill import _detect_region_prefix
|
|
|
|
# Moscow coords — not in any bbox → default EKB
|
|
prefix = _detect_region_prefix(55.75, 37.6)
|
|
assert prefix == "Свердловская область, Екатеринбург"
|
|
|
|
|
|
class TestEnrichAddress:
|
|
def test_prepends_region_when_no_marker(self):
|
|
from app.services.house_imv_backfill import _enrich_address_for_imv
|
|
|
|
result = _enrich_address_for_imv("улица Горького, 36", 56.83, 60.60)
|
|
assert result.startswith("Свердловская область, Екатеринбург,")
|
|
assert "улица Горького, 36" in result
|
|
|
|
def test_does_not_prepend_when_region_already_present(self):
|
|
from app.services.house_imv_backfill import _enrich_address_for_imv
|
|
|
|
addr = "Свердловская область, Екатеринбург, улица Горького, 36"
|
|
result = _enrich_address_for_imv(addr, 56.83, 60.60)
|
|
assert result == addr
|
|
|
|
def test_does_not_prepend_when_kraj_present(self):
|
|
from app.services.house_imv_backfill import _enrich_address_for_imv
|
|
|
|
addr = "Ставропольский край, Ставрополь, улица Ленина, 1"
|
|
result = _enrich_address_for_imv(addr, 45.05, 41.97)
|
|
assert result == addr
|
|
|
|
|
|
# ============================================================================
|
|
# T7 — backfill_house_imv: integration (mocked)
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_house_imv_ok_path():
|
|
"""Happy path: picks params, calls IMV, saves → saved=1."""
|
|
from app.services.house_imv_backfill import backfill_house_imv
|
|
from app.services.scrapers.avito_imv import IMVEvaluation, IMVGeo
|
|
|
|
fake_geo = IMVGeo(geo_hash="tok", lat=56.83, lon=60.60)
|
|
fake_eval = IMVEvaluation(
|
|
cache_key="abc123",
|
|
address="ЕКБ ул. Горького, 36",
|
|
rooms=2,
|
|
area_m2=52.0,
|
|
floor=3,
|
|
floor_at_home=10,
|
|
house_type="panel",
|
|
renovation_type="cosmetic",
|
|
has_balcony=True,
|
|
has_loggia=False,
|
|
geo=fake_geo,
|
|
recommended_price=5_000_000,
|
|
lower_price=4_500_000,
|
|
higher_price=5_500_000,
|
|
market_count=24,
|
|
placement_history=[],
|
|
suggestions=[],
|
|
)
|
|
|
|
fake_params = {
|
|
"rooms": 2,
|
|
"area_m2": 52.0,
|
|
"floor": 3,
|
|
"floor_at_home": 10,
|
|
"house_type": "panel",
|
|
"renovation_type": "cosmetic",
|
|
"has_balcony": True,
|
|
"has_loggia": False,
|
|
}
|
|
|
|
mock_db = MagicMock()
|
|
mock_db.commit = MagicMock()
|
|
|
|
houses = [
|
|
{"id": 7, "address": "улица Горького, 36", "full_address": None, "lat": 56.83, "lon": 60.60}
|
|
]
|
|
|
|
with (
|
|
patch(
|
|
"app.services.house_imv_backfill.pick_lot_params",
|
|
return_value=fake_params,
|
|
),
|
|
patch(
|
|
"app.services.house_imv_backfill.evaluate_via_imv",
|
|
new_callable=AsyncMock,
|
|
return_value=fake_eval,
|
|
),
|
|
patch("app.services.house_imv_backfill.save_imv_result") as mock_save,
|
|
):
|
|
mock_mappings = MagicMock()
|
|
mock_mappings.all.return_value = houses
|
|
mock_db.execute.return_value.mappings.return_value = mock_mappings
|
|
|
|
result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)
|
|
|
|
assert result.checked == 1
|
|
assert result.saved == 1
|
|
assert result.errors == 0
|
|
assert result.status_counts.get("ok") == 1
|
|
mock_save.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_house_imv_no_params():
|
|
"""pick_lot_params returns {} → skipped (no_params status)."""
|
|
from app.services.house_imv_backfill import backfill_house_imv
|
|
|
|
mock_db = MagicMock()
|
|
mock_db.execute = MagicMock()
|
|
mock_db.commit = MagicMock()
|
|
|
|
houses = [
|
|
{"id": 8, "address": "улица Ленина, 1", "full_address": None, "lat": 56.83, "lon": 60.60}
|
|
]
|
|
|
|
with (
|
|
patch(
|
|
"app.services.house_imv_backfill.pick_lot_params",
|
|
return_value={},
|
|
),
|
|
patch("app.services.house_imv_backfill._mark_status") as mock_mark,
|
|
):
|
|
mock_mappings = MagicMock()
|
|
mock_mappings.all.return_value = houses
|
|
mock_db.execute.return_value.mappings.return_value = mock_mappings
|
|
|
|
result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)
|
|
|
|
assert result.skipped == 1
|
|
assert result.saved == 0
|
|
mock_mark.assert_called_once_with(mock_db, 8, "no_params", "no listings with rooms+area")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_house_imv_not_found():
|
|
"""IMVAddressNotFoundError → status not_found, no crash."""
|
|
from app.services.house_imv_backfill import backfill_house_imv
|
|
from app.services.scrapers.avito_imv import IMVAddressNotFoundError
|
|
|
|
fake_params = {
|
|
"rooms": 2,
|
|
"area_m2": 52.0,
|
|
"floor": 3,
|
|
"floor_at_home": 10,
|
|
"house_type": "panel",
|
|
"renovation_type": "cosmetic",
|
|
"has_balcony": True,
|
|
"has_loggia": False,
|
|
}
|
|
|
|
mock_db = MagicMock()
|
|
mock_db.commit = MagicMock()
|
|
|
|
houses = [
|
|
{
|
|
"id": 9,
|
|
"address": "несуществующий адрес",
|
|
"full_address": None,
|
|
"lat": 56.83,
|
|
"lon": 60.60,
|
|
}
|
|
]
|
|
|
|
with (
|
|
patch(
|
|
"app.services.house_imv_backfill.pick_lot_params",
|
|
return_value=fake_params,
|
|
),
|
|
patch(
|
|
"app.services.house_imv_backfill.evaluate_via_imv",
|
|
new_callable=AsyncMock,
|
|
side_effect=IMVAddressNotFoundError("not found"),
|
|
),
|
|
patch("app.services.house_imv_backfill._mark_status") as mock_mark,
|
|
):
|
|
mock_mappings = MagicMock()
|
|
mock_mappings.all.return_value = houses
|
|
mock_db.execute.return_value.mappings.return_value = mock_mappings
|
|
|
|
result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)
|
|
|
|
assert result.errors == 1
|
|
assert result.saved == 0
|
|
mock_mark.assert_called_once_with(mock_db, 9, "not_found", "not found")
|