bot-backend
da28a52067
Some checks failed
Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
288 lines
11 KiB
Python
288 lines
11 KiB
Python
"""T6 — avito citywide + byrooms sweep mode.
|
||
|
||
Тесты:
|
||
1. _build_citywide_url — без geoCoords/radius, sort=104, page param.
|
||
2. _build_rooms_url — room-slug в пути, sort=104, page param.
|
||
3. ROOM_SLUGS — 7 категорий, структура кортежей.
|
||
4. fetch_city_wide break-on-empty — останавливается когда _parse_html вернул [].
|
||
5. fetch_city_wide dedup — дублирующийся source_id не дублируется в результате.
|
||
6. fetch_by_rooms break-on-empty per-category — каждая категория прерывается.
|
||
7. fetch_by_rooms global dedup — cross-category дедуп по source_id.
|
||
8. fetch_around не затронут — сигнатура и URL-режим не изменились.
|
||
|
||
Без сети, без БД, без curl_cffi.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from unittest.mock import AsyncMock, patch
|
||
|
||
import pytest
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
|
||
|
||
from app.services.scrapers.avito import ROOM_SLUGS, AvitoScraper
|
||
from app.services.scrapers.base import ScrapedLot
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _make_lot(source_id: str) -> ScrapedLot:
|
||
return ScrapedLot(
|
||
source="avito",
|
||
source_url=f"https://www.avito.ru/ekaterinburg/kvartiry/{source_id}",
|
||
source_id=source_id,
|
||
price_rub=5_000_000,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 1. _build_citywide_url — без geo, s=104
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_build_citywide_url_no_geocoords() -> None:
|
||
s = AvitoScraper()
|
||
url = s._build_citywide_url(page=1)
|
||
assert "geoCoords" not in url
|
||
assert "radius" not in url
|
||
assert "s=104" in url
|
||
assert "p=1" in url
|
||
assert "ekaterinburg/kvartiry/prodam-ASgBAgICAUSSA8YQ" in url
|
||
|
||
|
||
def test_build_citywide_url_page_param() -> None:
|
||
s = AvitoScraper()
|
||
assert "p=3" in s._build_citywide_url(page=3)
|
||
assert "p=10" in s._build_citywide_url(page=10)
|
||
|
||
|
||
def test_build_citywide_url_default_page() -> None:
|
||
s = AvitoScraper()
|
||
assert "p=1" in s._build_citywide_url()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 2. _build_rooms_url — slug в пути, без geo
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_build_rooms_url_contains_slug() -> None:
|
||
s = AvitoScraper()
|
||
slug = "1-komnatnye-ASgBAgICAkSSA8YQygiAWQ"
|
||
url = s._build_rooms_url(slug, page=1)
|
||
assert slug in url
|
||
assert "geoCoords" not in url
|
||
assert "radius" not in url
|
||
assert "s=104" in url
|
||
assert "p=1" in url
|
||
assert "ekaterinburg/kvartiry/prodam/" in url
|
||
|
||
|
||
def test_build_rooms_url_page_param() -> None:
|
||
s = AvitoScraper()
|
||
slug = "studii-ASgBAgICAkSSA8YQygj~WA"
|
||
assert "p=5" in s._build_rooms_url(slug, page=5)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 3. ROOM_SLUGS structure
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_room_slugs_count() -> None:
|
||
assert len(ROOM_SLUGS) == 7
|
||
|
||
|
||
def test_room_slugs_are_pairs() -> None:
|
||
for name, slug in ROOM_SLUGS:
|
||
assert isinstance(name, str) and name
|
||
assert isinstance(slug, str) and slug
|
||
# Каждый slug содержит ASgB-кодировку
|
||
assert "ASgB" in slug
|
||
|
||
|
||
def test_room_slugs_names_unique() -> None:
|
||
names = [name for name, _ in ROOM_SLUGS]
|
||
assert len(names) == len(set(names))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 4. fetch_city_wide — break-on-empty
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_fetch_city_wide_break_on_empty() -> None:
|
||
"""Если _parse_html вернул [] — цикл прерывается, не идёт на следующие страницы."""
|
||
s = AvitoScraper()
|
||
# page=1 возвращает 2 лота, page=2 возвращает [] → стоп
|
||
page_lots: dict[int, list[ScrapedLot]] = {
|
||
1: [_make_lot("AAA"), _make_lot("BBB")],
|
||
2: [],
|
||
}
|
||
call_count = 0
|
||
|
||
async def mock_fetch_html(url: str, page: int) -> str:
|
||
return f"<html>page{page}</html>"
|
||
|
||
def mock_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
|
||
nonlocal call_count
|
||
call_count += 1
|
||
# Derive page number from html string
|
||
for p, lots in page_lots.items():
|
||
if f"page{p}" in html:
|
||
return lots
|
||
return []
|
||
|
||
with patch.object(s, "_fetch_serp_html", side_effect=mock_fetch_html):
|
||
with patch.object(s, "_parse_html", side_effect=mock_parse):
|
||
with patch.object(s, "sleep_between_requests", new=AsyncMock()):
|
||
result = await s.fetch_city_wide(pages=10, delay_override_sec=0)
|
||
|
||
assert len(result) == 2
|
||
assert call_count == 2 # page1 + page2(empty) → stop; page3..10 не вызваны
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 5. fetch_city_wide — dedup по source_id
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_fetch_city_wide_dedup_by_source_id() -> None:
|
||
"""Лот с повторяющимся source_id добавляется только один раз."""
|
||
s = AvitoScraper()
|
||
dup_lot = _make_lot("SAME_ID")
|
||
# pages 1 и 2 возвращают один и тот же lot
|
||
pages_data: list[list[ScrapedLot]] = [[dup_lot], [dup_lot], []]
|
||
|
||
async def mock_html(url: str, page: int) -> str:
|
||
return f"<html>{page}</html>"
|
||
|
||
call_n = 0
|
||
|
||
def mock_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
|
||
nonlocal call_n
|
||
idx = call_n
|
||
call_n += 1
|
||
return pages_data[idx] if idx < len(pages_data) else []
|
||
|
||
with patch.object(s, "_fetch_serp_html", side_effect=mock_html):
|
||
with patch.object(s, "_parse_html", side_effect=mock_parse):
|
||
with patch.object(s, "sleep_between_requests", new=AsyncMock()):
|
||
result = await s.fetch_city_wide(pages=10, delay_override_sec=0)
|
||
|
||
assert len(result) == 1
|
||
assert result[0].source_id == "SAME_ID"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 6. fetch_by_rooms — break-on-empty per category
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_fetch_by_rooms_break_on_empty_per_category() -> None:
|
||
"""Каждая категория прерывается на пустой странице, следующая категория продолжается."""
|
||
s = AvitoScraper()
|
||
slugs = [("1-комн.", "slug-1k"), ("2-комн.", "slug-2k")]
|
||
|
||
# category 1: page1=2 lots, page2=[] → стоп на этой категории
|
||
# category 2: page1=3 lots, page2=[] → стоп
|
||
parse_responses: list[list[ScrapedLot]] = [
|
||
[_make_lot("C1P1a"), _make_lot("C1P1b")], # cat1 p1
|
||
[], # cat1 p2 → break
|
||
[_make_lot("C2P1a"), _make_lot("C2P1b"), _make_lot("C2P1c")], # cat2 p1
|
||
[], # cat2 p2 → break
|
||
]
|
||
parse_call = 0
|
||
|
||
async def mock_html(url: str, page: int) -> str:
|
||
return f"<html>{url}{page}</html>"
|
||
|
||
def mock_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
|
||
nonlocal parse_call
|
||
idx = parse_call
|
||
parse_call += 1
|
||
return parse_responses[idx] if idx < len(parse_responses) else []
|
||
|
||
with patch.object(s, "_fetch_serp_html", side_effect=mock_html):
|
||
with patch.object(s, "_parse_html", side_effect=mock_parse):
|
||
with patch.object(s, "sleep_between_requests", new=AsyncMock()):
|
||
result = await s.fetch_by_rooms(pages=10, delay_override_sec=0, room_slugs=slugs)
|
||
|
||
assert len(result) == 5 # 2 + 3 (нет дублей)
|
||
assert parse_call == 4 # 2 pages per category × 2 categories
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 7. fetch_by_rooms — global dedup cross-category
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_fetch_by_rooms_global_dedup() -> None:
|
||
"""Лот присутствующий в двух категориях добавляется только один раз."""
|
||
s = AvitoScraper()
|
||
slugs = [("A", "slug-a"), ("B", "slug-b")]
|
||
|
||
shared = _make_lot("SHARED")
|
||
unique_a = _make_lot("UNIQUE_A")
|
||
unique_b = _make_lot("UNIQUE_B")
|
||
|
||
parse_data: list[list[ScrapedLot]] = [
|
||
[shared, unique_a], # cat A p1
|
||
[], # cat A p2 → break
|
||
[shared, unique_b], # cat B p1 (shared — дубль)
|
||
[], # cat B p2 → break
|
||
]
|
||
call_idx = 0
|
||
|
||
async def mock_html(url: str, page: int) -> str:
|
||
return "<html/>"
|
||
|
||
def mock_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
|
||
nonlocal call_idx
|
||
i = call_idx
|
||
call_idx += 1
|
||
return parse_data[i] if i < len(parse_data) else []
|
||
|
||
with patch.object(s, "_fetch_serp_html", side_effect=mock_html):
|
||
with patch.object(s, "_parse_html", side_effect=mock_parse):
|
||
with patch.object(s, "sleep_between_requests", new=AsyncMock()):
|
||
result = await s.fetch_by_rooms(pages=10, delay_override_sec=0, room_slugs=slugs)
|
||
|
||
ids = [lot.source_id for lot in result]
|
||
assert ids.count("SHARED") == 1
|
||
assert "UNIQUE_A" in ids
|
||
assert "UNIQUE_B" in ids
|
||
assert len(result) == 3
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 8. fetch_around не затронут — URL содержит geoCoords
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_build_web_url_still_has_geocoords() -> None:
|
||
"""Существующий _build_web_url сохраняет geoCoords+radius — geo-режим не сломан."""
|
||
s = AvitoScraper()
|
||
url = s._build_web_url(56.838, 60.605, radius_km=2, page=1)
|
||
assert "geoCoords=56.838%2C60.605" in url or "geoCoords=56.838,60.605" in url
|
||
assert "radius=2" in url
|
||
assert "s=104" in url
|
||
assert "p=1" in url
|
||
|
||
|
||
def test_citywide_url_differs_from_geo_url() -> None:
|
||
"""citywide URL не содержит geoCoords — это принципиально разные URL."""
|
||
s = AvitoScraper()
|
||
geo_url = s._build_web_url(56.838, 60.605, radius_km=1, page=1)
|
||
cw_url = s._build_citywide_url(page=1)
|
||
assert geo_url != cw_url
|
||
assert "geoCoords" not in cw_url
|
||
assert "geoCoords" in geo_url
|