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

Remove ProcessTest and SiteTest. (#6885) 

* Remove ProcessTest and SiteTest.

* Restore the support for Windows line endings in TestParseCommand.

* Add a test for running a scrapy command in a project subdir.

* Remove pywin32 from test deps.
This commit is contained in:
Andrey Rakhmatullin 2025-06-13 22:18:27 +05:00 committed by GitHub
parent b4d11b8b25
commit 92c18d15b4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 360 additions and 478 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
tests/mockserver.py
View File

@ -18,11 +18,12 @@ from twisted.internet.task import deferLater
from twisted.names import dns, error
from twisted.names.server import DNSServerFactory
from twisted.web import resource, server
from twisted.web.server import NOT_DONE_YET, GzipEncoderFactory, Site
from twisted.web.static import File
from twisted.web.util import redirectTo
from twisted.web.server import NOT_DONE_YET, Site
from twisted.web.static import Data, File
from twisted.web.util import Redirect, redirectTo
from scrapy.utils.python import to_bytes, to_unicode
from tests import tests_datadir
if TYPE_CHECKING:
from twisted.internet.protocol import ServerFactory
@ -245,6 +246,14 @@ class ArbitraryLengthPayloadResource(LeafResource):
return request.content.read()
class NoMetaRefreshRedirect(Redirect):
def render(self, request: server.Request) -> bytes:
content = Redirect.render(self, request)
return content.replace(
b'http-equiv="refresh"', b'http-no-equiv="do-not-refresh-me"'
)
class Root(resource.Resource):
def __init__(self):
resource.Resource.__init__(self)
@ -256,18 +265,26 @@ class Root(resource.Resource):
self.putChild(b"raw", Raw())
self.putChild(b"echo", Echo())
self.putChild(b"payload", PayloadResource())
self.putChild(
b"xpayload",
resource.EncodingResourceWrapper(PayloadResource(), [GzipEncoderFactory()]),
)
self.putChild(b"alpayload", ArbitraryLengthPayloadResource())
try:
from tests import tests_datadir
self.putChild(b"files", File(str(Path(tests_datadir, "test_site/files/"))))
except Exception:
pass
self.putChild(b"files", File(str(Path(tests_datadir, "test_site/files/"))))
self.putChild(b"redirect-to", RedirectTo())
self.putChild(b"text", Data(b"Works", "text/plain"))
self.putChild(
b"html",
Data(
b"<body><p class='one'>Works</p><p class='two'>World</p></body>",
"text/html",
),
)
self.putChild(
b"enc-gb18030",
Data(b"<p>gb18030 encoding</p>", "text/html; charset=gb18030"),
)
self.putChild(b"redirect", Redirect(b"/redirected"))
self.putChild(
b"redirect-no-meta-refresh", NoMetaRefreshRedirect(b"/redirected")
)
self.putChild(b"redirected", Data(b"Redirected here", "text/plain"))
def getChild(self, name, request):
return self

6
tests/test_command_check.py
View File

@ -7,10 +7,8 @@ from tests.test_commands import TestCommandBase
class TestCheckCommand(TestCommandBase):
command = "check"
def setUp(self):
super().setUp()
def setup_method(self):
super().setup_method()
self.spider_name = "check_spider"
self.spider = (self.proj_mod_path / "spiders" / "checkspider.py").resolve()

46
tests/test_command_fetch.py
View File

@ -1,35 +1,35 @@
from twisted.internet.defer import inlineCallbacks
from twisted.trial import unittest
from tests.utils.testproc import ProcessTest
from tests.utils.testsite import SiteTest
from tests.mockserver import MockServer
from tests.test_commands import TestProjectBase
class TestFetchCommand(ProcessTest, SiteTest, unittest.TestCase):
command = "fetch"
class TestFetchCommand(TestProjectBase):
@classmethod
def setup_class(cls):
cls.mockserver = MockServer()
cls.mockserver.__enter__()
@classmethod
def teardown_class(cls):
cls.mockserver.__exit__(None, None, None)
@inlineCallbacks
def test_output(self):
_, out, _ = yield self.execute([self.url("/text")])
assert out.strip() == b"Works"
_, out, _ = self.proc("fetch", self.mockserver.url("/text"))
assert out.strip() == "Works"
@inlineCallbacks
def test_redirect_default(self):
_, out, _ = yield self.execute([self.url("/redirect")])
assert out.strip() == b"Redirected here"
_, out, _ = self.proc("fetch", self.mockserver.url("/redirect"))
assert out.strip() == "Redirected here"
@inlineCallbacks
def test_redirect_disabled(self):
_, out, err = yield self.execute(
["--no-redirect", self.url("/redirect-no-meta-refresh")]
_, _, err = self.proc(
"fetch", "--no-redirect", self.mockserver.url("/redirect-no-meta-refresh")
)
err = err.strip()
assert b"downloader/response_status_count/302" in err, err
assert b"downloader/response_status_count/200" not in err, err
assert "downloader/response_status_count/302" in err, err
assert "downloader/response_status_count/200" not in err, err
@inlineCallbacks
def test_headers(self):
_, out, _ = yield self.execute([self.url("/text"), "--headers"])
out = out.replace(b"\r", b"") # required on win32
assert b"Server: TwistedWeb" in out, out
assert b"Content-Type: text/plain" in out
_, out, _ = self.proc("fetch", self.mockserver.url("/text"), "--headers")
out = out.replace("\r", "") # required on win32
assert "Server: TwistedWeb" in out, out
assert "Content-Type: text/plain" in out

379
tests/test_command_parse.py
View File

@ -1,29 +1,25 @@
import argparse
import os
import re
from pathlib import Path
from twisted.internet.defer import inlineCallbacks
from scrapy.commands import parse
from scrapy.settings import Settings
from scrapy.utils.python import to_unicode
from tests.mockserver import MockServer
from tests.test_commands import TestCommandBase
from tests.utils.testproc import ProcessTest
from tests.utils.testsite import SiteTest
def _textmode(bstr: bytes) -> str:
"""Normalize input the same as writing to a file
and reading from it in text mode"""
return to_unicode(bstr).replace(os.linesep, "\n")
class TestParseCommand(TestCommandBase):
@classmethod
def setup_class(cls):
cls.mockserver = MockServer()
cls.mockserver.__enter__()
@classmethod
def teardown_class(cls):
cls.mockserver.__exit__(None, None, None)
class TestParseCommand(ProcessTest, SiteTest, TestCommandBase):
command = "parse"
def setUp(self):
super().setUp()
def setup_method(self):
super().setup_method()
self.spider_name = "parse_spider"
(self.proj_mod_path / "spiders" / "myspider.py").write_text(
f"""
@ -171,260 +167,253 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}}
"""
)
@inlineCallbacks
def test_spider_arguments(self):
_, _, stderr = yield self.execute(
[
"--spider",
self.spider_name,
"-a",
"test_arg=1",
"-c",
"parse",
"--verbose",
self.url("/html"),
]
_, _, stderr = self.proc(
"parse",
"--spider",
self.spider_name,
"-a",
"test_arg=1",
"-c",
"parse",
"--verbose",
self.mockserver.url("/html"),
)
assert "DEBUG: It Works!" in _textmode(stderr)
assert "DEBUG: It Works!" in stderr
@inlineCallbacks
def test_request_with_meta(self):
raw_json_string = '{"foo" : "baz"}'
_, _, stderr = yield self.execute(
[
"--spider",
self.spider_name,
"--meta",
raw_json_string,
"-c",
"parse_request_with_meta",
"--verbose",
self.url("/html"),
]
_, _, stderr = self.proc(
"parse",
"--spider",
self.spider_name,
"--meta",
raw_json_string,
"-c",
"parse_request_with_meta",
"--verbose",
self.mockserver.url("/html"),
)
assert "DEBUG: It Works!" in _textmode(stderr)
assert "DEBUG: It Works!" in stderr
_, _, stderr = yield self.execute(
[
"--spider",
self.spider_name,
"-m",
raw_json_string,
"-c",
"parse_request_with_meta",
"--verbose",
self.url("/html"),
]
_, _, stderr = self.proc(
"parse",
"--spider",
self.spider_name,
"-m",
raw_json_string,
"-c",
"parse_request_with_meta",
"--verbose",
self.mockserver.url("/html"),
)
assert "DEBUG: It Works!" in _textmode(stderr)
assert "DEBUG: It Works!" in stderr
@inlineCallbacks
def test_request_with_cb_kwargs(self):
raw_json_string = '{"foo" : "bar", "key": "value"}'
_, _, stderr = yield self.execute(
[
"--spider",
self.spider_name,
"--cbkwargs",
raw_json_string,
"-c",
"parse_request_with_cb_kwargs",
"--verbose",
self.url("/html"),
]
_, _, stderr = self.proc(
"parse",
"--spider",
self.spider_name,
"--cbkwargs",
raw_json_string,
"-c",
"parse_request_with_cb_kwargs",
"--verbose",
self.mockserver.url("/html"),
)
log = _textmode(stderr)
assert "DEBUG: It Works!" in log
assert "DEBUG: It Works!" in stderr
assert (
"DEBUG: request.callback signature: (response, foo=None, key=None)" in log
"DEBUG: request.callback signature: (response, foo=None, key=None)"
in stderr
)
@inlineCallbacks
def test_request_without_meta(self):
_, _, stderr = yield self.execute(
[
"--spider",
self.spider_name,
"-c",
"parse_request_without_meta",
"--nolinks",
self.url("/html"),
]
_, _, stderr = self.proc(
"parse",
"--spider",
self.spider_name,
"-c",
"parse_request_without_meta",
"--nolinks",
self.mockserver.url("/html"),
)
assert "DEBUG: It Works!" in _textmode(stderr)
assert "DEBUG: It Works!" in stderr
@inlineCallbacks
def test_pipelines(self):
_, _, stderr = yield self.execute(
[
"--spider",
self.spider_name,
"--pipelines",
"-c",
"parse",
"--verbose",
self.url("/html"),
]
_, _, stderr = self.proc(
"parse",
"--spider",
self.spider_name,
"--pipelines",
"-c",
"parse",
"--verbose",
self.mockserver.url("/html"),
)
assert "INFO: It Works!" in _textmode(stderr)
assert "INFO: It Works!" in stderr
@inlineCallbacks
def test_async_def_asyncio_parse_items_list(self):
status, out, stderr = yield self.execute(
[
"--spider",
"asyncdef_asyncio_return",
"-c",
"parse",
self.url("/html"),
]
_, out, stderr = self.proc(
"parse",
"--spider",
"asyncdef_asyncio_return",
"-c",
"parse",
self.mockserver.url("/html"),
)
assert "INFO: Got response 200" in _textmode(stderr)
assert "{'id': 1}" in _textmode(out)
assert "{'id': 2}" in _textmode(out)
assert "INFO: Got response 200" in stderr
assert "{'id': 1}" in out
assert "{'id': 2}" in out
@inlineCallbacks
def test_async_def_asyncio_parse_items_single_element(self):
status, out, stderr = yield self.execute(
[
"--spider",
"asyncdef_asyncio_return_single_element",
"-c",
"parse",
self.url("/html"),
]
_, out, stderr = self.proc(
"parse",
"--spider",
"asyncdef_asyncio_return_single_element",
"-c",
"parse",
self.mockserver.url("/html"),
)
assert "INFO: Got response 200" in _textmode(stderr)
assert "{'foo': 42}" in _textmode(out)
assert "INFO: Got response 200" in stderr
assert "{'foo': 42}" in out
@inlineCallbacks
def test_async_def_asyncgen_parse_loop(self):
status, out, stderr = yield self.execute(
[
"--spider",
"asyncdef_asyncio_gen_loop",
"-c",
"parse",
self.url("/html"),
]
_, out, stderr = self.proc(
"parse",
"--spider",
"asyncdef_asyncio_gen_loop",
"-c",
"parse",
self.mockserver.url("/html"),
)
assert "INFO: Got response 200" in _textmode(stderr)
assert "INFO: Got response 200" in stderr
for i in range(10):
assert f"{{'foo': {i}}}" in _textmode(out)
assert f"{{'foo': {i}}}" in out
@inlineCallbacks
def test_async_def_asyncgen_parse_exc(self):
status, out, stderr = yield self.execute(
[
"--spider",
"asyncdef_asyncio_gen_exc",
"-c",
"parse",
self.url("/html"),
]
_, out, stderr = self.proc(
"parse",
"--spider",
"asyncdef_asyncio_gen_exc",
"-c",
"parse",
self.mockserver.url("/html"),
)
assert "ValueError" in _textmode(stderr)
assert "ValueError" in stderr
for i in range(7):
assert f"{{'foo': {i}}}" in _textmode(out)
assert f"{{'foo': {i}}}" in out
@inlineCallbacks
def test_async_def_asyncio_parse(self):
_, _, stderr = yield self.execute(
[
"--spider",
"asyncdef_asyncio",
"-c",
"parse",
self.url("/html"),
]
_, _, stderr = self.proc(
"parse",
"--spider",
"asyncdef_asyncio",
"-c",
"parse",
self.mockserver.url("/html"),
)
assert "DEBUG: Got response 200" in _textmode(stderr)
assert "DEBUG: Got response 200" in stderr
@inlineCallbacks
def test_parse_items(self):
status, out, stderr = yield self.execute(
["--spider", self.spider_name, "-c", "parse", self.url("/html")]
_, out, _ = self.proc(
"parse",
"--spider",
self.spider_name,
"-c",
"parse",
self.mockserver.url("/html"),
)
assert "[{}, {'foo': 'bar'}]" in _textmode(out)
assert "[{}, {'foo': 'bar'}]" in out
@inlineCallbacks
def test_parse_items_no_callback_passed(self):
status, out, stderr = yield self.execute(
["--spider", self.spider_name, self.url("/html")]
_, out, _ = self.proc(
"parse", "--spider", self.spider_name, self.mockserver.url("/html")
)
assert "[{}, {'foo': 'bar'}]" in _textmode(out)
assert "[{}, {'foo': 'bar'}]" in out
@inlineCallbacks
def test_wrong_callback_passed(self):
status, out, stderr = yield self.execute(
["--spider", self.spider_name, "-c", "dummy", self.url("/html")]
_, out, stderr = self.proc(
"parse",
"--spider",
self.spider_name,
"-c",
"dummy",
self.mockserver.url("/html"),
)
assert re.search(r"# Scraped Items -+\n\[\]", _textmode(out))
assert "Cannot find callback" in _textmode(stderr)
assert re.search(r"# Scraped Items -+\r?\n\[\]", out)
assert "Cannot find callback" in stderr
@inlineCallbacks
def test_crawlspider_matching_rule_callback_set(self):
"""If a rule matches the URL, use it's defined callback."""
status, out, stderr = yield self.execute(
["--spider", "goodcrawl" + self.spider_name, "-r", self.url("/html")]
_, out, _ = self.proc(
"parse",
"--spider",
"goodcrawl" + self.spider_name,
"-r",
self.mockserver.url("/html"),
)
assert "[{}, {'foo': 'bar'}]" in _textmode(out)
assert "[{}, {'foo': 'bar'}]" in out
@inlineCallbacks
def test_crawlspider_matching_rule_default_callback(self):
"""If a rule match but it has no callback set, use the 'parse' callback."""
status, out, stderr = yield self.execute(
["--spider", "goodcrawl" + self.spider_name, "-r", self.url("/text")]
_, out, _ = self.proc(
"parse",
"--spider",
"goodcrawl" + self.spider_name,
"-r",
self.mockserver.url("/text"),
)
assert "[{}, {'nomatch': 'default'}]" in _textmode(out)
assert "[{}, {'nomatch': 'default'}]" in out
@inlineCallbacks
def test_spider_with_no_rules_attribute(self):
"""Using -r with a spider with no rule should not produce items."""
status, out, stderr = yield self.execute(
["--spider", self.spider_name, "-r", self.url("/html")]
_, out, stderr = self.proc(
"parse", "--spider", self.spider_name, "-r", self.mockserver.url("/html")
)
assert re.search(r"# Scraped Items -+\n\[\]", _textmode(out))
assert "No CrawlSpider rules found" in _textmode(stderr)
assert re.search(r"# Scraped Items -+\r?\n\[\]", out)
assert "No CrawlSpider rules found" in stderr
@inlineCallbacks
def test_crawlspider_missing_callback(self):
status, out, stderr = yield self.execute(
["--spider", "badcrawl" + self.spider_name, "-r", self.url("/html")]
_, out, _ = self.proc(
"parse",
"--spider",
"badcrawl" + self.spider_name,
"-r",
self.mockserver.url("/html"),
)
assert re.search(r"# Scraped Items -+\n\[\]", _textmode(out))
assert re.search(r"# Scraped Items -+\r?\n\[\]", out)
@inlineCallbacks
def test_crawlspider_no_matching_rule(self):
"""The requested URL has no matching rule, so no items should be scraped"""
status, out, stderr = yield self.execute(
["--spider", "badcrawl" + self.spider_name, "-r", self.url("/enc-gb18030")]
_, out, stderr = self.proc(
"parse",
"--spider",
"badcrawl" + self.spider_name,
"-r",
self.mockserver.url("/enc-gb18030"),
)
assert re.search(r"# Scraped Items -+\n\[\]", _textmode(out))
assert "Cannot find a rule that matches" in _textmode(stderr)
assert re.search(r"# Scraped Items -+\r?\n\[\]", out)
assert "Cannot find a rule that matches" in stderr
@inlineCallbacks
def test_crawlspider_not_exists_with_not_matched_url(self):
status, out, stderr = yield self.execute([self.url("/invalid_url")])
assert status == 0
assert self.call("parse", self.mockserver.url("/invalid_url")) == 0
@inlineCallbacks
def test_output_flag(self):
"""Checks if a file was created successfully having
correct format containing correct data in it.
"""
file_name = "data.json"
file_path = Path(self.proj_path, file_name)
yield self.execute(
[
"--spider",
self.spider_name,
"-c",
"parse",
"-o",
file_name,
self.url("/html"),
]
self.proc(
"parse",
"--spider",
self.spider_name,
"-c",
"parse",
"-o",
file_name,
self.mockserver.url("/html"),
)
assert file_path.exists()

17
tests/test_command_runspider.py
View File

@ -7,7 +7,6 @@ from contextlib import contextmanager
from pathlib import Path
from tempfile import TemporaryDirectory, mkdtemp
from typing import TYPE_CHECKING
from unittest import skipIf
import pytest
@ -58,7 +57,7 @@ class BadSpider(scrapy.Spider):
return self.proc("runspider", fname, *args)
def get_log(self, code, name=None, args=()):
p, stdout, stderr = self.runspider(code, name, args=args)
_, _, stderr = self.runspider(code, name, args=args)
return stderr
def test_runspider(self):
@ -288,7 +287,7 @@ class MySpider(scrapy.Spider):
log = self.get_log(spider_code, args=args)
assert "[myspider] DEBUG: FEEDS: {'stdout:': {'format': 'json'}}" in log
@skipIf(platform.system() == "Windows", reason="Linux only")
@pytest.mark.skipif(platform.system() == "Windows", reason="Linux only")
def test_absolute_path_linux(self):
spider_code = """
import scrapy
@ -317,7 +316,7 @@ class MySpider(scrapy.Spider):
in log
)
@skipIf(platform.system() != "Windows", reason="Windows only")
@pytest.mark.skipif(platform.system() != "Windows", reason="Windows only")
def test_absolute_path_windows(self):
spider_code = """
import scrapy
@ -370,18 +369,16 @@ class MySpider(scrapy.Spider):
assert "The value of FOO is 42" in log
@pytest.mark.skipif(
platform.system() != "Windows", reason="Windows required for .pyw files"
)
class TestWindowsRunSpiderCommand(TestRunSpiderCommand):
spider_filename = "myspider.pyw"
def setUp(self):
if platform.system() != "Windows":
pytest.skip("Windows required for .pyw files")
return super().setUp()
def test_start_errors(self):
log = self.get_log(self.badspider, name="badspider.pyw")
assert "start" in log
assert "badspider.pyw" in log
def test_runspider_unable_to_load(self):
pytest.skip("Already Tested in 'RunSpiderCommandTest' ")
pytest.skip("Already Tested in 'RunSpiderCommandTest'")

153
tests/test_command_shell.py
View File

@ -5,140 +5,137 @@ from pathlib import Path
import pytest
from pexpect.popen_spawn import PopenSpawn
from twisted.internet.defer import inlineCallbacks
from twisted.trial import unittest
from scrapy.utils.reactor import _asyncio_reactor_path
from tests import NON_EXISTING_RESOLVABLE, tests_datadir
from tests.mockserver import MockServer
from tests.utils.testproc import ProcessTest
from tests.utils.testsite import SiteTest
from tests.test_commands import TestProjectBase
class TestShellCommand(ProcessTest, SiteTest, unittest.TestCase):
command = "shell"
class TestShellCommand(TestProjectBase):
@classmethod
def setup_class(cls):
cls.mockserver = MockServer()
cls.mockserver.__enter__()
@classmethod
def teardown_class(cls):
cls.mockserver.__exit__(None, None, None)
@inlineCallbacks
def test_empty(self):
_, out, _ = yield self.execute(["-c", "item"])
assert b"{}" in out
_, out, _ = self.proc("shell", "-c", "item")
assert "{}" in out
@inlineCallbacks
def test_response_body(self):
_, out, _ = yield self.execute([self.url("/text"), "-c", "response.body"])
assert b"Works" in out
_, out, _ = self.proc(
"shell", self.mockserver.url("/text"), "-c", "response.body"
)
assert "Works" in out
@inlineCallbacks
def test_response_type_text(self):
_, out, _ = yield self.execute([self.url("/text"), "-c", "type(response)"])
assert b"TextResponse" in out
_, out, _ = self.proc(
"shell", self.mockserver.url("/text"), "-c", "type(response)"
)
assert "TextResponse" in out
@inlineCallbacks
def test_response_type_html(self):
_, out, _ = yield self.execute([self.url("/html"), "-c", "type(response)"])
assert b"HtmlResponse" in out
_, out, _ = self.proc(
"shell", self.mockserver.url("/html"), "-c", "type(response)"
)
assert "HtmlResponse" in out
@inlineCallbacks
def test_response_selector_html(self):
xpath = "response.xpath(\"//p[@class='one']/text()\").get()"
_, out, _ = yield self.execute([self.url("/html"), "-c", xpath])
assert out.strip() == b"Works"
_, out, _ = self.proc("shell", self.mockserver.url("/html"), "-c", xpath)
assert out.strip() == "Works"
@inlineCallbacks
def test_response_encoding_gb18030(self):
_, out, _ = yield self.execute(
[self.url("/enc-gb18030"), "-c", "response.encoding"]
_, out, _ = self.proc(
"shell", self.mockserver.url("/enc-gb18030"), "-c", "response.encoding"
)
assert out.strip() == b"gb18030"
assert out.strip() == "gb18030"
@inlineCallbacks
def test_redirect(self):
_, out, _ = yield self.execute([self.url("/redirect"), "-c", "response.url"])
assert out.strip().endswith(b"/redirected")
_, out, _ = self.proc(
"shell", self.mockserver.url("/redirect"), "-c", "response.url"
)
assert out.strip().endswith("/redirected")
@inlineCallbacks
def test_redirect_follow_302(self):
_, out, _ = yield self.execute(
[self.url("/redirect-no-meta-refresh"), "-c", "response.status"]
_, out, _ = self.proc(
"shell",
self.mockserver.url("/redirect-no-meta-refresh"),
"-c",
"response.status",
)
assert out.strip().endswith(b"200")
assert out.strip().endswith("200")
@inlineCallbacks
def test_redirect_not_follow_302(self):
_, out, _ = yield self.execute(
[
"--no-redirect",
self.url("/redirect-no-meta-refresh"),
"-c",
"response.status",
]
_, out, _ = self.proc(
"shell",
"--no-redirect",
self.mockserver.url("/redirect-no-meta-refresh"),
"-c",
"response.status",
)
assert out.strip().endswith(b"302")
assert out.strip().endswith("302")
@inlineCallbacks
def test_fetch_redirect_follow_302(self):
"""Test that calling ``fetch(url)`` follows HTTP redirects by default."""
url = self.url("/redirect-no-meta-refresh")
url = self.mockserver.url("/redirect-no-meta-refresh")
code = f"fetch('{url}')"
errcode, out, errout = yield self.execute(["-c", code])
assert errcode == 0, out
assert b"Redirecting (302)" in errout
assert b"Crawled (200)" in errout
p, out, errout = self.proc("shell", "-c", code)
assert p.returncode == 0, out
assert "Redirecting (302)" in errout
assert "Crawled (200)" in errout
@inlineCallbacks
def test_fetch_redirect_not_follow_302(self):
"""Test that calling ``fetch(url, redirect=False)`` disables automatic redirects."""
url = self.url("/redirect-no-meta-refresh")
url = self.mockserver.url("/redirect-no-meta-refresh")
code = f"fetch('{url}', redirect=False)"
errcode, out, errout = yield self.execute(["-c", code])
assert errcode == 0, out
assert b"Crawled (302)" in errout
p, out, errout = self.proc("shell", "-c", code)
assert p.returncode == 0, out
assert "Crawled (302)" in errout
@inlineCallbacks
def test_request_replace(self):
url = self.url("/text")
url = self.mockserver.url("/text")
code = f"fetch('{url}') or fetch(response.request.replace(method='POST'))"
errcode, out, _ = yield self.execute(["-c", code])
assert errcode == 0, out
p, out, _ = self.proc("shell", "-c", code)
assert p.returncode == 0, out
@inlineCallbacks
def test_scrapy_import(self):
url = self.url("/text")
url = self.mockserver.url("/text")
code = f"fetch(scrapy.Request('{url}'))"
errcode, out, _ = yield self.execute(["-c", code])
assert errcode == 0, out
p, out, _ = self.proc("shell", "-c", code)
assert p.returncode == 0, out
@inlineCallbacks
def test_local_file(self):
filepath = Path(tests_datadir, "test_site", "index.html")
_, out, _ = yield self.execute([str(filepath), "-c", "item"])
assert b"{}" in out
_, out, _ = self.proc("shell", str(filepath), "-c", "item")
assert "{}" in out
@inlineCallbacks
def test_local_nofile(self):
filepath = "file:///tests/sample_data/test_site/nothinghere.html"
errcode, out, err = yield self.execute(
[filepath, "-c", "item"], check_code=False
)
assert errcode == 1, out or err
assert b"No such file or directory" in err
p, out, err = self.proc("shell", filepath, "-c", "item")
assert p.returncode == 1, out or err
assert "No such file or directory" in err
@inlineCallbacks
def test_dns_failures(self):
if NON_EXISTING_RESOLVABLE:
pytest.skip("Non-existing hosts are resolvable")
url = "www.somedomainthatdoesntexi.st"
errcode, out, err = yield self.execute([url, "-c", "item"], check_code=False)
assert errcode == 1, out or err
assert b"DNS lookup failed" in err
p, out, err = self.proc("shell", url, "-c", "item")
assert p.returncode == 1, out or err
assert "DNS lookup failed" in err
@inlineCallbacks
def test_shell_fetch_async(self):
url = self.url("/html")
url = self.mockserver.url("/html")
code = f"fetch('{url}')"
args = ["-c", code, "--set", f"TWISTED_REACTOR={_asyncio_reactor_path}"]
_, _, err = yield self.execute(args, check_code=True)
assert b"RuntimeError: There is no current event loop in thread" not in err
p, _, err = self.proc(
"shell", "-c", code, "--set", f"TWISTED_REACTOR={_asyncio_reactor_path}"
)
assert p.returncode == 0, err
assert "RuntimeError: There is no current event loop in thread" not in err
class TestInteractiveShell:

4
tests/test_command_startproject.py
View File

@ -108,8 +108,8 @@ def get_permissions_dict(
class TestStartprojectTemplates(TestProjectBase):
maxDiff = None
def setUp(self):
super().setUp()
def setup_method(self):
super().setup_method()
self.tmpl = str(Path(self.temp_path, "templates"))
self.tmpl_proj = str(Path(self.tmpl, "project"))

26
tests/test_command_version.py
View File

@ -1,29 +1,15 @@
import sys
from twisted.internet.defer import inlineCallbacks
from twisted.trial import unittest
import scrapy
from tests.utils.testproc import ProcessTest
from tests.test_commands import TestProjectBase
class TestVersionCommand(ProcessTest, unittest.TestCase):
command = "version"
@inlineCallbacks
class TestVersionCommand(TestProjectBase):
def test_output(self):
encoding = sys.stdout.encoding or "utf-8"
_, out, _ = yield self.execute([])
assert out.strip().decode(encoding) == f"Scrapy {scrapy.__version__}"
_, out, _ = self.proc("version")
assert out.strip() == f"Scrapy {scrapy.__version__}"
@inlineCallbacks
def test_verbose_output(self):
encoding = sys.stdout.encoding or "utf-8"
_, out, _ = yield self.execute(["-v"])
headers = [
line.partition(":")[0].strip()
for line in out.strip().decode(encoding).splitlines()
]
_, out, _ = self.proc("version", "-v")
headers = [line.partition(":")[0].strip() for line in out.strip().splitlines()]
assert headers == [
"Scrapy",
"lxml",

49
tests/test_commands.py
View File

@ -10,11 +10,9 @@ from pathlib import Path
from shutil import rmtree
from tempfile import TemporaryFile, mkdtemp
from threading import Timer
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any
from unittest import mock
from twisted.trial import unittest
import scrapy
from scrapy.cmdline import _pop_command_name, _print_unknown_command_msg
from scrapy.commands import ScrapyCommand, ScrapyHelpFormatter, view
@ -61,28 +59,30 @@ class TestCommandSettings:
)
class TestProjectBase(unittest.TestCase):
class TestProjectBase:
project_name = "testproject"
def setUp(self):
def setup_method(self):
self.temp_path = mkdtemp()
self.cwd = self.temp_path
self.proj_path = Path(self.temp_path, self.project_name)
self.proj_mod_path = self.proj_path / self.project_name
self.env = get_testenv()
def tearDown(self):
def teardown_method(self):
rmtree(self.temp_path)
def call(self, *new_args, **kwargs):
def call(self, *args: str, **popen_kwargs: Any) -> int:
with TemporaryFile() as out:
args = (sys.executable, "-m", "scrapy.cmdline", *new_args)
args = (sys.executable, "-m", "scrapy.cmdline", *args)
return subprocess.call(
args, stdout=out, stderr=out, cwd=self.cwd, env=self.env, **kwargs
args, stdout=out, stderr=out, cwd=self.cwd, env=self.env, **popen_kwargs
)
def proc(self, *new_args, **popen_kwargs):
args = (sys.executable, "-m", "scrapy.cmdline", *new_args)
def proc(
self, *args: str, **popen_kwargs: Any
) -> tuple[subprocess.Popen[bytes], str, str]:
args = (sys.executable, "-m", "scrapy.cmdline", *args)
p = subprocess.Popen(
args,
cwd=popen_kwargs.pop("cwd", self.cwd),
@ -118,10 +118,10 @@ class TestProjectBase(unittest.TestCase):
class TestCommandBase(TestProjectBase):
def setUp(self):
super().setUp()
def setup_method(self):
super().setup_method()
self.call("startproject", self.project_name)
self.cwd = Path(self.temp_path, self.project_name)
self.cwd = self.proj_path
self.env["SCRAPY_SETTINGS_MODULE"] = f"{self.project_name}.settings"
@ -136,8 +136,8 @@ class TestCommandCrawlerProcess(TestCommandBase):
"Type of self.crawler_process: <class 'scrapy.crawler.AsyncCrawlerProcess'>"
)
def setUp(self):
super().setUp()
def setup_method(self):
super().setup_method()
(self.cwd / self.project_name / "commands").mkdir(exist_ok=True)
(self.cwd / self.project_name / "commands" / "__init__.py").touch()
(self.cwd / self.project_name / "commands" / f"{self.name}.py").write_text("""
@ -363,6 +363,19 @@ Unknown command: abc
assert out.getvalue().strip() == message.strip()
class TestProjectSubdir(TestProjectBase):
"""Test that commands work in a subdirectory of the project."""
def setup_method(self):
super().setup_method()
self.call("startproject", self.project_name)
self.cwd = self.proj_path / "subdir"
self.cwd.mkdir(exist_ok=True)
def test_list(self):
assert self.call("list") == 0
class TestBenchCommand(TestCommandBase):
def test_run(self):
_, _, log = self.proc(
@ -389,8 +402,8 @@ class TestViewCommand(TestCommandBase):
class TestHelpMessage(TestCommandBase):
def setUp(self):
super().setUp()
def setup_method(self):
super().setup_method()
self.commands = [
"parse",
"startproject",

67
tests/utils/testproc.py
View File

@ -1,67 +0,0 @@
from __future__ import annotations
import os
import sys
from typing import TYPE_CHECKING, cast
from twisted.internet.defer import Deferred
from twisted.internet.error import ProcessTerminated
from twisted.internet.protocol import ProcessProtocol
if TYPE_CHECKING:
from collections.abc import Iterable
from twisted.python.failure import Failure
class ProcessTest:
command: str | None = None
prefix = [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)

47
tests/utils/testsite.py
View File

@ -1,47 +0,0 @@
from urllib.parse import urljoin
from twisted.web import resource, server, static, util
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)

1
tox.ini
View File

@ -19,7 +19,6 @@ deps =
pytest-xdist
sybil >= 1.3.0 # https://github.com/cjw296/sybil/issues/20#issuecomment-605433422
testfixtures
pywin32; sys_platform == "win32"
Twisted < 25.5.0 # https://github.com/twisted/twisted/issues/12467
[testenv]