RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove deprecated test utils.

This commit is contained in:
Andrey Rakhmatullin 2026-05-03 14:11:55 +05:00
parent b9c2240040
commit 14f49ab63c
5 changed files with 0 additions and 250 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
conftest.py
View File

@ -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

1
pyproject.toml
View File

@ -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

103
scrapy/utils/test.py
View File

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

78
scrapy/utils/testproc.py
View File

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

65
scrapy/utils/testsite.py
View File

@ -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"<body><p class='one'>Works</p><p class='two'>World</p></body>",
"text/html",
),
)
r.putChild(
b"enc-gb18030",
static.Data(b"<p>gb18030 encoding</p>", "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()