robotex/tests/unit/test_actions.py

"""api.actions.execute на поддельной странице: порядок, остановка на ошибке, антибот, длительность."""
import base64

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


class FakeImage:
	"""Locator картинки капчи: скриншот элемента — заранее заданные байты."""

	def __init__(self, png):
		self.png = png
		self.waited = False

	@property
	def first(self):
		return self

	async def wait_for(self):
		self.waited = True

	async def evaluate(self, script):
		return True

	async def screenshot(self, type):
		return self.png


class ImagePage(FakePage):
	def __init__(self, image):
		super().__init__()
		self.image = image
		self.locators = []

	def locator(self, selector):
		self.locators.append(selector)
		return self.image


async def test_captcha_image(challenge, monkeypatch):
	"""Снимок элемента уходит в 2captcha base64 с подсказками, ответ набирается в target и попадает в result."""
	from captcha import image as image_mod
	from twocaptcha import TwoCaptcha

	sent, typed = {}, {}

	def fake_solve(self, **kwargs):
		sent.update(kwargs)
		return {"captchaId": "1", "code": "x7Kp2"}

	async def fake_typing(page, locator, text):
		typed[locator] = text

	monkeypatch.setattr(TwoCaptcha, "solve", fake_solve)
	monkeypatch.setattr(image_mod, "human_typing", fake_typing)

	page = ImagePage(FakeImage(b"\x89PNG-captcha"))
	result = await actions.execute(page, acts(
		{"type": "captcha", "selector": "#popup-signup img.captcha", "value": "image",
			"target": "#popup-signup input[name=\"captcha\"]", "min_len": 5, "numeric": 4, "case_sensitive": True},
	), captcha_key="KEY")

	assert [(r["ok"], r.get("result")) for r in result] == [(True, "x7Kp2")]
	assert page.locators == ["#popup-signup img.captcha"]
	assert page.image.waited
	assert sent == {
		"method": "base64", "body": base64.b64encode(b"\x89PNG-captcha").decode(),
		"minLen": 5, "numeric": 4, "caseSensitive": 1,
	}
	assert typed == {"#popup-signup input[name=\"captcha\"]": "x7Kp2"}


async def test_captcha_image_unsolved(challenge, monkeypatch):
	"""2captcha не решила — действие падает, в target ничего не вводится."""
	from captcha import image as image_mod
	from twocaptcha import TwoCaptcha
	from twocaptcha.solver import ApiException

	def fake_solve(self, **kwargs):
		raise ApiException("ERROR_CAPTCHA_UNSOLVABLE")

	typed = []

	async def fake_typing(page, locator, text):
		typed.append(text)

	monkeypatch.setattr(TwoCaptcha, "solve", fake_solve)
	monkeypatch.setattr(image_mod, "human_typing", fake_typing)

	result = await actions.execute(ImagePage(FakeImage(b"png")), acts(
		{"type": "captcha", "selector": "img", "value": "image", "target": "#in"},
	), captcha_key="KEY")

	assert [(r["ok"], r["result"]) for r in result] == [(False, "ERROR_CAPTCHA_UNSOLVABLE")]
	assert typed == []