gendesign/tradein-mvp/browser/test_server_anchor_origin_switch.py

"""test_server_anchor_origin_switch.py — якорь обязан стоять на ЗАПРОШЕННОМ origin.

Контракт формулируется от того, чего требует площадка, а не от того, что делал код:
Referer целевой навигации называет страницу выдачи, поэтому эта выдача обязана быть
в контексте реально открыта. До этой правки `_ensure_anchor_page` возвращала True на
ЛЮБУЮ живую вкладку, не сверяя её origin с запрошенным. Оба нынешних caller'а выводят
origin из URL карточки, и он меняется вместе с городом (Авито — сегмент пути, Домклик —
поддомен), так что после первой же карточки другого города Referer называл выдачу,
которой этот контекст никогда не открывал: ни куки её, ни тайминга. Ровно тот вид
«заявленного, но не случившегося перехода», который запретил #3258.

camoufox НЕ запускается — context/page подделаны.

Запуск (из tradein-mvp/browser/)::

    python -m pytest test_server_anchor_origin_switch.py -q
"""

from __future__ import annotations

import asyncio
import importlib.util
from pathlib import Path
from typing import Any

import pytest

_SERVER_PATH = Path(__file__).resolve().parent / "server.py"
_spec = importlib.util.spec_from_file_location("tradein_browser_server_anchor_switch", _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)

_PROVIDER = "avito"
_EKB = "https://www.avito.ru/ekaterinburg/kvartiry/prodam"
_PYSHMA = "https://www.avito.ru/verhnyaya_pyshma/kvartiry/prodam"


@pytest.fixture(autouse=True)
def _reset_state() -> None:
    server._contexts = {}
    server._anchor_pages = {}
    server._anchor_origins = {}


class _Page:
    def __init__(self) -> None:
        self.goto_urls: list[str] = []
        self.closed = 0
        self.url = ""

    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)
        self.url = url

    async def wait_for_timeout(self, ms: int) -> None:
        return None

    async def close(self) -> None:
        self.closed += 1


class _Context:
    """new_page() отдаёт КАЖДЫЙ раз новую вкладку — как настоящий BrowserContext."""

    def __init__(self) -> None:
        self.pages: list[_Page] = []

    async def new_page(self) -> _Page:
        page = _Page()
        self.pages.append(page)
        return page

    async def close(self) -> None:
        return None


def test_same_origin_reuses_the_open_tab() -> None:
    """Повторный запрос того же origin не платит за навигацию второй раз."""
    ctx = _Context()
    server._contexts[_PROVIDER] = ctx

    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _EKB)) is True
    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _EKB)) is True

    assert len(ctx.pages) == 1, "вторая вкладка не нужна — origin тот же"
    assert ctx.pages[0].goto_urls == [_EKB]
    assert server._anchor_origins[_PROVIDER] == _EKB


def test_changed_origin_reopens_the_tab_on_the_new_origin() -> None:
    """Другой город — якорь ОБЯЗАН переехать, иначе Referer называет чужую выдачу."""
    ctx = _Context()
    server._contexts[_PROVIDER] = ctx

    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _EKB)) is True
    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _PYSHMA)) is True

    assert len(ctx.pages) == 2, "на новый origin поднимается новая вкладка"
    assert ctx.pages[0].closed == 1, "старая вкладка не должна течь"
    assert ctx.pages[1].goto_urls == [_PYSHMA]
    assert server._anchor_origins[_PROVIDER] == _PYSHMA
    assert server._anchor_pages[_PROVIDER] is ctx.pages[1]


def test_dead_tab_is_replaced_even_on_the_same_origin() -> None:
    """Вкладку закрыли извне (recycle/краш) — поднимаем заново, не отдаём мёртвую."""
    ctx = _Context()
    server._contexts[_PROVIDER] = ctx

    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _EKB)) is True
    asyncio.run(ctx.pages[0].close())

    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _EKB)) is True
    assert len(ctx.pages) == 2
    assert server._anchor_pages[_PROVIDER] is ctx.pages[1]


def test_closing_context_forgets_the_anchor_origin() -> None:
    """Сброс context'а обнуляет и память об origin — иначе новый context унаследует
    чужое значение и вернёт «годен» на вкладку, которой уже нет."""
    ctx = _Context()
    server._contexts[_PROVIDER] = ctx
    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _EKB)) is True

    asyncio.run(server._close_reusable_context(_PROVIDER))

    assert _PROVIDER not in server._anchor_origins
    assert _PROVIDER not in server._anchor_pages


def test_no_context_means_no_anchor() -> None:
    """Без переиспользуемого context'а держать вкладку негде — прежнее поведение."""
    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _EKB)) is False