diff --git a/Makefile b/Makefile index c33d653..a730bb9 100644 --- a/Makefile +++ b/Makefile @@ -1,14 +1,14 @@ VENV := .venv PYTHON := $(VENV)/bin/python -PIP := $(VENV)/bin/pip +PIP := $(PYTHON) -m pip WHEELS_DIR := wheels REQUIREMENTS := requirements.txt ver ?= latest -MAKECMDGOALS_TARGETS := install freeze wheels sync sync-offline clean help build release fp-check +MAKECMDGOALS_TARGETS := install freeze wheels sync sync-offline clean help build release sync-dev test test-browser PKG := $(filter-out $(MAKECMDGOALS_TARGETS),$(MAKECMDGOALS)) -.PHONY: install freeze wheels sync sync-offline clean help venv build release fp-check +.PHONY: install freeze wheels sync sync-offline clean help venv build release sync-dev test test-browser help: @echo "make install [ ...] - установить пакет(ы) в $(VENV), обновить $(REQUIREMENTS), собрать wheel в $(WHEELS_DIR)/" @@ -19,7 +19,9 @@ help: @echo "make clean - удалить $(WHEELS_DIR)/" @echo "make build [ver=latest] - собрать docker-образ robotex:" @echo "make release ver= - собрать образ и пакет для деплоя в build// (compose, .env, образ .tar.gz)" - @echo "make fp-check [ver=latest] - собрать образ и проверить, что отпечатки профилей различаются и стабильны" + @echo "make sync-dev - установить зависимости для тестов (requirements-dev.txt)" + @echo "make test - юнит-тесты без браузера" + @echo "make test-browser [ver=latest] - собрать образ и прогнать сквозные тесты с Chrome в Docker (нужна сеть)" venv: @test -d $(VENV) || python3 -m venv $(VENV) @@ -54,9 +56,18 @@ build: @echo "build an application" docker build -t robotex:$(ver) . -fp-check: build - docker run --rm --shm-size=2g -v $(CURDIR)/tests:/app/tests:ro robotex:$(ver) \ - xvfb-run -a --server-args="-screen 0 1920x1080x24" python /app/tests/fingerprint_check.py +sync-dev: venv + $(PIP) install -r requirements-dev.txt + +test: + $(PYTHON) -m pytest + +test-browser: build + docker run --rm --shm-size=2g \ + -v $(CURDIR)/tests:/app/tests:ro -v $(CURDIR)/pytest.ini:/app/pytest.ini:ro \ + -v $(CURDIR)/requirements-dev.txt:/app/requirements-dev.txt:ro \ + -w /app robotex:$(ver) sh -c 'pip install -q --user -r requirements-dev.txt && \ + xvfb-run -a --server-args="-screen 0 1920x1080x24" python -m pytest -m browser -p no:cacheprovider -s' RELEASE_DIR := build/$(ver) diff --git a/README.md b/README.md index 4c59d92..46ee3e0 100644 --- a/README.md +++ b/README.md @@ -67,8 +67,9 @@ src/ captcha/ # 2captcha solvers: Turnstile, reCAPTCHA, hCaptcha docs/ # API, deployment, anti-bot notes tests/ + unit/ # pytest without a browser: make test + browser/ # end-to-end with real Chrome in Docker: make test-browser loadtest.py # concurrent load test + docker stats - fingerprint_check.py # profile fingerprints are applied and stable (make fp-check) test_main.http # manual requests (JetBrains HTTP client) wheels/ # wheels for offline install on Ubuntu Noble Dockerfile @@ -131,6 +132,14 @@ cd src && uvicorn main:app --reload Chrome runs headful, so a display is required (locally — your desktop, in Docker — Xvfb). +Tests: + +```bash +make sync-dev # pytest into .venv +make test # unit tests, no browser (~5 s) +make test-browser # build the image and run end-to-end tests with real Chrome (needs network) +``` + ## Docker ```bash diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..61000d7 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,6 @@ +[pytest] +pythonpath = src +testpaths = tests +addopts = -m "not browser" +markers = + browser: настоящий Chrome и сеть, запускать в Docker: make test-browser diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..c341c30 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,5 @@ +# зависимости для тестов, в прод-образ не попадают: make sync-dev +pytest==9.1.1 +pluggy==1.6.0 +iniconfig==2.3.1 +packaging==26.3 diff --git a/src/antibot/known_providers/cloudflare.py b/src/antibot/known_providers/cloudflare.py index e0d0119..ebb9a7d 100644 --- a/src/antibot/known_providers/cloudflare.py +++ b/src/antibot/known_providers/cloudflare.py @@ -195,9 +195,12 @@ class CloudflareSolver(Solver): frame_el = await iframe.frame_element() box = await frame_el.bounding_box() else: - box = await page.query_selector("xpath=//input[@name='cf-turnstile-response']/../div").bounding_box() + container = await page.query_selector("xpath=//input[@name='cf-turnstile-response']/../div") + box = await container.bounding_box() if container else None if info.stage == Stage.CHECKBOX: + if box is None: + raise RuntimeError("turnstile widget not found on the page") # # Calculate the Captcha coordinates for any viewport captcha_x, captcha_y = box["x"] + randint(26, 28), box["y"] + randint(25, 27) diff --git a/tests/browser/test_e2e.py b/tests/browser/test_e2e.py new file mode 100644 index 0000000..e02c367 --- /dev/null +++ b/tests/browser/test_e2e.py @@ -0,0 +1,136 @@ +"""Сквозные тесты с настоящим Chrome и сетью. Запускаются в Docker (нужны Chrome и Xvfb): + + make test-browser +""" +import json +import uuid + +import pytest +from fastapi.testclient import TestClient + +import main +from api import profile + +pytestmark = pytest.mark.browser + +CF_URL = "https://freelancehunt.com/projects" +CF_SELECTOR = 'a[href*="/project/"]' + +# то, что обычно входит в хеш fingerprint-библиотек и видно странице +FP_JS = """async () => { + const c = document.createElement('canvas'); c.width = 240; c.height = 60; + const x = c.getContext('2d'); x.textBaseline = 'top'; x.font = '16px Arial'; + x.fillStyle = '#f60'; x.fillRect(10, 10, 100, 30); x.fillStyle = '#069'; x.fillText('Robotex fp', 4, 20); + const buf = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(c.toDataURL())); + const uad = await navigator.userAgentData.getHighEntropyValues(['fullVersionList']); + document.body.innerHTML = '
' + JSON.stringify({
+		ua: navigator.userAgent,
+		ua_ch: uad.fullVersionList.map(b => b.brand + '/' + b.version.split('.')[0]).join(','),
+		screen: [screen.width, screen.height, devicePixelRatio],
+		canvas: Array.from(new Uint8Array(buf)).slice(0, 6).map(b => b.toString(16).padStart(2, '0')).join(''),
+	}) + '
'; +}""" + + +@pytest.fixture +def client(monkeypatch, tmp_path): + monkeypatch.setattr(main, "USER_DATA_ROOT", tmp_path) + return TestClient(main.app) + + +def solve(client, session_id=None, **body): + client.cookies.clear() + if session_id: + client.cookies.set("session_id", session_id) + resp = client.post("/v1/solve", json={"actions": [], **body}) + assert resp.status_code == 200, resp.text + return resp.json() | {"session_id": resp.cookies["session_id"]} + + +def test_basic_page(client): + data = solve(client, url="https://example.com", screenshot=True, actions=[ + {"type": "wait_for", "selector": "a", "timeout": 10}, + {"type": "inner_html", "selector": "a"}, + ]) + assert data["status"] == "ok", data + assert data["actions"][1]["result"] == "Learn more" + assert len(data["screenshot"]) > 1000 + + +def test_action_failed_on_real_page(client): + data = solve(client, url="https://example.com", actions=[{"type": "inner_html", "selector": "#missing"}]) + assert data["code"] == "action_failed" + assert data["error"] == "inner_html: selector not found: #missing" + + +def test_user_agent_once_and_cookies_replace(client): + url = "https://httpbin.org/cookies" + read = [{"type": "inner_html", "selector": "pre"}] + cookie = lambda name: {"name": name, "value": "1", "url": "https://httpbin.org", "expires": 4102444800} + + first = solve(client, url=url, user_agent="First/1", cookies=[cookie("a")], actions=read) + sid = first["session_id"] + assert first["user_agent"] == "First/1" + assert json.loads(first["actions"][0]["result"])["cookies"] == {"a": "1"} + + # повторный UA игнорируется; без cookies в запросе профильные сохраняются + second = solve(client, sid, url=url, user_agent="Second/2", actions=read) + assert second["user_agent"] == "First/1" + assert json.loads(second["actions"][0]["result"])["cookies"] == {"a": "1"} + + # переданные cookies заменяют профильные + third = solve(client, sid, url=url, cookies=[cookie("b")], actions=read) + assert json.loads(third["actions"][0]["result"])["cookies"] == {"b": "1"} + + +def test_generated_fingerprint_applied_and_stable(client, monkeypatch): + async def page_fp(page): + await page.evaluate(FP_JS) + return json.loads(await page.inner_html("#fp")) + + results = {} + original = main.make_page_action + + def capture(payload): + action = original(payload) + + async def wrapped(page): + out = await action(page) + out["fp"] = await page_fp(page) + return out + return wrapped + + monkeypatch.setattr(main, "make_page_action", capture) + + class Spy(main.AsyncStealthySession): + async def fetch(self, url, page_action): + out = await super().fetch(url, page_action=page_action) + results.setdefault(self._user_data_dir, []).append(out["fp"]) + return out + + monkeypatch.setattr(main, "AsyncStealthySession", Spy) + + sids = [solve(client, url="https://example.com")["session_id"] for _ in range(4)] + solve(client, sids[0], url="https://example.com") + + major = profile.chrome_major() + fps = {} + for d, measured in results.items(): + saved = profile.load(main.USER_DATA_ROOT / d.rsplit("/", 1)[-1]) + fp = saved["fingerprint"] + for m in measured: + assert m["screen"] == [fp["screen"]["width"], fp["screen"]["height"], fp["device_scale_factor"]] + assert f"Chrome/{major}." in m["ua"] and f"Google Chrome/{major}" in m["ua_ch"] + assert all(m == measured[0] for m in measured), "fingerprint changed within a profile" + fps[d] = measured[0] + assert len(fps) == 4 + print("distinct fingerprints:", len({json.dumps(f, sort_keys=True) for f in fps.values()}), "of", len(fps)) + + +def test_cloudflare(client): + data = solve(client, url=CF_URL, captcha_type="cloudflare", timeout=120, actions=[ + {"type": "wait_for", "selector": CF_SELECTOR, "timeout": 60}, + ]) + assert data["status"] == "ok", data.get("error") + assert "cf_clearance" in {c["name"] for c in data["cookies"]} + assert uuid.UUID(data["session_id"]) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..2c61d83 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,6 @@ +import pytest + + +@pytest.fixture +def anyio_backend(): + return "asyncio" diff --git a/tests/fingerprint_check.py b/tests/fingerprint_check.py deleted file mode 100644 index 0bd6c30..0000000 --- a/tests/fingerprint_check.py +++ /dev/null @@ -1,76 +0,0 @@ -"""Проверка отпечатков профилей: сгенерированный отпечаток применяется, стабилен внутри профиля, -и сколько профилей из выборки различаются. - -Запускается внутри образа (нужны Chrome и Xvfb): - - make fp-check -""" -import asyncio -import json -import sys -import tempfile -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) - -from api import profile # noqa: E402 -from engine import AsyncStealthySession # noqa: E402 - -# то, что обычно входит в хеш fingerprint-библиотек и видно странице -FP_JS = """async () => { - const c = document.createElement('canvas'); c.width = 240; c.height = 60; - const x = c.getContext('2d'); x.textBaseline = 'top'; x.font = '16px Arial'; - x.fillStyle = '#f60'; x.fillRect(10, 10, 100, 30); x.fillStyle = '#069'; x.fillText('Robotex fp', 4, 20); - const digest = async s => Array.from(new Uint8Array(await crypto.subtle.digest('SHA-256', new TextEncoder().encode(s)))) - .slice(0, 6).map(b => b.toString(16).padStart(2, '0')).join(''); - const uad = await navigator.userAgentData.getHighEntropyValues(['fullVersionList']); - return { - ua: navigator.userAgent, - ua_ch: uad.fullVersionList.map(b => b.brand + '/' + b.version.split('.')[0]).join(','), - screen: [screen.width, screen.height, devicePixelRatio], - hwc: navigator.hardwareConcurrency, - canvas: await digest(c.toDataURL()), - }; -}""" - - -async def measure(user_data_dir: Path) -> dict: - settings = profile.load_or_create(user_data_dir, None) - kwargs = profile.browser_kwargs(settings) - async with AsyncStealthySession(headless=False, real_chrome=True, user_data_dir=str(user_data_dir), retries=1, **kwargs) as s: - return await s.fetch("https://example.com", page_action=lambda page: page.evaluate(FP_JS)) - - -async def main(count: int = 6) -> int: - dirs = [tempfile.TemporaryDirectory() for _ in range(count)] - try: - fps = [await measure(Path(d.name)) for d in dirs] - again = await measure(Path(dirs[0].name)) - saved = [profile.load(Path(d.name)) for d in dirs] - finally: - for d in dirs: - d.cleanup() - - for i, fp in enumerate(fps): - print(f"profile {i} {json.dumps(fp, ensure_ascii=False)}") - print(f"profile 0 again {json.dumps(again, ensure_ascii=False)}") - - errors = [] - if fps[0] != again: - errors.append("profile 0 fingerprint changed between launches") - major = profile.chrome_major() - for i, (fp, st) in enumerate(zip(fps, saved)): - want = [st["fingerprint"]["screen"]["width"], st["fingerprint"]["screen"]["height"], st["fingerprint"]["device_scale_factor"]] - if fp["screen"] != want: - errors.append(f"profile {i}: screen {fp['screen']} != saved {want}") - if f"Chrome/{major}." not in fp["ua"] or f"Google Chrome/{major}" not in fp["ua_ch"]: - errors.append(f"profile {i}: user agent does not match installed Chrome {major}") - - distinct = len({json.dumps(fp, sort_keys=True) for fp in fps}) - print(f"distinct fingerprints: {distinct} of {count}") - print("\n".join(errors) or "OK: generated fingerprints are applied and stable within a profile") - return 1 if errors else 0 - - -if __name__ == "__main__": - sys.exit(asyncio.run(main())) diff --git a/tests/unit/test_actions.py b/tests/unit/test_actions.py new file mode 100644 index 0000000..1c39b56 --- /dev/null +++ b/tests/unit/test_actions.py @@ -0,0 +1,115 @@ +"""api.actions.execute на поддельной странице: порядок, остановка на ошибке, антибот, длительность.""" +import pytest + +from antibot.orchestrator import ChallengeOutcome +from antibot.base import Stage +from api import actions +from api.schemas import ActionRequest + +pytestmark = pytest.mark.anyio + + +class FakeElement: + def __init__(self, html): + self.html = html + self.hovered = False + + async def inner_html(self): + return self.html + + async def hover(self, **kwargs): + self.hovered = True + + +class FakePage: + def __init__(self, elements=None): + self.elements = elements or {} + self.filled = {} + + async def query_selector(self, selector): + return self.elements.get(selector) + + async def wait_for_selector(self, selector, timeout): + if selector not in self.elements: + raise TimeoutError(f"Timeout {timeout}ms exceeded") + + async def fill(self, selector, value): + self.filled[selector] = value + + +@pytest.fixture +def challenge(monkeypatch): + """Подмена прохождения антибота; outcome можно поменять в тесте.""" + state = {"outcome": ChallengeOutcome(True, reason="antibot not requested"), "calls": 0} + + async def fake(*args, **kwargs): + state["calls"] += 1 + return state["outcome"] + + monkeypatch.setattr(actions, "pass_challenges", fake) + return state + + +def acts(*items): + return [ActionRequest(**a) for a in items] + + +async def test_actions_run_in_order(challenge): + page = FakePage({"h1": FakeElement("Title"), "a": FakeElement("link")}) + result = await actions.execute(page, acts( + {"type": "wait_for", "selector": "h1", "timeout": 1}, + {"type": "fill", "selector": "#q", "value": "text"}, + {"type": "mouse_move", "selector": "a"}, + {"type": "inner_html", "selector": "h1"}, + )) + assert [(r["type"], r["ok"], r.get("result")) for r in result] == [ + ("wait_for", True, None), ("fill", True, None), ("mouse_move", True, None), ("inner_html", True, "Title"), + ] + assert all(r["duration_ms"] >= 0 for r in result) + assert page.filled == {"#q": "text"} + assert page.elements["a"].hovered + + +async def test_stops_on_first_failure(challenge): + page = FakePage({"h1": FakeElement("Title")}) + result = await actions.execute(page, acts( + {"type": "inner_html", "selector": "h1"}, + {"type": "inner_html", "selector": "#missing"}, + {"type": "inner_html", "selector": "h1"}, + )) + assert [r["ok"] for r in result] == [True, False] + assert result[1]["result"] == "selector not found: #missing" + + +@pytest.mark.parametrize("action", [ + {"type": "mouse_move", "selector": "#missing"}, + {"type": "inner_html", "selector": "#missing"}, +]) +async def test_selector_not_found(challenge, action): + result = await actions.execute(FakePage(), acts(action)) + assert result == [{"type": action["type"], "ok": False, "result": "selector not found: #missing", "duration_ms": result[0]["duration_ms"]}] + + +async def test_wait_for_timeout_is_failure(challenge): + result = await actions.execute(FakePage(), acts({"type": "wait_for", "selector": "#x", "timeout": 2})) + assert result[0]["ok"] is False + assert "2000ms" in result[0]["result"] + + +async def test_empty_actions(challenge): + assert await actions.execute(FakePage(), []) == [] + + +async def test_challenge_failed_skips_actions(challenge): + challenge["outcome"] = ChallengeOutcome(False, Stage.CHECKBOX, "cloudflare", "timeout") + page = FakePage() + with pytest.raises(actions.ChallengeFailed, match=r"cloudflare not passed: timeout \(stage: checkbox\)"): + await actions.execute(page, acts({"type": "fill", "selector": "#q", "value": "x"}), antibot_name="cloudflare") + assert page.filled == {} + + +async def test_pass_challenge_action(challenge): + challenge["outcome"] = ChallengeOutcome(True, reason="passed") + result = await actions.execute(FakePage(), acts({"type": "pass_challenge"}), antibot_name="cloudflare") + assert result[0]["ok"] and result[0]["result"] == "passed" + assert challenge["calls"] == 2 diff --git a/tests/unit/test_orchestrator.py b/tests/unit/test_orchestrator.py new file mode 100644 index 0000000..8945624 --- /dev/null +++ b/tests/unit/test_orchestrator.py @@ -0,0 +1,90 @@ +"""Цикл detect → solve → verify в antibot.orchestrator на поддельном провайдере.""" +import pytest +from patchright.async_api import Error as PlaywrightError + +from antibot import orchestrator +from antibot.base import ChallengeInfo, Stage +from antibot.exceptions import ProviderNotDetected + +pytestmark = pytest.mark.anyio + + +class FakePage: + def __init__(self): + self.waits = [] + + async def wait_for_load_state(self, state): + pass + + async def wait_for_timeout(self, ms): + self.waits.append(ms) + + +class FakeProvider: + provider = "fake" + + def __init__(self, detects, solve_error=None): + self.detects = list(detects) # что вернуть/бросить на каждый detect + self.solve_error = solve_error + self.solves = 0 + + async def detect(self, page): + item = self.detects.pop(0) if len(self.detects) > 1 else self.detects[0] + if isinstance(item, Exception): + raise item + return item + + async def solve(self, page, info): + self.solves += 1 + if self.solve_error: + raise self.solve_error + + +def info(success, stage=Stage.CHECKBOX): + return ChallengeInfo(stage=stage, provider="fake", success=success, meta={"reason": "done" if success else None}) + + +@pytest.fixture +def use_provider(monkeypatch): + def install(provider): + async def select_one(page, name, **kwargs): + return provider + monkeypatch.setattr(orchestrator, "select_one", select_one) + return provider + return install + + +async def test_not_requested(): + outcome = await orchestrator.pass_challenges(None, FakePage()) + assert outcome.cleared and outcome.reason == "antibot not requested" + + +async def test_solved_after_one_step(use_provider): + provider = use_provider(FakeProvider([info(False), info(True)])) + outcome = await orchestrator.pass_challenges("fake", FakePage()) + assert outcome.cleared and outcome.reason == "done" + assert provider.solves == 1 + + +async def test_detect_survives_navigation(use_provider): + """Челлендж уходит на целевую страницу посреди detect — повторяем detect, а не падаем.""" + destroyed = PlaywrightError("Page.query_selector: Execution context was destroyed, most likely because of a navigation.") + provider = use_provider(FakeProvider([destroyed, info(True)])) + outcome = await orchestrator.pass_challenges("fake", FakePage()) + assert outcome.cleared + assert provider.solves == 0 + + +async def test_failing_solve_pauses_and_times_out(use_provider): + page = FakePage() + use_provider(FakeProvider([info(False)], solve_error=RuntimeError("click failed"))) + outcome = await orchestrator.pass_challenges("fake", page, timeout_ms=50) + assert not outcome.cleared + assert outcome.reason == "timeout" + assert outcome.last_stage == Stage.CHECKBOX + assert page.waits and set(page.waits) == {orchestrator.SOLVE_RETRY_DELAY_MS} + + +async def test_unknown_provider(): + with pytest.raises(ProviderNotDetected, match="unknown captcha_type 'datadome'"): + await orchestrator.pass_challenges("datadome", FakePage()) diff --git a/tests/unit/test_profile.py b/tests/unit/test_profile.py new file mode 100644 index 0000000..20e01e0 --- /dev/null +++ b/tests/unit/test_profile.py @@ -0,0 +1,67 @@ +import json + +import pytest + +from api import profile + + +@pytest.fixture(autouse=True) +def chrome_155(monkeypatch): + monkeypatch.setattr(profile, "chrome_major", lambda: "155") + + +def test_passed_user_agent_is_saved_once(tmp_path): + first = profile.load_or_create(tmp_path, "First/1") + second = profile.load_or_create(tmp_path, "Second/2") + + assert first == second == {"user_agent": "First/1"} + assert json.loads((tmp_path / profile.SETTINGS_FILE).read_text()) == {"user_agent": "First/1"} + + +def test_generated_fingerprint_without_user_agent(tmp_path): + settings = profile.load_or_create(tmp_path, None) + + assert settings["user_agent"] == profile.UA_TEMPLATE.format(major="155") + screen = settings["fingerprint"]["screen"] + assert screen["width"] >= profile.MIN_SCREEN[0] and screen["height"] >= profile.MIN_SCREEN[1] + assert settings["fingerprint"]["device_scale_factor"] >= 1 + + +def test_generated_fingerprint_is_stable_and_ignores_later_user_agent(tmp_path): + first = profile.load_or_create(tmp_path, None) + assert profile.load_or_create(tmp_path, None) == first + assert profile.load_or_create(tmp_path, "Later/1") == first + + +def test_profiles_get_independent_fingerprints(tmp_path): + """Отпечатки генерируются независимо: из выборки профилей получается больше одного варианта.""" + variants = set() + for i in range(30): + d = tmp_path / str(i) + d.mkdir() + fp = profile.load_or_create(d, None)["fingerprint"] + variants.add((fp["screen"]["width"], fp["screen"]["height"], fp["device_scale_factor"])) + assert len(variants) > 1 + + +def test_generate_falls_back_when_browserforge_fails(monkeypatch): + class Failing: + def generate(self, **kwargs): + raise ValueError("Cannot generate headers") + + monkeypatch.setattr(profile, "FingerprintGenerator", Failing) + fp = profile.generate_fingerprint(attempts=2) + assert fp["screen"] == {"width": 1920, "height": 1080} + assert fp["device_scale_factor"] == 1 + + +def test_browser_kwargs(): + assert profile.browser_kwargs({"user_agent": "UA/1"}) == {"useragent": "UA/1"} + kwargs = profile.browser_kwargs({ + "user_agent": "UA/1", + "fingerprint": {"screen": {"width": 2560, "height": 1440}, "device_scale_factor": 1.25}, + }) + assert kwargs == { + "useragent": "UA/1", + "additional_args": {"screen": {"width": 2560, "height": 1440}, "device_scale_factor": 1.25}, + } diff --git a/tests/unit/test_schemas.py b/tests/unit/test_schemas.py new file mode 100644 index 0000000..e883606 --- /dev/null +++ b/tests/unit/test_schemas.py @@ -0,0 +1,75 @@ +import pytest +from pydantic import ValidationError + +from api.schemas import ActionRequest, SolveRequest, WAIT_FOR_DEFAULT_TIMEOUT + + +def solve(**kwargs) -> SolveRequest: + return SolveRequest(**{"url": "https://example.com", "actions": [], **kwargs}) + + +@pytest.mark.parametrize("action, error", [ + ({"type": "nope"}, "unknown action type"), + ({"type": "wait_for"}, "requires: selector"), + ({"type": "fill", "selector": "#q"}, "requires: value"), + ({"type": "fill", "selector": "#q", "value": ""}, "requires: value"), + ({"type": "click"}, "requires: selector"), + ({"type": "delay"}, "requires: value"), + ({"type": "delay", "value": "abc"}, "integer"), + ({"type": "delay", "value": "601"}, "between 0 and 600"), + ({"type": "wait_for", "selector": "a", "timeout": 0}, "between 1 and 7200"), + ({"type": "wait_for", "selector": "a", "timeout": 7201}, "between 1 and 7200"), + ({"type": "captcha", "selector": "#c", "value": "datadome"}, "captcha value must be one of"), +]) +def test_invalid_action(action, error): + with pytest.raises(ValidationError, match=error): + ActionRequest(**action) + + +@pytest.mark.parametrize("action", [ + {"type": "wait_for", "selector": "a", "timeout": 10}, + {"type": "fill", "selector": "#q", "value": "x"}, + {"type": "delay", "value": "0"}, + {"type": "pass_challenge"}, + {"type": "captcha", "selector": "#c", "value": "hcaptcha"}, + {"type": "inner_html", "selector": "h1"}, +]) +def test_valid_action(action): + assert ActionRequest(**action).type == action["type"] + + +def test_wait_for_default_timeout(): + assert ActionRequest(type="wait_for", selector="a").timeout == WAIT_FOR_DEFAULT_TIMEOUT + + +@pytest.mark.parametrize("proxy", [ + "http://host:8080", + "https://user:pass@host:443", + "socks5://10.0.0.1:1080", +]) +def test_valid_proxy(proxy): + assert solve(proxy=proxy).proxy == proxy + + +@pytest.mark.parametrize("proxy, error", [ + ("host:8080", "scheme"), + ("ftp://host:21", "scheme"), + ("http://host", "host:port"), + ("http://:8080", "host:port"), + ("http://host:99999", "invalid proxy url"), +]) +def test_invalid_proxy(proxy, error): + with pytest.raises(ValidationError, match=error): + solve(proxy=proxy) + + +def test_proxy_optional(): + assert solve().proxy is None + assert solve(proxy="").proxy is None + + +def test_captcha_action_requires_key(): + action = {"type": "captcha", "selector": "#c", "value": "recaptcha"} + with pytest.raises(ValidationError, match="requires request field 'captcha'"): + solve(actions=[action]) + assert solve(actions=[action], captcha="key").captcha == "key" diff --git a/tests/unit/test_solve.py b/tests/unit/test_solve.py new file mode 100644 index 0000000..826dd6d --- /dev/null +++ b/tests/unit/test_solve.py @@ -0,0 +1,241 @@ +"""/v1/solve без браузера: движок подменён, проверяются сборка ответа, коды ошибок, слоты, сессии и логи.""" +import asyncio +import logging +import uuid + +import pytest +from fastapi.testclient import TestClient + +import main +from api import actions, profile + + +class FakeContext: + async def cookies(self): + return [{"name": "cf_clearance", "value": "secret-cookie"}] + + +class FakePage: + context = FakeContext() + + async def screenshot(self, type): + return b"png" + + async def inner_html(self, selector): + return "page" + + async def evaluate(self, js): + return "UA/engine" + + +class FakeSession: + """Вместо AsyncStealthySession: запоминает kwargs и вызывает page_action на FakePage.""" + calls: list[dict] = [] + error: Exception | None = None + + def __init__(self, **kwargs): + FakeSession.calls.append(kwargs) + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def fetch(self, url, page_action): + if FakeSession.error: + raise FakeSession.error + return await page_action(FakePage()) + + +@pytest.fixture +def client(monkeypatch, tmp_path): + FakeSession.calls, FakeSession.error = [], None + monkeypatch.setattr(main, "AsyncStealthySession", FakeSession) + monkeypatch.setattr(main, "USER_DATA_ROOT", tmp_path) + monkeypatch.setattr(main, "_slots", asyncio.Semaphore(2)) + monkeypatch.setattr(profile, "chrome_major", lambda: "155") + return TestClient(main.app) + + +def fake_execute(result=None, exc=None): + async def execute(*args, **kwargs): + if exc: + raise exc + return result + return execute + + +def post(client, **body): + return client.post("/v1/solve", json={"url": "https://example.com", "actions": [], **body}) + + +def test_ok(client, monkeypatch): + monkeypatch.setattr(actions, "execute", fake_execute([{"type": "wait_for", "ok": True, "duration_ms": 5}])) + resp = post(client, html=True, screenshot=True) + + assert resp.status_code == 200 + data = resp.json() + assert data["status"] == "ok" + assert "code" not in data and "error" not in data + assert data["user_agent"] == "UA/engine" + assert data["actions"] == [{"type": "wait_for", "ok": True, "duration_ms": 5}] + assert data["html"] == "page" + assert data["screenshot"] == "cG5n" # base64(b"png") + assert data["duration_ms"] >= 0 + assert uuid.UUID(resp.cookies["session_id"]) + + +def test_html_and_screenshot_only_on_request(client, monkeypatch): + monkeypatch.setattr(actions, "execute", fake_execute([])) + data = post(client).json() + assert "html" not in data and "screenshot" not in data + assert data["actions"] == [] + + +def test_action_failed(client, monkeypatch): + monkeypatch.setattr(actions, "execute", fake_execute([ + {"type": "click", "ok": True, "duration_ms": 1}, + {"type": "inner_html", "ok": False, "result": "selector not found: #x", "duration_ms": 1}, + ])) + data = post(client, screenshot=True).json() + assert data["status"] == "err" + assert data["code"] == "action_failed" + assert data["error"] == "inner_html: selector not found: #x" + assert data["screenshot"] + + +def test_challenge_failed(client, monkeypatch): + monkeypatch.setattr(actions, "execute", fake_execute(exc=actions.ChallengeFailed("cloudflare not passed: timeout"))) + data = post(client, screenshot=True).json() + assert data["status"] == "err" + assert data["code"] == "challenge_failed" + assert data["error"] == "cloudflare not passed: timeout" + assert data["actions"] == [] + assert data["cookies"] and data["screenshot"] + + +def test_request_failed(client): + FakeSession.error = RuntimeError("net::ERR_PROXY_CONNECTION_FAILED") + data = post(client).json() + assert data == { + "status": "err", "code": "request_failed", "error": "net::ERR_PROXY_CONNECTION_FAILED", + "actions": [], "duration_ms": data["duration_ms"], + } + + +def test_no_slots(client, monkeypatch): + monkeypatch.setattr(main, "_slots", asyncio.Semaphore(0)) + resp = post(client) + assert resp.status_code == 503 + assert resp.headers["retry-after"] == main.RETRY_AFTER + assert resp.json() == {"status": "err", "code": "no_slots", "error": "No slots available"} + assert FakeSession.calls == [] + + +def test_slot_released_after_request(client, monkeypatch): + sem = asyncio.Semaphore(1) + monkeypatch.setattr(main, "_slots", sem) + monkeypatch.setattr(actions, "execute", fake_execute([])) + FakeSession.error = RuntimeError("boom") + for _ in range(3): + assert post(client).status_code == 200 + assert not sem.locked() + + +def test_invalid_request_is_422_without_browser(client): + resp = post(client, actions=[{"type": "nope"}]) + assert resp.status_code == 422 + assert FakeSession.calls == [] + + +def test_session_cookie_reused(client, monkeypatch, tmp_path): + monkeypatch.setattr(actions, "execute", fake_execute([])) + sid = str(uuid.uuid4()) + client.cookies.set("session_id", sid) + resp = post(client) + assert resp.cookies["session_id"] == sid + assert FakeSession.calls[0]["user_data_dir"] == str(tmp_path / sid) + + +@pytest.mark.parametrize("bad", ["../../etc", "not-a-uuid", "/tmp/x"]) +def test_bad_session_cookie_gets_new_session(client, monkeypatch, tmp_path, bad): + monkeypatch.setattr(actions, "execute", fake_execute([])) + client.cookies.set("session_id", bad) + resp = post(client) + sid = resp.cookies["session_id"] + assert sid != bad and uuid.UUID(sid) + assert [p.name for p in tmp_path.iterdir()] == [sid] + + +def test_user_agent_set_once_per_profile(client, monkeypatch): + monkeypatch.setattr(actions, "execute", fake_execute([])) + post(client, user_agent="First/1") + post(client, user_agent="Second/2") + assert [c["useragent"] for c in FakeSession.calls] == ["First/1", "First/1"] + + +def test_generated_fingerprint_passed_to_browser(client, monkeypatch): + monkeypatch.setattr(actions, "execute", fake_execute([])) + post(client) + kwargs = FakeSession.calls[0] + assert kwargs["useragent"] == profile.UA_TEMPLATE.format(major="155") + assert set(kwargs["additional_args"]) == {"screen", "device_scale_factor"} + + +def test_cookies_replace_profile_cookies_only_when_passed(client, monkeypatch): + monkeypatch.setattr(actions, "execute", fake_execute([])) + post(client) + post(client, cookies=[{"name": "a", "value": "1", "url": "https://example.com"}]) + post(client, cookies=[]) + assert [c["clear_cookies"] for c in FakeSession.calls] == [False, True, True] + + +def test_proxy_passed_and_geo_resolved(client, monkeypatch): + monkeypatch.setattr(actions, "execute", fake_execute([])) + seen = {} + + async def fake_geo(proxy): + seen["proxy"] = proxy + return {"locale": "de-DE", "timezone_id": "Europe/Berlin"} + + monkeypatch.setattr(main, "resolve_geo", fake_geo) + post(client, proxy="http://u:p@host:3128") + kwargs = FakeSession.calls[0] + assert kwargs["proxy"] == {"server": "http://host:3128", "username": "u", "password": "p"} + assert seen["proxy"] == kwargs["proxy"] + assert (kwargs["locale"], kwargs["timezone_id"]) == ("de-DE", "Europe/Berlin") + + +def test_without_proxy_no_geo(client, monkeypatch): + monkeypatch.setattr(actions, "execute", fake_execute([])) + + async def fail_geo(proxy): + raise AssertionError("geo must not be resolved without proxy") + + monkeypatch.setattr(main, "resolve_geo", fail_geo) + post(client) + assert FakeSession.calls[0]["proxy"] is None + + +def test_secrets_not_logged(client, monkeypatch, caplog): + monkeypatch.setattr(actions, "execute", fake_execute([])) + + async def no_geo(proxy): + return None + + monkeypatch.setattr(main, "resolve_geo", no_geo) + caplog.set_level(logging.DEBUG, logger="main") + post( + client, proxy="http://someuser:TopSecret1@host:3128", captcha="KEYSECRET2", html=True, + cookies=[{"name": "sid", "value": "CookieSecret3", "url": "https://example.com"}], + actions=[{"type": "fill", "selector": "#p", "value": "PassSecret4"}], + ) + log = caplog.text + assert "Incoming payload" in log and "Outgoing payload" in log + for secret in ("someuser", "TopSecret1", "KEYSECRET2", "CookieSecret3", "PassSecret4", "secret-cookie", "page"): + assert secret not in log + + +def test_health(client): + assert client.get("/health").json() == {"status": "ok", "browser_slots": main.BROWSER_SLOTS, "free_slots": 2}