поведенческое тестирование

Author: Miskler

chizhik_api/manager.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-import os
+from collections import defaultdict
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -12,6 +12,11 @@
     HumanPage,
     api_child_field,
 )
+from human_requests.network_analyzer.anomaly_sniffer import (
+    HeaderAnomalySniffer,
+    WaitHeader,
+    WaitSource,
+)
 from human_requests.abstraction import FetchResponse, HttpMethod, Proxy
 from playwright.async_api import TimeoutError as PWTimeoutError
 
@@ -31,6 +36,8 @@ class ChizhikAPI(ApiParent):
     """Время ожидания ответа от сервера в миллисекундах."""
     headless: bool = True
     """Запускать браузер в headless режиме?"""
+    test_mode: bool = False
+    """Режим тестирования предполагает более глубокий _warmup который не требуется для обычного использования"""
     proxy: str | dict | Proxy | None = field(default_factory=Proxy.from_env)
     """Прокси-сервер для всех запросов (если нужен). По умолчанию берет из окружения (если есть).
     Принимает как формат Playwright, так и строчный формат."""
@@ -51,6 +58,11 @@ class ChizhikAPI(ApiParent):
     page: HumanPage = field(init=False, repr=False)
     """Внутренний страница сессии браузера"""
 
+    unstandard_headers: dict[str, str] = field(init=False, repr=False)
+    """Список нестандартных заголовков пойманных при инициализации"""
+    unstandard_urls: dict[str, list[str]] = field(init=False, repr=False)
+    """Список нестандартных заголовков пойманных при инициализации"""
+
     Geolocation: ClassGeolocation = api_child_field(ClassGeolocation)
     """API для работы с геолокацией."""
     Catalog: ClassCatalog = api_child_field(ClassCatalog)
@@ -74,30 +86,43 @@ async def _warmup(self) -> None:
             proxy=px.as_dict(),
             **self.browser_opts,
             block_images=True,
+            i_know_what_im_doing=True,
         ).start()
 
         self.session = HumanBrowser.replace(br)
         self.ctx = await self.session.new_context()
         self.page = await self.ctx.new_page()
         self.page.on_error_screenshot_path = "screenshot.png"
+
+        if self.test_mode:
+            sniffer = HeaderAnomalySniffer(
+                include_subresources=True,  # или False, если интересны только документы
+                url_filter=lambda u: u.startswith(self.CATALOG_URL),
+            )
+            await sniffer.start(self.ctx)
+
+            await self.page.goto(self.MAIN_SITE_URL, wait_until="networkidle")
+            await self.page.wait_for_selector("next-route-announcer", state="attached")
+            
+            result_sniffer = await sniffer.complete()
+
+            # Результат: {заголовок: [уникальные значения]}
+            result = defaultdict(set)
+
+            # Проходим по всем URL в 'request'
+            for _url, headers in result_sniffer["request"].items():
+                for header, values in headers.items():
+                    result[header].update(values)  # добавляем значения, set уберёт дубли
+
+            # Преобразуем set обратно в list
+            self.unstandard_headers = {k: list(v)[0] for k, v in result.items()}
+            self.unstandard_urls = result_sniffer["request"]
+
         await self.page.goto(self.CATALOG_URL, wait_until="networkidle")
 
-        ok = False
-        try_count = 3
-        while not ok and try_count > 0:
-            try_count -= 1
-            try:
-                await self.page.wait_for_selector(
-                    "pre", timeout=self.timeout_ms, state="attached"
-                )
-                ok = True
-            except PWTimeoutError:
-                await self.page.reload()
-        if not ok:
-            raise RuntimeError(await self.page.content())
-
-        # await self.page.wait_for_load_state("networkidle")
-        # await asyncio.sleep(3)
+        await self.page.wait_for_selector(
+            "pre", timeout=self.timeout_ms, state="attached"
+        )
 
     async def __aexit__(self, *exc):
         """Выход из контекстного менеджера с закрытием сессии."""

tests/__snapshots__/unstandard_headers.schema.json

@@ -0,0 +1,4 @@
+{
+  "type": "object",
+  "$schema": "https://json-schema.org/draft/2020-12/schema"
+}

tests/__snapshots__/unstandard_urls.schema.json

@@ -0,0 +1,4 @@
+{
+  "type": "object",
+  "$schema": "https://json-schema.org/draft/2020-12/schema"
+}

tests/api_test.py

@@ -2,9 +2,10 @@
 
 import aiohttp
 import pytest
-from human_requests import autotest_depends_on, autotest_hook, autotest_params
+from human_requests import autotest_depends_on, autotest_hook, autotest_params, autotest_data
 from human_requests.abstraction import Proxy
-from human_requests.autotest import AutotestCallContext, AutotestContext
+from human_requests.autotest import AutotestCallContext, AutotestContext,AutotestDataContext
+
 from PIL import Image
 
 from chizhik_api import ChizhikAPI
@@ -24,7 +25,7 @@ def anyio_backend():
 @pytest.fixture(scope="session")
 async def api():
     """Фикстура для инициализации API в рамках сессии."""
-    async with ChizhikAPI() as api_instance:
+    async with ChizhikAPI(headless=False, test_mode=True) as api_instance:
         yield api_instance
 
 
@@ -87,6 +88,16 @@ def _product_info_params(ctx: AutotestCallContext) -> dict[str, int]:
     pytest.fail("ProductService.info depends on Catalog.products_list.")
 
 
+
+@autotest_data(name="unstandard_headers")
+def _unstandard_headers_data(ctx: AutotestDataContext) -> dict[str, Any]:
+    return ctx.api.unstandard_headers
+
+@autotest_data(name="unstandard_urls")
+def _unstandard_urls_data(ctx: AutotestDataContext) -> dict[str, Any]:
+    return ctx.api.unstandard_urls
+
+
 async def test_download_image(api):
     url = (
         "https://chizhik.x5static.net/media/chizhik-assets/categories/icon/"