lekss361
7daeaa4b59
All checks were successful
Deploy Trade-In / changes (push) Successful in 13s
Deploy Trade-In / test (push) Has been skipped
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-backend (push) Has been skipped
Deploy Trade-In / build-browser (push) Successful in 2m51s
Deploy Trade-In / deploy (push) Successful in 1m35s
Deploy Trade-In / deploy-status (push) Successful in 1s
Deploy Trade-In / perimeter-smoke (push) Successful in 11s
Co-authored-by: lekss361 <lekss361@gendsgn.local> Co-committed-by: lekss361 <lekss361@gendsgn.local>
349 lines
14 KiB
Python
349 lines
14 KiB
Python
"""test_server_behaviour.py — имитация чтения страницы человеком (#3283).
|
||
|
||
По умолчанию (body["behaviour"] отсутствует/пуст) поведение /fetch байт-в-байт то
|
||
же, что и до #3283 — вклад имитации в проходимость анти-бота пока НЕ доказан, и
|
||
прод должен уметь честно померить "с behaviour vs без". Проверяет:
|
||
|
||
1. behaviour отсутствует → ни page.mouse.wheel, ни доп. dwell-пауза, ни побочный
|
||
заход на рекомендательную ссылку НЕ вызываются;
|
||
2. behaviour.scroll=True → 3-6 вызовов page.mouse.wheel(0, 500-1400px);
|
||
3. behaviour.dwell_ms — валидный [min, max] отрабатывает; невалидный (не пара
|
||
int, min>max, отрицательные) — /fetch отвечает 400, как и невалидный
|
||
fetch_mode;
|
||
4. behaviour.hop_recommended — заходит на ОДНУ ссылку ТОГО ЖЕ хоста, отказ на
|
||
этом заходе не роняет основной ответ; ссылка с чужого хоста не берётся;
|
||
работает только при reuse_context=True.
|
||
5. body["behaviour"] не-dict → 400.
|
||
|
||
camoufox НЕ запускается — browser/context/page подделаны, в стиле
|
||
test_server_reuse_context.py.
|
||
|
||
Запуск (из tradein-mvp/browser/)::
|
||
|
||
python -m pytest test_server_behaviour.py -q
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import importlib.util
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from aiohttp.test_utils import make_mocked_request
|
||
|
||
_SERVER_PATH = Path(__file__).resolve().parent / "server.py"
|
||
_spec = importlib.util.spec_from_file_location("tradein_browser_server", _SERVER_PATH)
|
||
assert _spec is not None and _spec.loader is not None
|
||
server = importlib.util.module_from_spec(_spec)
|
||
_spec.loader.exec_module(server)
|
||
|
||
|
||
def _json_body(response: Any) -> dict[str, Any]:
|
||
return json.loads(response.body.decode())
|
||
|
||
|
||
async def _coro(value: Any) -> Any:
|
||
return value
|
||
|
||
|
||
def _make_fetch_request(body: dict[str, Any]) -> Any:
|
||
request = make_mocked_request("POST", "/fetch")
|
||
request.json = lambda: _coro(body) # type: ignore[method-assign]
|
||
return request
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setattr(server, "_browsers", {})
|
||
monkeypatch.setattr(server, "_browser_cms", {})
|
||
monkeypatch.setattr(server, "_contexts", {})
|
||
monkeypatch.setattr(server, "_anchor_pages", {})
|
||
monkeypatch.setattr(server, "_page_counters", {})
|
||
monkeypatch.setattr(server, "_locks", {})
|
||
monkeypatch.setattr(server, "_last_goto_at", {})
|
||
monkeypatch.setattr(
|
||
server, "_RECYCLE_PAGES_BY_PROVIDER", {p: 10_000 for p in server.PROVIDERS}
|
||
)
|
||
# Джиттер пейсинга (#3283) не имеет отношения к behaviour — гасим, чтобы не
|
||
# добавлять произвольный real-time sleep в тест.
|
||
monkeypatch.setattr(server, "_MIN_PAGE_INTERVAL_BY_PROVIDER", {p: 0.0 for p in server.PROVIDERS})
|
||
|
||
|
||
_DEFAULT_HTML = "<html>ok<script>window.__SSR_STATE__={}</script></html>"
|
||
|
||
|
||
class _FakeMouse:
|
||
def __init__(self) -> None:
|
||
self.wheel_calls: list[tuple[int, int]] = []
|
||
|
||
async def wheel(self, dx: int, dy: int) -> None:
|
||
self.wheel_calls.append((dx, dy))
|
||
|
||
|
||
class _FakeLocator:
|
||
def __init__(self, hrefs: list[str]) -> None:
|
||
self._hrefs = hrefs
|
||
|
||
async def evaluate_all(self, _js: str) -> list[str]:
|
||
return self._hrefs
|
||
|
||
|
||
class _FakePage:
|
||
"""Поддельная page: goto/content/mouse/locator/context, минимальный контракт _fetch_once."""
|
||
|
||
def __init__(self, context: "_FakeContext", links: list[str] | None = None) -> None:
|
||
self.context = context
|
||
self.mouse = _FakeMouse()
|
||
self.goto_urls: list[str] = []
|
||
self.wait_for_timeout_calls: list[int] = []
|
||
self.closed = 0
|
||
self._links = links or []
|
||
|
||
def is_closed(self) -> bool:
|
||
return self.closed > 0
|
||
|
||
async def route(self, pattern: str, handler: Any) -> None:
|
||
return None
|
||
|
||
async def goto(self, url: str, **kwargs: Any) -> None:
|
||
self.goto_urls.append(url)
|
||
|
||
async def wait_for_timeout(self, ms: int) -> None:
|
||
self.wait_for_timeout_calls.append(ms)
|
||
|
||
async def content(self) -> str:
|
||
return _DEFAULT_HTML
|
||
|
||
def locator(self, selector: str) -> _FakeLocator:
|
||
return _FakeLocator(self._links)
|
||
|
||
async def close(self) -> None:
|
||
self.closed += 1
|
||
|
||
|
||
class _FailingHopPage(_FakePage):
|
||
"""Побочная (hop) страница, у которой goto всегда падает — для теста best-effort."""
|
||
|
||
async def goto(self, url: str, **kwargs: Any) -> None:
|
||
raise TimeoutError("hop nav timeout (simulated)")
|
||
|
||
|
||
class _FakeContext:
|
||
def __init__(self, links: list[str] | None = None, hop_page_cls: type = _FakePage) -> None:
|
||
self.add_cookies_calls: list[list[dict]] = []
|
||
self.pages: list[_FakePage] = []
|
||
self._links = links or []
|
||
self._hop_page_cls = hop_page_cls
|
||
self.closed = 0
|
||
|
||
async def add_cookies(self, cookies: list[dict]) -> None:
|
||
self.add_cookies_calls.append(cookies)
|
||
|
||
async def new_page(self) -> _FakePage:
|
||
# Первая страница — "основная" (несёт links для hop-детекции), последующие —
|
||
# побочные (hop) вкладки, открытые из _hop_to_recommended_link.
|
||
cls = _FakePage if not self.pages else self._hop_page_cls
|
||
page = cls(self, links=self._links)
|
||
self.pages.append(page)
|
||
return page
|
||
|
||
async def close(self) -> None:
|
||
self.closed += 1
|
||
|
||
|
||
class _FakeReuseBrowser:
|
||
def __init__(self, links: list[str] | None = None, hop_page_cls: type = _FakePage) -> None:
|
||
self.contexts_created: list[_FakeContext] = []
|
||
self._links = links or []
|
||
self._hop_page_cls = hop_page_cls
|
||
|
||
async def new_context(self) -> _FakeContext:
|
||
ctx = _FakeContext(links=self._links, hop_page_cls=self._hop_page_cls)
|
||
self.contexts_created.append(ctx)
|
||
return ctx
|
||
|
||
async def new_page(self) -> _FakePage:
|
||
# Не переиспользуемый путь (reuse_context=False) в этих тестах не участвует.
|
||
raise AssertionError("browser.new_page() не должен вызываться при reuse_context=True")
|
||
|
||
|
||
# ── 1. behaviour отсутствует — ни одна новая ветка не исполняется ────────────────
|
||
|
||
|
||
def test_behaviour_absent_is_noop() -> None:
|
||
browser = _FakeReuseBrowser()
|
||
server._browsers["avito"] = browser
|
||
|
||
html = asyncio.run(
|
||
server._fetch_once("avito", "https://www.avito.ru/1", reuse_context=True)
|
||
)
|
||
|
||
assert html == _DEFAULT_HTML
|
||
page = browser.contexts_created[0].pages[0]
|
||
assert page.mouse.wheel_calls == [], "без behaviour scroll не должен вызываться"
|
||
# Единственный wait_for_timeout — обязательный BROWSER_WAIT_MS settle, не dwell.
|
||
assert page.wait_for_timeout_calls == [server.BROWSER_WAIT_MS]
|
||
assert len(browser.contexts_created[0].pages) == 1, "hop_recommended не должен заходить"
|
||
|
||
|
||
# ── 2. scroll=True — 3-6 вызовов wheel в заданном диапазоне px ───────────────────
|
||
|
||
|
||
def test_scroll_true_triggers_wheel_calls_in_range() -> None:
|
||
browser = _FakeReuseBrowser()
|
||
server._browsers["avito"] = browser
|
||
behaviour = server._Behaviour(scroll=True, dwell_ms=None, hop_recommended=False)
|
||
|
||
asyncio.run(
|
||
server._fetch_once(
|
||
"avito", "https://www.avito.ru/1", reuse_context=True, behaviour=behaviour
|
||
)
|
||
)
|
||
|
||
calls = browser.contexts_created[0].pages[0].mouse.wheel_calls
|
||
assert server.BROWSER_SCROLL_MIN_COUNT <= len(calls) <= server.BROWSER_SCROLL_MAX_COUNT
|
||
for dx, dy in calls:
|
||
assert dx == 0
|
||
assert server.BROWSER_SCROLL_MIN_PX <= dy <= server.BROWSER_SCROLL_MAX_PX
|
||
|
||
|
||
# ── 3. dwell_ms — валидация в fetch_handler (400, как fetch_mode) ────────────────
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"dwell_ms",
|
||
[
|
||
[500], # не пара
|
||
[5000, 100], # min > max
|
||
[-10, 100], # отрицательное
|
||
["a", "b"], # не int
|
||
"500,1000", # вообще не список
|
||
],
|
||
)
|
||
def test_invalid_dwell_ms_returns_400(dwell_ms: Any) -> None:
|
||
request = _make_fetch_request(
|
||
{"url": "https://www.avito.ru/1", "behaviour": {"dwell_ms": dwell_ms}}
|
||
)
|
||
response = asyncio.run(server.fetch_handler(request))
|
||
assert response.status == 400
|
||
assert "dwell_ms" in _json_body(response)["error"]
|
||
|
||
|
||
def test_behaviour_not_dict_returns_400() -> None:
|
||
request = _make_fetch_request({"url": "https://www.avito.ru/1", "behaviour": "scroll"})
|
||
response = asyncio.run(server.fetch_handler(request))
|
||
assert response.status == 400
|
||
|
||
|
||
def test_valid_dwell_ms_applies_extra_wait() -> None:
|
||
browser = _FakeReuseBrowser()
|
||
server._browsers["avito"] = browser
|
||
behaviour = server._Behaviour(scroll=False, dwell_ms=(100, 100), hop_recommended=False)
|
||
|
||
asyncio.run(
|
||
server._fetch_once(
|
||
"avito", "https://www.avito.ru/1", reuse_context=True, behaviour=behaviour
|
||
)
|
||
)
|
||
|
||
page = browser.contexts_created[0].pages[0]
|
||
# BROWSER_WAIT_MS settle + ровно 100мс dwell (min==max=100 — детерминировано).
|
||
assert page.wait_for_timeout_calls == [server.BROWSER_WAIT_MS, 100]
|
||
|
||
|
||
# ── 4. hop_recommended — same-host only, best-effort, требует reuse_context ──────
|
||
|
||
|
||
def test_hop_recommended_visits_same_host_link_only() -> None:
|
||
links = [
|
||
"https://www.avito.ru/ekaterinburg/kvartiry/2-k._kvartira_505_m_1216_et._8285475966", # == url, должен быть исключён
|
||
"https://sub.other.ru/kvartiry/foreign_1234567", # чужой хост, исключён
|
||
"https://www.avito.ru/ekaterinburg/kvartiry", # рубрика, не карточка — исключена
|
||
"https://www.avito.ru/rossiya/nedvizhimost", # навигация, исключена
|
||
"https://www.avito.ru/ekaterinburg/kvartiry/1-k._kvartira_302_m_25_et._8242031370", # валидный кандидат
|
||
]
|
||
browser = _FakeReuseBrowser(links=links)
|
||
server._browsers["avito"] = browser
|
||
behaviour = server._Behaviour(scroll=False, dwell_ms=None, hop_recommended=True)
|
||
|
||
asyncio.run(
|
||
server._fetch_once("avito", "https://www.avito.ru/ekaterinburg/kvartiry/2-k._kvartira_505_m_1216_et._8285475966", reuse_context=True, behaviour=behaviour)
|
||
)
|
||
|
||
ctx = browser.contexts_created[0]
|
||
assert len(ctx.pages) == 2, "должна открыться ровно одна побочная (hop) страница"
|
||
hop_page = ctx.pages[1]
|
||
assert hop_page.goto_urls == ["https://www.avito.ru/ekaterinburg/kvartiry/1-k._kvartira_302_m_25_et._8242031370"]
|
||
assert hop_page.closed == 1, "hop-страница должна закрыться"
|
||
|
||
|
||
def test_hop_recommended_skips_same_host_non_card_links() -> None:
|
||
"""#3283: одного host-фильтра мало — без шаблона URL хоп уводил бы на рубрику.
|
||
|
||
Шаблон Авито (путь с /kvartiry/ + id из 7+ цифр) снят на живом замере 31.08.2026.
|
||
"""
|
||
links = [
|
||
"https://www.avito.ru/ekaterinburg/kvartiry",
|
||
"https://www.avito.ru/ekaterinburg/kvartiry/prodam",
|
||
"https://www.avito.ru/company/about",
|
||
]
|
||
browser = _FakeReuseBrowser(links=links)
|
||
server._browsers["avito"] = browser
|
||
behaviour = server._Behaviour(scroll=False, dwell_ms=None, hop_recommended=True)
|
||
|
||
asyncio.run(
|
||
server._fetch_once("avito", "https://www.avito.ru/ekaterinburg/kvartiry/2-k._kvartira_505_m_1216_et._8285475966", reuse_context=True, behaviour=behaviour)
|
||
)
|
||
|
||
assert len(browser.contexts_created[0].pages) == 1, "рубрика не карточка — хопа быть не должно"
|
||
|
||
|
||
def test_hop_recommended_skips_when_only_foreign_host_links() -> None:
|
||
links = ["https://sub.other.ru/kvartiry/foreign_1234567", "https://another.tld/kvartiry/x_7654321"]
|
||
browser = _FakeReuseBrowser(links=links)
|
||
server._browsers["avito"] = browser
|
||
behaviour = server._Behaviour(scroll=False, dwell_ms=None, hop_recommended=True)
|
||
|
||
asyncio.run(
|
||
server._fetch_once(
|
||
"avito", "https://www.avito.ru/1", reuse_context=True, behaviour=behaviour
|
||
)
|
||
)
|
||
|
||
assert len(browser.contexts_created[0].pages) == 1, "чужой хост не должен открывать hop"
|
||
|
||
|
||
def test_hop_recommended_failure_does_not_break_main_response() -> None:
|
||
links = ["https://www.avito.ru/ekaterinburg/kvartiry/1-k._kvartira_302_m_25_et._8242031370"]
|
||
browser = _FakeReuseBrowser(links=links, hop_page_cls=_FailingHopPage)
|
||
server._browsers["avito"] = browser
|
||
behaviour = server._Behaviour(scroll=False, dwell_ms=None, hop_recommended=True)
|
||
|
||
html = asyncio.run(
|
||
server._fetch_once(
|
||
"avito", "https://www.avito.ru/1", reuse_context=True, behaviour=behaviour
|
||
)
|
||
)
|
||
|
||
assert html == _DEFAULT_HTML, "отказ на hop не должен ронять/менять основной ответ"
|
||
|
||
|
||
def test_hop_recommended_noop_without_reuse_context() -> None:
|
||
"""hop_recommended=True, но reuse_context=False — hop не выполняется (нет context)."""
|
||
server._browsers["avito"] = object()
|
||
behaviour = server._Behaviour(scroll=False, dwell_ms=None, hop_recommended=True)
|
||
|
||
# _fetch_once без reuse_context идёт через browser.new_page() — подделаем browser.
|
||
class _PlainBrowser:
|
||
async def new_page(self) -> _FakePage:
|
||
return _FakePage(context=_FakeContext())
|
||
|
||
server._browsers["avito"] = _PlainBrowser()
|
||
|
||
html = asyncio.run(
|
||
server._fetch_once("avito", "https://www.avito.ru/1", behaviour=behaviour)
|
||
)
|
||
assert html == _DEFAULT_HTML
|