From 14f49ab63c54007a8914706454ba69db4199845e Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Sun, 3 May 2026 14:11:55 +0500 Subject: [PATCH] Remove deprecated test utils. --- conftest.py | 3 -- pyproject.toml | 1 - scrapy/utils/test.py | 103 --------------------------------------- scrapy/utils/testproc.py | 78 ----------------------------- scrapy/utils/testsite.py | 65 ------------------------ 5 files changed, 250 deletions(-) delete mode 100644 scrapy/utils/testproc.py delete mode 100644 scrapy/utils/testsite.py diff --git a/conftest.py b/conftest.py index d49901a7c..4f8d32e1b 100644 --- a/conftest.py +++ b/conftest.py @@ -23,9 +23,6 @@ def _py_files(folder): collect_ignore = [ # may need extra deps "docs/_ext", - # not a test, but looks like a test - "scrapy/utils/testproc.py", - "scrapy/utils/testsite.py", # contains scripts to be run by tests/test_crawler.py::AsyncCrawlerProcessSubprocess *_py_files("tests/AsyncCrawlerProcess"), # contains scripts to be run by tests/test_crawler.py::AsyncCrawlerRunnerSubprocess diff --git a/pyproject.toml b/pyproject.toml index d96790873..d35b333ff 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -127,7 +127,6 @@ ignore_errors = true module = [ "scrapy.core.downloader.webclient", "scrapy.spiders.init", - "scrapy.utils.testsite", "tests.test_webclient", ] allow_any_generics = true diff --git a/scrapy/utils/test.py b/scrapy/utils/test.py index 5c69dc76d..b4e20c3c6 100644 --- a/scrapy/utils/test.py +++ b/scrapy/utils/test.py @@ -7,20 +7,14 @@ from __future__ import annotations import asyncio import os import warnings -from ftplib import FTP from importlib import import_module from pathlib import Path -from posixpath import split from typing import TYPE_CHECKING, Any, TypeVar, cast -from unittest import mock -from twisted.trial.unittest import SkipTest from twisted.web.client import Agent from scrapy.crawler import AsyncCrawlerRunner, CrawlerRunner, CrawlerRunnerBase from scrapy.exceptions import ScrapyDeprecationWarning -from scrapy.utils.boto import is_botocore_available -from scrapy.utils.deprecate import create_deprecated_class from scrapy.utils.reactor import is_asyncio_reactor_installed, is_reactor_installed from scrapy.utils.spider import DefaultSpider @@ -37,80 +31,6 @@ if TYPE_CHECKING: _T = TypeVar("_T") -def assert_gcs_environ() -> None: # pragma: no cover - warnings.warn( - "The assert_gcs_environ() function is deprecated and will be removed in a future version of Scrapy." - " Check GCS_PROJECT_ID directly.", - category=ScrapyDeprecationWarning, - stacklevel=2, - ) - if "GCS_PROJECT_ID" not in os.environ: - raise SkipTest("GCS_PROJECT_ID not found") - - -def skip_if_no_boto() -> None: # pragma: no cover - warnings.warn( - "The skip_if_no_boto() function is deprecated and will be removed in a future version of Scrapy." - " Check scrapy.utils.boto.is_botocore_available() directly.", - category=ScrapyDeprecationWarning, - stacklevel=2, - ) - if not is_botocore_available(): - raise SkipTest("missing botocore library") - - -def get_gcs_content_and_delete( - bucket: Any, path: str -) -> tuple[bytes, list[dict[str, str]], Any]: # pragma: no cover - from google.cloud import storage # noqa: PLC0415 - - warnings.warn( - "The get_gcs_content_and_delete() function is deprecated and will be removed in a future version of Scrapy.", - category=ScrapyDeprecationWarning, - stacklevel=2, - ) - client = storage.Client(project=os.environ.get("GCS_PROJECT_ID")) - bucket = client.get_bucket(bucket) - blob = bucket.get_blob(path) - content = blob.download_as_string() - acl = list(blob.acl) # loads acl before it will be deleted - bucket.delete_blob(path) - return content, acl, blob - - -def get_ftp_content_and_delete( - path: str, - host: str, - port: int, - username: str, - password: str, - use_active_mode: bool = False, -) -> bytes: # pragma: no cover - warnings.warn( - "The get_ftp_content_and_delete() function is deprecated and will be removed in a future version of Scrapy.", - category=ScrapyDeprecationWarning, - stacklevel=2, - ) - ftp = FTP() - ftp.connect(host, port) - ftp.login(username, password) - if use_active_mode: - ftp.set_pasv(False) - ftp_data: list[bytes] = [] - - def buffer_data(data: bytes) -> None: - ftp_data.append(data) - - ftp.retrbinary(f"RETR {path}", buffer_data) - dirname, filename = split(path) - ftp.cwd(dirname) - ftp.delete(filename) - return b"".join(ftp_data) - - -TestSpider = create_deprecated_class("TestSpider", DefaultSpider) - - def get_reactor_settings() -> dict[str, Any]: """Return a settings dict that works with the installed reactor. @@ -185,29 +105,6 @@ def get_from_asyncio_queue(value: _T) -> Awaitable[_T]: return getter -def mock_google_cloud_storage() -> tuple[Any, Any, Any]: # pragma: no cover - """Creates autospec mocks for google-cloud-storage Client, Bucket and Blob - classes and set their proper return values. - """ - from google.cloud.storage import Blob, Bucket, Client # noqa: PLC0415 - - warnings.warn( - "The mock_google_cloud_storage() function is deprecated and will be removed in a future version of Scrapy.", - category=ScrapyDeprecationWarning, - stacklevel=2, - ) - - client_mock = mock.create_autospec(Client) - - bucket_mock = mock.create_autospec(Bucket) - client_mock.get_bucket.return_value = bucket_mock - - blob_mock = mock.create_autospec(Blob) - bucket_mock.blob.return_value = blob_mock - - return (client_mock, bucket_mock, blob_mock) - - def get_web_client_agent_req(url: str) -> Deferred[TxResponse]: # pragma: no cover warnings.warn( "The get_web_client_agent_req() function is deprecated" diff --git a/scrapy/utils/testproc.py b/scrapy/utils/testproc.py deleted file mode 100644 index 7d548524a..000000000 --- a/scrapy/utils/testproc.py +++ /dev/null @@ -1,78 +0,0 @@ -# pragma: no file cover -from __future__ import annotations - -import os -import sys -import warnings -from typing import TYPE_CHECKING, ClassVar, cast - -from twisted.internet.defer import Deferred -from twisted.internet.protocol import ProcessProtocol - -from scrapy.exceptions import ScrapyDeprecationWarning - -if TYPE_CHECKING: - from collections.abc import Iterable - - from twisted.internet.error import ProcessTerminated - from twisted.python.failure import Failure - - -warnings.warn( - "The scrapy.utils.testproc module is deprecated.", - ScrapyDeprecationWarning, - stacklevel=2, -) - - -class ProcessTest: - command: str | None = None - prefix: ClassVar[list[str]] = [sys.executable, "-m", "scrapy.cmdline"] - cwd = os.getcwd() # trial chdirs to temp dir # noqa: PTH109 - - def execute( - self, - args: Iterable[str], - check_code: bool = True, - settings: str | None = None, - ) -> Deferred[TestProcessProtocol]: - from twisted.internet import reactor - - env = os.environ.copy() - if settings is not None: - env["SCRAPY_SETTINGS_MODULE"] = settings - assert self.command - cmd = [*self.prefix, self.command, *args] - pp = TestProcessProtocol() - pp.deferred.addCallback(self._process_finished, cmd, check_code) - reactor.spawnProcess(pp, cmd[0], cmd, env=env, path=self.cwd) - return pp.deferred - - def _process_finished( - self, pp: TestProcessProtocol, cmd: list[str], check_code: bool - ) -> tuple[int, bytes, bytes]: - if pp.exitcode and check_code: - msg = f"process {cmd} exit with code {pp.exitcode}" - msg += f"\n>>> stdout <<<\n{pp.out.decode()}" - msg += "\n" - msg += f"\n>>> stderr <<<\n{pp.err.decode()}" - raise RuntimeError(msg) - return cast("int", pp.exitcode), pp.out, pp.err - - -class TestProcessProtocol(ProcessProtocol): - def __init__(self) -> None: - self.deferred: Deferred[TestProcessProtocol] = Deferred() - self.out: bytes = b"" - self.err: bytes = b"" - self.exitcode: int | None = None - - def outReceived(self, data: bytes) -> None: - self.out += data - - def errReceived(self, data: bytes) -> None: - self.err += data - - def processEnded(self, status: Failure) -> None: - self.exitcode = cast("ProcessTerminated", status.value).exitCode - self.deferred.callback(self) diff --git a/scrapy/utils/testsite.py b/scrapy/utils/testsite.py deleted file mode 100644 index 4089e86ed..000000000 --- a/scrapy/utils/testsite.py +++ /dev/null @@ -1,65 +0,0 @@ -# pragma: no file cover -import warnings -from urllib.parse import urljoin - -from twisted.web import resource, server, static, util - -from scrapy.exceptions import ScrapyDeprecationWarning - -warnings.warn( - "The scrapy.utils.testsite module is deprecated.", - ScrapyDeprecationWarning, - stacklevel=2, -) - - -class SiteTest: - def setUp(self): - from twisted.internet import reactor - - super().setUp() - self.site = reactor.listenTCP(0, test_site(), interface="127.0.0.1") - self.baseurl = f"http://localhost:{self.site.getHost().port}/" - - def tearDown(self): - super().tearDown() - self.site.stopListening() - - def url(self, path: str) -> str: - return urljoin(self.baseurl, path) - - -class NoMetaRefreshRedirect(util.Redirect): - def render(self, request: server.Request) -> bytes: - content = util.Redirect.render(self, request) - return content.replace( - b'http-equiv="refresh"', b'http-no-equiv="do-not-refresh-me"' - ) - - -def test_site(): - r = resource.Resource() - r.putChild(b"text", static.Data(b"Works", "text/plain")) - r.putChild( - b"html", - static.Data( - b"

Works

World

", - "text/html", - ), - ) - r.putChild( - b"enc-gb18030", - static.Data(b"

gb18030 encoding

", "text/html; charset=gb18030"), - ) - r.putChild(b"redirect", util.Redirect(b"/redirected")) - r.putChild(b"redirect-no-meta-refresh", NoMetaRefreshRedirect(b"/redirected")) - r.putChild(b"redirected", static.Data(b"Redirected here", "text/plain")) - return server.Site(r) - - -if __name__ == "__main__": - from twisted.internet import reactor # pylint: disable=ungrouped-imports - - port = reactor.listenTCP(0, test_site(), interface="127.0.0.1") - print(f"http://localhost:{port.getHost().port}/") - reactor.run()