protokeyandClaude Opus 5.5
7b511d5d38
- Сервис api (FastAPI, без Chrome): POST /v2/tasks, GET /v2/tasks/{id} (доступ по cookie
сессии), DELETE /v2/sessions/current; /v1/solve — прежний контракт поверх очереди.
- Сервис worker (dramatiq + RabbitMQ): слоты браузера, окна сессий между задачами
(browser_idle_timeout 0–120 мин, простой — TTL ключа в Redis), heartbeat.
- Маршрутизация задач сессии в персональную очередь воркера с открытым окном.
- Redis: задачи, результаты (RESULT_TTL_HOURS), окна, воркеры. Прокси и ключ 2captcha
в Redis не сохраняются.
- Движок: AsyncStealthySession.run() — действия на текущей странице без навигации.
- Два образа (robotex-api, robotex-worker), compose с rabbitmq и redis,
make up/down/logs/smoke, релиз собирает все образы.
- Тесты: api + worker в одном процессе (fakeredis), настоящий воркер dramatiq на
StubBroker, браузерные тесты на новой схеме, smoke для запущенного стенда.
- Документация: README, API (v2), Deployment, Usage.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
181 lines
6.8 KiB
Python
181 lines
6.8 KiB
Python
"""/v2: задачи через очередь, доступ по cookie сессии, удержание окна и маршрутизация в него (#80–#83)."""
|
||
import uuid
|
||
|
||
import pytest
|
||
|
||
from tasks import store as task_store
|
||
from tasks.broker import COMMON_QUEUE, personal_queue
|
||
from worker.windows import Outcome
|
||
|
||
from .conftest import FakeSession, WORKER_ID
|
||
|
||
|
||
def create(env, **body):
|
||
return env.client.post("/v2/tasks", json={"actions": [], **body})
|
||
|
||
|
||
def get(env, task_id):
|
||
return env.client.get(f"/v2/tasks/{task_id}")
|
||
|
||
|
||
def session_key(sid):
|
||
return f"{task_store.PREFIX}:session:{sid}"
|
||
|
||
|
||
def test_create_and_get_result(env):
|
||
resp = create(env, url="https://example.com", html=True)
|
||
assert resp.status_code == 202
|
||
body = resp.json()
|
||
assert body["status"] == "queued" and body["url"] == f"/v2/tasks/{body['task_id']}"
|
||
assert env.messages[0].queue_name == COMMON_QUEUE
|
||
|
||
task = get(env, body["task_id"]).json()
|
||
assert task["status"] == "done"
|
||
assert task["result"]["status"] == "ok"
|
||
assert task["result"]["html"] == "<body>page</body>"
|
||
assert task["result"]["user_agent"] == "UA/engine"
|
||
|
||
|
||
def test_queued_until_worker_takes_it(env):
|
||
env.auto = False
|
||
task_id = create(env, url="https://example.com").json()["task_id"]
|
||
assert get(env, task_id).json() == {"task_id": task_id, "status": "queued"}
|
||
env.deliver_pending()
|
||
assert get(env, task_id).json()["status"] == "done"
|
||
|
||
|
||
@pytest.mark.parametrize("cookie", [None, "other", "garbage"])
|
||
def test_task_visible_only_to_its_session(env, cookie):
|
||
task_id = create(env, url="https://example.com").json()["task_id"]
|
||
env.client.cookies.clear()
|
||
if cookie:
|
||
env.client.cookies.set("session_id", str(uuid.uuid4()) if cookie == "other" else cookie)
|
||
assert get(env, task_id).status_code == 404
|
||
|
||
|
||
@pytest.mark.parametrize("task_id", [str(uuid.uuid4()), "../etc", "x"])
|
||
def test_unknown_task_404(env, task_id):
|
||
create(env, url="https://example.com")
|
||
assert get(env, task_id).status_code == 404
|
||
|
||
|
||
def test_result_ttl(env):
|
||
task_id = create(env, url="https://example.com").json()["task_id"]
|
||
ttl = env.redis.ttl(f"{task_store.PREFIX}:task:{task_id}")
|
||
assert task_store.RESULT_TTL - 5 < ttl <= task_store.RESULT_TTL
|
||
|
||
|
||
def test_url_required_without_window(env):
|
||
resp = create(env)
|
||
assert resp.status_code == 422
|
||
assert "no open browser window" in resp.text
|
||
assert env.messages == []
|
||
|
||
|
||
@pytest.mark.parametrize("value", [-1, 121])
|
||
def test_idle_timeout_range(env, value):
|
||
assert create(env, url="https://example.com", browser_idle_timeout=value).status_code == 422
|
||
|
||
|
||
def test_window_closed_by_default(env):
|
||
create(env, url="https://example.com")
|
||
sid = env.client.cookies["session_id"]
|
||
assert env.pool.windows == {} and env.pool.used == 0
|
||
assert FakeSession.instances[0].closed
|
||
assert not env.redis.exists(session_key(sid))
|
||
|
||
|
||
def test_window_held_and_reused(env):
|
||
create(env, url="https://example.com", browser_idle_timeout=5)
|
||
sid = env.client.cookies["session_id"]
|
||
engine = FakeSession.instances[0]
|
||
assert not engine.closed and env.pool.used == 1
|
||
assert env.redis.hgetall(session_key(sid)) == {"worker_id": WORKER_ID, "idle_timeout": "5"}
|
||
assert 290 < env.redis.ttl(session_key(sid)) <= 300
|
||
|
||
# следующая задача сессии — в персональную очередь воркера с окном, без url — на текущей странице
|
||
task_id = create(env, browser_idle_timeout=2).json()["task_id"]
|
||
assert env.messages[-1].queue_name == personal_queue(WORKER_ID)
|
||
assert get(env, task_id).json()["result"]["status"] == "ok"
|
||
assert FakeSession.instances == [engine]
|
||
assert engine.calls == [("fetch", "https://example.com"), ("run", None)]
|
||
# простой продлён значением из последней задачи
|
||
assert 110 < env.redis.ttl(session_key(sid)) <= 120
|
||
|
||
# задача с browser_idle_timeout=0 закрывает окно после себя
|
||
create(env, url="https://example.com/next")
|
||
assert engine.closed and env.pool.used == 0
|
||
assert not env.redis.exists(session_key(sid))
|
||
|
||
|
||
def test_cookies_replace_in_open_window(env):
|
||
create(env, url="https://example.com", browser_idle_timeout=5)
|
||
create(env, browser_idle_timeout=5, cookies=[{"name": "a", "value": "1", "url": "https://example.com"}])
|
||
create(env, browser_idle_timeout=5)
|
||
assert FakeSession.instances[0].context.ops == [
|
||
"clear", ("add", [{"name": "a", "value": "1", "url": "https://example.com"}]),
|
||
]
|
||
|
||
|
||
def test_proxy_change_restarts_window(env):
|
||
create(env, url="https://example.com", browser_idle_timeout=5)
|
||
create(env, url="https://example.com", browser_idle_timeout=5, proxy="http://host:3128")
|
||
first, second = FakeSession.instances
|
||
assert first.closed and not second.closed
|
||
assert second.kwargs["proxy"] == {"server": "http://host:3128"}
|
||
assert env.pool.used == 1
|
||
|
||
|
||
def test_failure_in_window_closes_it(env):
|
||
create(env, url="https://example.com", browser_idle_timeout=5)
|
||
sid = env.client.cookies["session_id"]
|
||
FakeSession.error = RuntimeError("net::ERR_TIMED_OUT")
|
||
task_id = create(env, url="https://example.com", browser_idle_timeout=5).json()["task_id"]
|
||
assert get(env, task_id).json()["result"]["code"] == "request_failed"
|
||
assert FakeSession.instances[0].closed and env.pool.used == 0
|
||
assert not env.redis.exists(session_key(sid))
|
||
|
||
|
||
def test_idle_window_expires(env):
|
||
create(env, url="https://example.com", browser_idle_timeout=5)
|
||
sid = env.client.cookies["session_id"]
|
||
env.call(env.pool.expire_idle)
|
||
assert env.pool.used == 1
|
||
|
||
env.redis.delete(session_key(sid)) # TTL истёк
|
||
env.call(env.pool.expire_idle)
|
||
assert env.pool.used == 0 and FakeSession.instances[0].closed
|
||
|
||
|
||
def test_delete_session_closes_window(env):
|
||
create(env, url="https://example.com", browser_idle_timeout=5)
|
||
sid = env.client.cookies["session_id"]
|
||
assert env.client.delete("/v2/sessions/current").json() == {"closed": True}
|
||
assert env.messages[-1].actor_name == "close_window"
|
||
assert env.messages[-1].queue_name == personal_queue(WORKER_ID)
|
||
assert FakeSession.instances[0].closed and env.pool.used == 0
|
||
assert not env.redis.exists(session_key(sid))
|
||
assert env.client.delete("/v2/sessions/current").json() == {"closed": False}
|
||
|
||
|
||
def test_open_windows_take_slots(env):
|
||
for _ in range(2):
|
||
env.client.cookies.clear()
|
||
create(env, url="https://example.com", browser_idle_timeout=5)
|
||
assert env.pool.used == 2
|
||
|
||
env.client.cookies.clear()
|
||
create(env, url="https://example.com")
|
||
assert env.outcomes[-1] == Outcome.NO_SLOT
|
||
# /v1/solve без окна при занятых слотах — сразу 503, как раньше
|
||
env.client.cookies.clear()
|
||
assert env.client.post("/v1/solve", json={"url": "https://example.com", "actions": []}).status_code == 503
|
||
|
||
|
||
def test_dead_worker_window_goes_to_common_queue(env):
|
||
create(env, url="https://example.com", browser_idle_timeout=5)
|
||
env.redis.delete(f"{task_store.PREFIX}:worker:{WORKER_ID}")
|
||
env.auto = False
|
||
create(env, url="https://example.com")
|
||
assert env.messages[-1].queue_name == COMMON_QUEUE
|