protokeyandClaude Opus 5.5
0ae5b0c83f
- tests/unit (make test, без браузера, ~5 с): схемы и валидация, /v1/solve с подменённым движком (коды ошибок, слоты, сессии, UA и cookies профиля, маскирование секретов в логах), действия, оркестратор антибота, профиль и отпечаток - tests/browser (make test-browser, в Docker): example.com, UA один раз и замена cookies на httpbin, применение и стабильность отпечатка, прохождение Cloudflare - requirements-dev.txt, make sync-dev; Makefile вызывает pip через python -m pip (у перенесённого .venv битый shebang bin/pip) - fix: в CloudflareSolver вызов bounding_box() у неожиданной корутины — ветка без iframe всегда падала с AttributeError Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
116 lines
3.9 KiB
Python
116 lines
3.9 KiB
Python
"""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
|