From 5964548e3d92731e6603085d9c5cf9182cf33d54 Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Thu, 13 Aug 2026 17:00:12 +0500 Subject: [PATCH] Add /status, small adjustments. --- scrapy/extensions/remote_control.py | 64 ++++++-- scrapy/utils/_remote_control.py | 16 +- tests/test_extension_remote_control.py | 207 ++++++++++++++++++++----- 3 files changed, 232 insertions(+), 55 deletions(-) diff --git a/scrapy/extensions/remote_control.py b/scrapy/extensions/remote_control.py index 0e0b5dbc3..c3f3e0e72 100644 --- a/scrapy/extensions/remote_control.py +++ b/scrapy/extensions/remote_control.py @@ -7,6 +7,7 @@ import hmac import inspect import io import logging +import os import secrets import time import traceback @@ -19,7 +20,8 @@ import scrapy from scrapy import signals from scrapy.exceptions import NotConfigured from scrapy.utils._remote_control import ( - Envelope, + ExecuteResult, + StatusResult, job_files_dir, new_job_file_name, write_job_file, @@ -27,6 +29,7 @@ from scrapy.utils._remote_control import ( from scrapy.utils.asyncio import is_asyncio_available if TYPE_CHECKING: + from datetime import datetime from pathlib import Path # typing.Self requires Python 3.11 @@ -58,6 +61,16 @@ class RemoteControl: Available endpoints: + - ``/status``: expects a ``GET`` request and returns a JSON object with the + following keys: + + - ``pid`` (number): the process ID of the Scrapy process. + - ``spider`` (string): the name of the currently running spider. + - ``project`` (string): the name of the Scrapy project. + - ``scrapy_version`` (string): the version of Scrapy. + - ``start_time`` (number or null): the start time of the Scrapy process + as a UNIX timestamp. + - ``/execute``: expects a ``POST`` request with a JSON object containing the following keys: @@ -74,9 +87,9 @@ class RemoteControl: raised. - ``elapsed_sec`` (number): the number of seconds the code took to run. - ``output_truncated`` (boolean, optional): whether the output was - truncated. + truncated (omitted if ``false``). - ``traceback_truncated`` (boolean, optional): whether the traceback - was truncated. + was truncated (omitted if ``false``). .. versionadded:: VERSION """ @@ -132,6 +145,7 @@ class RemoteControl: try: self._auth_token = secrets.token_urlsafe(32) app = web.Application() + app.router.add_get("/status", self._handle_status, allow_head=False) app.router.add_post("/execute", self._handle_execute) self._runner = web.AppRunner( app, access_log=None, shutdown_timeout=STOP_TIMEOUT @@ -186,14 +200,19 @@ class RemoteControl: self._runner = None self._auth_token = None + async def _handle_status(self, request: web.Request) -> web.Response: + """An aiohttp request handler for the ``/status`` endpoint.""" + if request.method != "GET": + return web.json_response({"error": "method not allowed"}, status=405) + if not self._is_authenticated(request): + return web.json_response({"error": "unauthorized"}, status=401) + return web.json_response(self._get_status()) + async def _handle_execute(self, request: web.Request) -> web.Response: """An aiohttp request handler for the ``/execute`` endpoint.""" - token = request.headers.get("Authorization", "").removeprefix("Bearer ") - if ( - not self._auth_token - or not token.isascii() - or not hmac.compare_digest(token, self._auth_token) - ): + if request.method != "POST": + return web.json_response({"error": "method not allowed"}, status=405) + if not self._is_authenticated(request): return web.json_response({"error": "unauthorized"}, status=401) try: @@ -215,7 +234,7 @@ class RemoteControl: requested_timeout, self._default_timeout, self._max_timeout ) compiled = _compile(body["code"]) - result: Envelope + result: ExecuteResult if isinstance(compiled, CodeType): result = await self._run_code(compiled, timeout) else: @@ -227,7 +246,16 @@ class RemoteControl: } return web.json_response(result) - async def _run_code(self, code_obj: CodeType, timeout: float) -> Envelope: + def _is_authenticated(self, request: web.Request) -> bool: + """Check if the request is authenticated with the correct Bearer token.""" + token = request.headers.get("Authorization", "").removeprefix("Bearer ") + return ( + self._auth_token is not None + and token.isascii() + and hmac.compare_digest(token, self._auth_token) + ) + + async def _run_code(self, code_obj: CodeType, timeout: float) -> ExecuteResult: """Run a compiled code object with a timeout and capture its output.""" buf = io.StringIO() ns = self._make_namespace(buf) @@ -254,7 +282,7 @@ class RemoteControl: tb, tb_was_truncated = _cap(tb, self._traceback_max_bytes) else: tb_was_truncated = False - result: Envelope = { + result: ExecuteResult = { "status": status, "output": output, "traceback": tb, @@ -266,6 +294,18 @@ class RemoteControl: result["traceback_truncated"] = True return result + def _get_status(self) -> StatusResult: + """Return the data for the ``/status`` response.""" + assert self.crawler.spider + start_time: datetime | None = self.crawler.stats.get_value("start_time") + return { + "pid": os.getpid(), + "spider": self.crawler.spider.name, + "project": self.crawler.settings.get("BOT_NAME"), + "scrapy_version": scrapy.__version__, + "start_time": start_time.timestamp() if start_time is not None else None, + } + def _cap(s: str, limit: int) -> tuple[str, bool]: """Cap a string to ``limit`` bytes, appending an inline truncation marker.""" diff --git a/scrapy/utils/_remote_control.py b/scrapy/utils/_remote_control.py index e601fc933..c8423802c 100644 --- a/scrapy/utils/_remote_control.py +++ b/scrapy/utils/_remote_control.py @@ -23,8 +23,18 @@ logger = logging.getLogger(__name__) JOB_FILE_VERSION = 1 -class Envelope(TypedDict): - """The result of one ``/execute`` call.""" +class StatusResult(TypedDict): + """The result of a ``/status`` call.""" + + pid: int + spider: str + project: str | None + scrapy_version: str + start_time: float | None + + +class ExecuteResult(TypedDict): + """The result of an ``/execute`` call.""" status: Literal["ok", "compile_error", "error", "timeout"] output: str @@ -50,7 +60,7 @@ def new_job_file_name() -> str: def write_job_file( path: Path, *, - spider: str | None, + spider: str, project: str | None, scrapy_version: str, port: int, diff --git a/tests/test_extension_remote_control.py b/tests/test_extension_remote_control.py index eb8c7b882..2f084f562 100644 --- a/tests/test_extension_remote_control.py +++ b/tests/test_extension_remote_control.py @@ -4,7 +4,9 @@ import asyncio import contextlib import json import logging +import os from contextlib import asynccontextmanager +from datetime import datetime, timezone from pathlib import Path from typing import TYPE_CHECKING, Any @@ -12,6 +14,7 @@ import aiohttp import pytest from aiohttp import web +import scrapy from scrapy.exceptions import NotConfigured from scrapy.extensions import remote_control from scrapy.extensions.remote_control import ( @@ -60,21 +63,42 @@ async def _started_extension( await extension.stop() -async def _post(extension: RemoteControl, **kwargs: Any) -> tuple[int, Any]: +async def _request( + extension: RemoteControl, method: str, path: str, **kwargs: Any +) -> tuple[int, Any]: assert extension._runner host, port = extension._runner.addresses[0] - url = f"http://{host}:{port}/execute" + url = f"http://{host}:{port}{path}" async with ( aiohttp.ClientSession() as session, - session.post(url, **kwargs) as response, + session.request(method, url, **kwargs) as response, ): - return response.status, await response.json(content_type=None) + try: + return response.status, await response.json(content_type=None) + except json.JSONDecodeError: + return response.status, await response.text() + + +async def _request_execute(extension: RemoteControl, **kwargs: Any) -> tuple[int, Any]: + return await _request(extension, "POST", "/execute", **kwargs) + + +async def _request_status(extension: RemoteControl, **kwargs: Any) -> tuple[int, Any]: + return await _request(extension, "GET", "/status", **kwargs) def _auth(extension: RemoteControl) -> dict[str, str]: return {"Authorization": f"Bearer {extension._auth_token}"} +BAD_AUTH_HEADERS = [ + {}, + {"Authorization": "Bearer nope"}, + {"Authorization": "Bearer ünicode"}, + {"Authorization": "Basic nope"}, +] + + @coroutine_test async def test_ok_output() -> None: extension = _get_extension() @@ -277,73 +301,74 @@ def test_settings_are_applied() -> None: @coroutine_test -async def test_ok_envelope(tmp_path: Path) -> None: +async def test_status_ok(tmp_path: Path) -> None: async with _started_extension(tmp_path) as extension: - status, envelope = await _post( + status, result = await _request_execute( extension, json={"code": "print(6 * 7)"}, headers=_auth(extension) ) assert status == 200 - assert envelope["status"] == "ok" - assert envelope["output"] == "42\n" + assert result["status"] == "ok" + assert result["output"] == "42\n" @coroutine_test -async def test_compile_error_envelope(tmp_path: Path) -> None: +async def test_status_compile_error(tmp_path: Path) -> None: async with _started_extension(tmp_path) as extension: - status, envelope = await _post( + status, result = await _request_execute( extension, json={"code": "def (:"}, headers=_auth(extension) ) assert status == 200 - assert envelope["status"] == "compile_error" - assert "SyntaxError" in envelope["traceback"] + assert result["status"] == "compile_error" + assert "SyntaxError" in result["traceback"] @coroutine_test -async def test_runtime_error_envelope(tmp_path: Path) -> None: +async def test_status_error(tmp_path: Path) -> None: async with _started_extension(tmp_path) as extension: - status, envelope = await _post( + status, result = await _request_execute( extension, json={"code": "raise ValueError('boom')"}, headers=_auth(extension), ) assert status == 200 - assert envelope["status"] == "error" - assert "boom" in envelope["traceback"] + assert result["status"] == "error" + assert "boom" in result["traceback"] @coroutine_test -async def test_live_crawler_is_reachable(tmp_path: Path) -> None: - # Using the real token covers the auth path end to end as well. +async def test_crawler_var(tmp_path: Path) -> None: async with _started_extension(tmp_path) as extension: - status, envelope = await _post( + status, result = await _request_execute( extension, - json={"code": "print(type(crawler).__name__)"}, + json={"code": "print(type(crawler).__name__, crawler.crawling)"}, headers=_auth(extension), ) assert status == 200 - assert envelope["status"] == "ok" - assert "Crawler" in envelope["output"] + assert result["status"] == "ok" + assert result["output"] == "Crawler False\n" -@pytest.mark.parametrize( - "headers", - [ - {}, - {"Authorization": "Bearer nope"}, - {"Authorization": "Bearer ünicode"}, - {"Authorization": "Basic nope"}, - ], -) +@pytest.mark.parametrize("headers", BAD_AUTH_HEADERS) @coroutine_test -async def test_unauthorized(tmp_path: Path, headers: dict[str, str]) -> None: +async def test_execute_unauthorized(tmp_path: Path, headers: dict[str, str]) -> None: async with _started_extension(tmp_path) as extension: - status, body = await _post( + status, body = await _request_execute( extension, json={"code": "print(1)"}, headers=headers ) assert status == 401 assert body == {"error": "unauthorized"} +@coroutine_test +async def test_execute_rejects_other_methods(tmp_path: Path) -> None: + async with _started_extension(tmp_path) as extension: + results = [ + await _request(extension, method, "/execute", headers=_auth(extension)) + for method in ("HEAD", "GET", "PUT", "DELETE") + ] + assert [status for status, _ in results] == [405] * 4 + + @pytest.mark.parametrize( ("kwargs", "error"), [ @@ -354,18 +379,22 @@ async def test_unauthorized(tmp_path: Path, headers: dict[str, str]) -> None: ], ) @coroutine_test -async def test_bad_requests(tmp_path: Path, kwargs: dict[str, Any], error: str) -> None: +async def test_execute_bad_requests( + tmp_path: Path, kwargs: dict[str, Any], error: str +) -> None: async with _started_extension(tmp_path) as extension: - status, body = await _post(extension, headers=_auth(extension), **kwargs) + status, body = await _request_execute( + extension, headers=_auth(extension), **kwargs + ) assert status == 400 assert body == {"error": error} @pytest.mark.parametrize("timeout_sec", ["abc", [1]]) @coroutine_test -async def test_bad_timeout_type(tmp_path: Path, timeout_sec: Any) -> None: +async def test_execute_bad_timeout_type(tmp_path: Path, timeout_sec: Any) -> None: async with _started_extension(tmp_path) as extension: - status, body = await _post( + status, body = await _request_execute( extension, json={"code": "print(1)", "timeout_sec": timeout_sec}, headers=_auth(extension), @@ -376,9 +405,11 @@ async def test_bad_timeout_type(tmp_path: Path, timeout_sec: Any) -> None: @pytest.mark.parametrize("timeout_sec", [None, 0, -1]) @coroutine_test -async def test_unset_timeout_is_accepted(tmp_path: Path, timeout_sec: Any) -> None: +async def test_execute_unset_timeout_is_accepted( + tmp_path: Path, timeout_sec: Any +) -> None: async with _started_extension(tmp_path) as extension: - status, envelope = await _post( + status, envelope = await _request_execute( extension, json={"code": "print(1)", "timeout_sec": timeout_sec}, headers=_auth(extension), @@ -387,6 +418,102 @@ async def test_unset_timeout_is_accepted(tmp_path: Path, timeout_sec: Any) -> No assert envelope["status"] == "ok" +def test_get_status_fields() -> None: + extension = _get_extension() + assert extension._get_status() == { + "pid": os.getpid(), + "spider": extension.crawler.spidercls.name, + "project": "scrapybot", + "scrapy_version": scrapy.__version__, + "start_time": None, + } + + +def test_get_status_project() -> None: + extension = _get_extension({"BOT_NAME": "my_project"}) + assert extension._get_status()["project"] == "my_project" + + +def test_get_status_project_unset() -> None: + extension = _get_extension({"BOT_NAME": None}) + assert extension._get_status()["project"] is None + + +def test_get_status_start_time() -> None: + extension = _get_extension() + start_time = datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc) + extension.crawler.stats.set_value("start_time", start_time) + assert extension._get_status()["start_time"] == start_time.timestamp() + + +@coroutine_test +async def test_status(tmp_path: Path) -> None: + async with _started_extension(tmp_path) as extension: + extension.crawler.stats.set_value( + "start_time", datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc) + ) + status, body = await _request_status(extension, headers=_auth(extension)) + assert status == 200 + assert body == { + "pid": os.getpid(), + "spider": extension.crawler.spidercls.name, + "project": "scrapybot", + "scrapy_version": scrapy.__version__, + "start_time": datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc).timestamp(), + } + + +@coroutine_test +async def test_compare_status_with_job_file(tmp_path: Path) -> None: + async with _started_extension(tmp_path) as extension: + status, body = await _request_status(extension, headers=_auth(extension)) + (job_file,) = tmp_path.glob("*.json") + record = json.loads(job_file.read_text(encoding="utf-8")) + assert status == 200 + for key in ("pid", "spider", "project", "scrapy_version"): + assert body[key] == record[key] + + +@pytest.mark.parametrize("headers", BAD_AUTH_HEADERS) +@coroutine_test +async def test_status_unauthorized(tmp_path: Path, headers: dict[str, str]) -> None: + async with _started_extension(tmp_path) as extension: + status, body = await _request_status(extension, headers=headers) + assert status == 401 + assert body == {"error": "unauthorized"} + + +@coroutine_test +async def test_status_rejects_other_methods(tmp_path: Path) -> None: + async with _started_extension(tmp_path) as extension: + results = [ + await _request(extension, method, "/status", headers=_auth(extension)) + for method in ("HEAD", "POST", "PUT", "DELETE") + ] + assert [status for status, _ in results] == [405] * 4 + + +@coroutine_test +async def test_status_answers_while_code_is_running(tmp_path: Path) -> None: + async with _started_extension(tmp_path) as extension: + request = asyncio.ensure_future( + _request_execute( + extension, + json={"code": "import asyncio\nawait asyncio.sleep(30)"}, + headers=_auth(extension), + ) + ) + await asyncio.sleep(0.1) # let the request reach the handler + status, body = await asyncio.wait_for( + _request_status(extension, headers=_auth(extension)), 10 + ) + request.cancel() + with contextlib.suppress(asyncio.CancelledError, aiohttp.ClientError): + await request + assert status == 200 + assert body["pid"] == os.getpid() + + @coroutine_test async def test_job_file_written_and_removed(tmp_path: Path) -> None: async with _started_extension(tmp_path) as extension: @@ -492,7 +619,7 @@ async def test_stop_does_not_wait_for_a_running_snippet( extension.crawler.spider = extension.crawler.spidercls() await extension.start() request = asyncio.ensure_future( - _post( + _request_execute( extension, json={"code": "import asyncio\nawait asyncio.sleep(30)"}, headers=_auth(extension),