robotex/tests/browser/test_e2e.py

"""Сквозные тесты с настоящим 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 = '<pre id="fp">' + 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(''),
	}) + '</pre>';
}"""


@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"])