mirror of https://github.com/scrapy/scrapy.git
Split tests/test_commands.py. (#6836)
This commit is contained in:
Andrey Rakhmatullin
GitHub
parent
e0b9f2d8f6
commit
a724541a71
|
|
@ -0,0 +1,93 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from tests.test_commands import TestCommandBase
|
||||
|
||||
|
||||
class TestCrawlCommand(TestCommandBase):
|
||||
def crawl(self, code, args=()):
|
||||
Path(self.proj_mod_path, "spiders", "myspider.py").write_text(
|
||||
code, encoding="utf-8"
|
||||
)
|
||||
return self.proc("crawl", "myspider", *args)
|
||||
|
||||
def get_log(self, code, args=()):
|
||||
_, _, stderr = self.crawl(code, args=args)
|
||||
return stderr
|
||||
|
||||
def test_no_output(self):
|
||||
spider_code = """
|
||||
import scrapy
|
||||
|
||||
class MySpider(scrapy.Spider):
|
||||
name = 'myspider'
|
||||
|
||||
async def start(self):
|
||||
self.logger.debug('It works!')
|
||||
return
|
||||
yield
|
||||
"""
|
||||
log = self.get_log(spider_code)
|
||||
assert "[myspider] DEBUG: It works!" in log
|
||||
|
||||
def test_output(self):
|
||||
spider_code = """
|
||||
import scrapy
|
||||
|
||||
class MySpider(scrapy.Spider):
|
||||
name = 'myspider'
|
||||
|
||||
async def start(self):
|
||||
self.logger.debug('FEEDS: {}'.format(self.settings.getdict('FEEDS')))
|
||||
return
|
||||
yield
|
||||
"""
|
||||
args = ["-o", "example.json"]
|
||||
log = self.get_log(spider_code, args=args)
|
||||
assert "[myspider] DEBUG: FEEDS: {'example.json': {'format': 'json'}}" in log
|
||||
|
||||
def test_overwrite_output(self):
|
||||
spider_code = """
|
||||
import json
|
||||
import scrapy
|
||||
|
||||
class MySpider(scrapy.Spider):
|
||||
name = 'myspider'
|
||||
|
||||
async def start(self):
|
||||
self.logger.debug(
|
||||
'FEEDS: {}'.format(
|
||||
json.dumps(self.settings.getdict('FEEDS'), sort_keys=True)
|
||||
)
|
||||
)
|
||||
return
|
||||
yield
|
||||
"""
|
||||
Path(self.cwd, "example.json").write_text("not empty", encoding="utf-8")
|
||||
args = ["-O", "example.json"]
|
||||
log = self.get_log(spider_code, args=args)
|
||||
assert (
|
||||
'[myspider] DEBUG: FEEDS: {"example.json": {"format": "json", "overwrite": true}}'
|
||||
in log
|
||||
)
|
||||
with Path(self.cwd, "example.json").open(encoding="utf-8") as f2:
|
||||
first_line = f2.readline()
|
||||
assert first_line != "not empty"
|
||||
|
||||
def test_output_and_overwrite_output(self):
|
||||
spider_code = """
|
||||
import scrapy
|
||||
|
||||
class MySpider(scrapy.Spider):
|
||||
name = 'myspider'
|
||||
|
||||
async def start(self):
|
||||
return
|
||||
yield
|
||||
"""
|
||||
args = ["-o", "example1.json", "-O", "example2.json"]
|
||||
log = self.get_log(spider_code, args=args)
|
||||
assert (
|
||||
"error: Please use only one of -o/--output and -O/--overwrite-output" in log
|
||||
)
|
||||
|
|
@ -0,0 +1,208 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from tests.test_commands import TestCommandBase, TestProjectBase
|
||||
|
||||
|
||||
class TestGenspiderCommand(TestCommandBase):
|
||||
def test_arguments(self):
|
||||
# only pass one argument. spider script shouldn't be created
|
||||
assert self.call("genspider", "test_name") == 2
|
||||
assert not Path(self.proj_mod_path, "spiders", "test_name.py").exists()
|
||||
# pass two arguments <name> <domain>. spider script should be created
|
||||
assert self.call("genspider", "test_name", "test.com") == 0
|
||||
assert Path(self.proj_mod_path, "spiders", "test_name.py").exists()
|
||||
|
||||
def test_template(self, tplname="crawl"):
|
||||
args = [f"--template={tplname}"] if tplname else []
|
||||
spname = "test_spider"
|
||||
spmodule = f"{self.project_name}.spiders.{spname}"
|
||||
p, out, err = self.proc("genspider", spname, "test.com", *args)
|
||||
assert (
|
||||
f"Created spider {spname!r} using template {tplname!r} in module:{os.linesep} {spmodule}"
|
||||
in out
|
||||
)
|
||||
assert Path(self.proj_mod_path, "spiders", "test_spider.py").exists()
|
||||
modify_time_before = (
|
||||
Path(self.proj_mod_path, "spiders", "test_spider.py").stat().st_mtime
|
||||
)
|
||||
p, out, err = self.proc("genspider", spname, "test.com", *args)
|
||||
assert f"Spider {spname!r} already exists in module" in out
|
||||
modify_time_after = (
|
||||
Path(self.proj_mod_path, "spiders", "test_spider.py").stat().st_mtime
|
||||
)
|
||||
assert modify_time_after == modify_time_before
|
||||
|
||||
def test_template_basic(self):
|
||||
self.test_template("basic")
|
||||
|
||||
def test_template_csvfeed(self):
|
||||
self.test_template("csvfeed")
|
||||
|
||||
def test_template_xmlfeed(self):
|
||||
self.test_template("xmlfeed")
|
||||
|
||||
def test_list(self):
|
||||
assert self.call("genspider", "--list") == 0
|
||||
|
||||
def test_dump(self):
|
||||
assert self.call("genspider", "--dump=basic") == 0
|
||||
assert self.call("genspider", "-d", "basic") == 0
|
||||
|
||||
def test_same_name_as_project(self):
|
||||
assert self.call("genspider", self.project_name) == 2
|
||||
assert not Path(
|
||||
self.proj_mod_path, "spiders", f"{self.project_name}.py"
|
||||
).exists()
|
||||
|
||||
def test_same_filename_as_existing_spider(self, force=False):
|
||||
file_name = "example"
|
||||
file_path = Path(self.proj_mod_path, "spiders", f"{file_name}.py")
|
||||
assert self.call("genspider", file_name, "example.com") == 0
|
||||
assert file_path.exists()
|
||||
|
||||
# change name of spider but not its file name
|
||||
with file_path.open("r+", encoding="utf-8") as spider_file:
|
||||
file_data = spider_file.read()
|
||||
file_data = file_data.replace('name = "example"', 'name = "renamed"')
|
||||
spider_file.seek(0)
|
||||
spider_file.write(file_data)
|
||||
spider_file.truncate()
|
||||
modify_time_before = file_path.stat().st_mtime
|
||||
file_contents_before = file_data
|
||||
|
||||
if force:
|
||||
p, out, err = self.proc("genspider", "--force", file_name, "example.com")
|
||||
assert (
|
||||
f"Created spider {file_name!r} using template 'basic' in module" in out
|
||||
)
|
||||
modify_time_after = file_path.stat().st_mtime
|
||||
assert modify_time_after != modify_time_before
|
||||
file_contents_after = file_path.read_text(encoding="utf-8")
|
||||
assert file_contents_after != file_contents_before
|
||||
else:
|
||||
p, out, err = self.proc("genspider", file_name, "example.com")
|
||||
assert f"{file_path.resolve()} already exists" in out
|
||||
modify_time_after = file_path.stat().st_mtime
|
||||
assert modify_time_after == modify_time_before
|
||||
file_contents_after = file_path.read_text(encoding="utf-8")
|
||||
assert file_contents_after == file_contents_before
|
||||
|
||||
def test_same_filename_as_existing_spider_force(self):
|
||||
self.test_same_filename_as_existing_spider(force=True)
|
||||
|
||||
def test_url(self, url="test.com", domain="test.com"):
|
||||
assert self.call("genspider", "--force", "test_name", url) == 0
|
||||
assert (
|
||||
self.find_in_file(
|
||||
Path(self.proj_mod_path, "spiders", "test_name.py"),
|
||||
r"allowed_domains\s*=\s*\[['\"](.+)['\"]\]",
|
||||
).group(1)
|
||||
== domain
|
||||
)
|
||||
assert (
|
||||
self.find_in_file(
|
||||
Path(self.proj_mod_path, "spiders", "test_name.py"),
|
||||
r"start_urls\s*=\s*\[['\"](.+)['\"]\]",
|
||||
).group(1)
|
||||
== f"https://{domain}"
|
||||
)
|
||||
|
||||
def test_url_schema(self):
|
||||
self.test_url("https://test.com", "test.com")
|
||||
|
||||
def test_template_start_urls(
|
||||
self, url="test.com", expected="https://test.com", template="basic"
|
||||
):
|
||||
assert self.call("genspider", "-t", template, "--force", "test_name", url) == 0
|
||||
assert (
|
||||
self.find_in_file(
|
||||
Path(self.proj_mod_path, "spiders", "test_name.py"),
|
||||
r"start_urls\s*=\s*\[['\"](.+)['\"]\]",
|
||||
).group(1)
|
||||
== expected
|
||||
)
|
||||
|
||||
def test_genspider_basic_start_urls(self):
|
||||
self.test_template_start_urls("https://test.com", "https://test.com", "basic")
|
||||
self.test_template_start_urls("http://test.com", "http://test.com", "basic")
|
||||
self.test_template_start_urls(
|
||||
"http://test.com/other/path", "http://test.com/other/path", "basic"
|
||||
)
|
||||
self.test_template_start_urls(
|
||||
"test.com/other/path", "https://test.com/other/path", "basic"
|
||||
)
|
||||
|
||||
def test_genspider_crawl_start_urls(self):
|
||||
self.test_template_start_urls("https://test.com", "https://test.com", "crawl")
|
||||
self.test_template_start_urls("http://test.com", "http://test.com", "crawl")
|
||||
self.test_template_start_urls(
|
||||
"http://test.com/other/path", "http://test.com/other/path", "crawl"
|
||||
)
|
||||
self.test_template_start_urls(
|
||||
"test.com/other/path", "https://test.com/other/path", "crawl"
|
||||
)
|
||||
self.test_template_start_urls("test.com", "https://test.com", "crawl")
|
||||
|
||||
def test_genspider_xmlfeed_start_urls(self):
|
||||
self.test_template_start_urls(
|
||||
"https://test.com/feed.xml", "https://test.com/feed.xml", "xmlfeed"
|
||||
)
|
||||
self.test_template_start_urls(
|
||||
"http://test.com/feed.xml", "http://test.com/feed.xml", "xmlfeed"
|
||||
)
|
||||
self.test_template_start_urls(
|
||||
"test.com/feed.xml", "https://test.com/feed.xml", "xmlfeed"
|
||||
)
|
||||
|
||||
def test_genspider_csvfeed_start_urls(self):
|
||||
self.test_template_start_urls(
|
||||
"https://test.com/feed.csv", "https://test.com/feed.csv", "csvfeed"
|
||||
)
|
||||
self.test_template_start_urls(
|
||||
"http://test.com/feed.xml", "http://test.com/feed.xml", "csvfeed"
|
||||
)
|
||||
self.test_template_start_urls(
|
||||
"test.com/feed.csv", "https://test.com/feed.csv", "csvfeed"
|
||||
)
|
||||
|
||||
|
||||
class TestGenspiderStandaloneCommand(TestProjectBase):
|
||||
def test_generate_standalone_spider(self):
|
||||
self.call("genspider", "example", "example.com")
|
||||
assert Path(self.temp_path, "example.py").exists()
|
||||
|
||||
def test_same_name_as_existing_file(self, force=False):
|
||||
file_name = "example"
|
||||
file_path = Path(self.temp_path, file_name + ".py")
|
||||
p, out, err = self.proc("genspider", file_name, "example.com")
|
||||
assert f"Created spider {file_name!r} using template 'basic' " in out
|
||||
assert file_path.exists()
|
||||
modify_time_before = file_path.stat().st_mtime
|
||||
file_contents_before = file_path.read_text(encoding="utf-8")
|
||||
|
||||
if force:
|
||||
# use different template to ensure contents were changed
|
||||
p, out, err = self.proc(
|
||||
"genspider", "--force", "-t", "crawl", file_name, "example.com"
|
||||
)
|
||||
assert f"Created spider {file_name!r} using template 'crawl' " in out
|
||||
modify_time_after = file_path.stat().st_mtime
|
||||
assert modify_time_after != modify_time_before
|
||||
file_contents_after = file_path.read_text(encoding="utf-8")
|
||||
assert file_contents_after != file_contents_before
|
||||
else:
|
||||
p, out, err = self.proc("genspider", file_name, "example.com")
|
||||
assert (
|
||||
f"{Path(self.temp_path, file_name + '.py').resolve()} already exists"
|
||||
in out
|
||||
)
|
||||
modify_time_after = file_path.stat().st_mtime
|
||||
assert modify_time_after == modify_time_before
|
||||
file_contents_after = file_path.read_text(encoding="utf-8")
|
||||
assert file_contents_after == file_contents_before
|
||||
|
||||
def test_same_name_as_existing_file_force(self):
|
||||
self.test_same_name_as_existing_file(force=True)
|
||||
|
|
@ -0,0 +1,375 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
import platform
|
||||
import sys
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory, mkdtemp
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest import skipIf
|
||||
|
||||
import pytest
|
||||
from twisted.trial import unittest
|
||||
|
||||
from tests.test_commands import TestCommandBase
|
||||
from tests.test_crawler import ExceptionSpider, NoRequestsSpider
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
|
||||
|
||||
class TestRunSpiderCommand(TestCommandBase):
|
||||
spider_filename = "myspider.py"
|
||||
|
||||
debug_log_spider = """
|
||||
import scrapy
|
||||
|
||||
class MySpider(scrapy.Spider):
|
||||
name = 'myspider'
|
||||
|
||||
async def start(self):
|
||||
self.logger.debug("It Works!")
|
||||
return
|
||||
yield
|
||||
"""
|
||||
|
||||
badspider = """
|
||||
import scrapy
|
||||
|
||||
class BadSpider(scrapy.Spider):
|
||||
name = "bad"
|
||||
async def start(self):
|
||||
raise Exception("oops!")
|
||||
yield
|
||||
"""
|
||||
|
||||
@contextmanager
|
||||
def _create_file(self, content: str, name: str | None = None) -> Iterator[str]:
|
||||
with TemporaryDirectory() as tmpdir:
|
||||
if name:
|
||||
fname = Path(tmpdir, name).resolve()
|
||||
else:
|
||||
fname = Path(tmpdir, self.spider_filename).resolve()
|
||||
fname.write_text(content, encoding="utf-8")
|
||||
yield str(fname)
|
||||
|
||||
def runspider(self, code, name=None, args=()):
|
||||
with self._create_file(code, name) as fname:
|
||||
return self.proc("runspider", fname, *args)
|
||||
|
||||
def get_log(self, code, name=None, args=()):
|
||||
p, stdout, stderr = self.runspider(code, name, args=args)
|
||||
return stderr
|
||||
|
||||
def test_runspider(self):
|
||||
log = self.get_log(self.debug_log_spider)
|
||||
assert "DEBUG: It Works!" in log
|
||||
assert "INFO: Spider opened" in log
|
||||
assert "INFO: Closing spider (finished)" in log
|
||||
assert "INFO: Spider closed (finished)" in log
|
||||
|
||||
def test_run_fail_spider(self):
|
||||
proc, _, _ = self.runspider(
|
||||
"import scrapy\n" + inspect.getsource(ExceptionSpider)
|
||||
)
|
||||
ret = proc.returncode
|
||||
assert ret != 0
|
||||
|
||||
def test_run_good_spider(self):
|
||||
proc, _, _ = self.runspider(
|
||||
"import scrapy\n" + inspect.getsource(NoRequestsSpider)
|
||||
)
|
||||
ret = proc.returncode
|
||||
assert ret == 0
|
||||
|
||||
def test_runspider_log_level(self):
|
||||
log = self.get_log(self.debug_log_spider, args=("-s", "LOG_LEVEL=INFO"))
|
||||
assert "DEBUG: It Works!" not in log
|
||||
assert "INFO: Spider opened" in log
|
||||
|
||||
def test_runspider_dnscache_disabled(self):
|
||||
# see https://github.com/scrapy/scrapy/issues/2811
|
||||
# The spider below should not be able to connect to localhost:12345,
|
||||
# which is intended,
|
||||
# but this should not be because of DNS lookup error
|
||||
# assumption: localhost will resolve in all cases (true?)
|
||||
dnscache_spider = """
|
||||
import scrapy
|
||||
|
||||
class MySpider(scrapy.Spider):
|
||||
name = 'myspider'
|
||||
start_urls = ['http://localhost:12345']
|
||||
|
||||
def parse(self, response):
|
||||
return {'test': 'value'}
|
||||
"""
|
||||
log = self.get_log(dnscache_spider, args=("-s", "DNSCACHE_ENABLED=False"))
|
||||
assert "DNSLookupError" not in log
|
||||
assert "INFO: Spider opened" in log
|
||||
|
||||
def test_runspider_log_short_names(self):
|
||||
log1 = self.get_log(self.debug_log_spider, args=("-s", "LOG_SHORT_NAMES=1"))
|
||||
assert "[myspider] DEBUG: It Works!" in log1
|
||||
assert "[scrapy]" in log1
|
||||
assert "[scrapy.core.engine]" not in log1
|
||||
|
||||
log2 = self.get_log(self.debug_log_spider, args=("-s", "LOG_SHORT_NAMES=0"))
|
||||
assert "[myspider] DEBUG: It Works!" in log2
|
||||
assert "[scrapy]" not in log2
|
||||
assert "[scrapy.core.engine]" in log2
|
||||
|
||||
def test_runspider_no_spider_found(self):
|
||||
log = self.get_log("from scrapy.spiders import Spider\n")
|
||||
assert "No spider found in file" in log
|
||||
|
||||
def test_runspider_file_not_found(self):
|
||||
_, _, log = self.proc("runspider", "some_non_existent_file")
|
||||
assert "File not found: some_non_existent_file" in log
|
||||
|
||||
def test_runspider_unable_to_load(self):
|
||||
log = self.get_log("", name="myspider.txt")
|
||||
assert "Unable to load" in log
|
||||
|
||||
def test_start_errors(self):
|
||||
log = self.get_log(self.badspider, name="badspider.py")
|
||||
assert "start" in log
|
||||
assert "badspider.py" in log, log
|
||||
|
||||
def test_asyncio_enabled_true(self):
|
||||
log = self.get_log(
|
||||
self.debug_log_spider,
|
||||
args=[
|
||||
"-s",
|
||||
"TWISTED_REACTOR=twisted.internet.asyncioreactor.AsyncioSelectorReactor",
|
||||
],
|
||||
)
|
||||
assert (
|
||||
"Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
|
||||
in log
|
||||
)
|
||||
|
||||
def test_asyncio_enabled_default(self):
|
||||
log = self.get_log(self.debug_log_spider, args=[])
|
||||
assert (
|
||||
"Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
|
||||
in log
|
||||
)
|
||||
|
||||
def test_asyncio_enabled_false(self):
|
||||
log = self.get_log(
|
||||
self.debug_log_spider,
|
||||
args=["-s", "TWISTED_REACTOR=twisted.internet.selectreactor.SelectReactor"],
|
||||
)
|
||||
assert "Using reactor: twisted.internet.selectreactor.SelectReactor" in log
|
||||
assert (
|
||||
"Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
|
||||
not in log
|
||||
)
|
||||
|
||||
@pytest.mark.requires_uvloop
|
||||
def test_custom_asyncio_loop_enabled_true(self):
|
||||
log = self.get_log(
|
||||
self.debug_log_spider,
|
||||
args=[
|
||||
"-s",
|
||||
"TWISTED_REACTOR=twisted.internet.asyncioreactor.AsyncioSelectorReactor",
|
||||
"-s",
|
||||
"ASYNCIO_EVENT_LOOP=uvloop.Loop",
|
||||
],
|
||||
)
|
||||
assert "Using asyncio event loop: uvloop.Loop" in log
|
||||
|
||||
def test_custom_asyncio_loop_enabled_false(self):
|
||||
log = self.get_log(
|
||||
self.debug_log_spider,
|
||||
args=[
|
||||
"-s",
|
||||
"TWISTED_REACTOR=twisted.internet.asyncioreactor.AsyncioSelectorReactor",
|
||||
],
|
||||
)
|
||||
import asyncio
|
||||
|
||||
if sys.platform != "win32":
|
||||
loop = asyncio.new_event_loop()
|
||||
else:
|
||||
loop = asyncio.SelectorEventLoop()
|
||||
assert (
|
||||
f"Using asyncio event loop: {loop.__module__}.{loop.__class__.__name__}"
|
||||
in log
|
||||
)
|
||||
|
||||
def test_output(self):
|
||||
spider_code = """
|
||||
import scrapy
|
||||
|
||||
class MySpider(scrapy.Spider):
|
||||
name = 'myspider'
|
||||
|
||||
async def start(self):
|
||||
self.logger.debug('FEEDS: {}'.format(self.settings.getdict('FEEDS')))
|
||||
return
|
||||
yield
|
||||
"""
|
||||
args = ["-o", "example.json"]
|
||||
log = self.get_log(spider_code, args=args)
|
||||
assert "[myspider] DEBUG: FEEDS: {'example.json': {'format': 'json'}}" in log
|
||||
|
||||
def test_overwrite_output(self):
|
||||
spider_code = """
|
||||
import json
|
||||
import scrapy
|
||||
|
||||
class MySpider(scrapy.Spider):
|
||||
name = 'myspider'
|
||||
|
||||
async def start(self):
|
||||
self.logger.debug(
|
||||
'FEEDS: {}'.format(
|
||||
json.dumps(self.settings.getdict('FEEDS'), sort_keys=True)
|
||||
)
|
||||
)
|
||||
return
|
||||
yield
|
||||
"""
|
||||
Path(self.cwd, "example.json").write_text("not empty", encoding="utf-8")
|
||||
args = ["-O", "example.json"]
|
||||
log = self.get_log(spider_code, args=args)
|
||||
assert (
|
||||
'[myspider] DEBUG: FEEDS: {"example.json": {"format": "json", "overwrite": true}}'
|
||||
in log
|
||||
)
|
||||
with Path(self.cwd, "example.json").open(encoding="utf-8") as f2:
|
||||
first_line = f2.readline()
|
||||
assert first_line != "not empty"
|
||||
|
||||
def test_output_and_overwrite_output(self):
|
||||
spider_code = """
|
||||
import scrapy
|
||||
|
||||
class MySpider(scrapy.Spider):
|
||||
name = 'myspider'
|
||||
|
||||
async def start(self):
|
||||
return
|
||||
yield
|
||||
"""
|
||||
args = ["-o", "example1.json", "-O", "example2.json"]
|
||||
log = self.get_log(spider_code, args=args)
|
||||
assert (
|
||||
"error: Please use only one of -o/--output and -O/--overwrite-output" in log
|
||||
)
|
||||
|
||||
def test_output_stdout(self):
|
||||
spider_code = """
|
||||
import scrapy
|
||||
|
||||
class MySpider(scrapy.Spider):
|
||||
name = 'myspider'
|
||||
|
||||
async def start(self):
|
||||
self.logger.debug('FEEDS: {}'.format(self.settings.getdict('FEEDS')))
|
||||
return
|
||||
yield
|
||||
"""
|
||||
args = ["-o", "-:json"]
|
||||
log = self.get_log(spider_code, args=args)
|
||||
assert "[myspider] DEBUG: FEEDS: {'stdout:': {'format': 'json'}}" in log
|
||||
|
||||
@skipIf(platform.system() == "Windows", reason="Linux only")
|
||||
def test_absolute_path_linux(self):
|
||||
spider_code = """
|
||||
import scrapy
|
||||
|
||||
class MySpider(scrapy.Spider):
|
||||
name = 'myspider'
|
||||
|
||||
start_urls = ["data:,"]
|
||||
|
||||
def parse(self, response):
|
||||
yield {"hello": "world"}
|
||||
"""
|
||||
temp_dir = mkdtemp()
|
||||
|
||||
args = ["-o", f"{temp_dir}/output1.json:json"]
|
||||
log = self.get_log(spider_code, args=args)
|
||||
assert (
|
||||
f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {temp_dir}/output1.json"
|
||||
in log
|
||||
)
|
||||
|
||||
args = ["-o", f"{temp_dir}/output2.json"]
|
||||
log = self.get_log(spider_code, args=args)
|
||||
assert (
|
||||
f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {temp_dir}/output2.json"
|
||||
in log
|
||||
)
|
||||
|
||||
@skipIf(platform.system() != "Windows", reason="Windows only")
|
||||
def test_absolute_path_windows(self):
|
||||
spider_code = """
|
||||
import scrapy
|
||||
|
||||
class MySpider(scrapy.Spider):
|
||||
name = 'myspider'
|
||||
|
||||
start_urls = ["data:,"]
|
||||
|
||||
def parse(self, response):
|
||||
yield {"hello": "world"}
|
||||
"""
|
||||
temp_dir = mkdtemp()
|
||||
|
||||
args = ["-o", f"{temp_dir}\\output1.json:json"]
|
||||
log = self.get_log(spider_code, args=args)
|
||||
assert (
|
||||
f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {temp_dir}\\output1.json"
|
||||
in log
|
||||
)
|
||||
|
||||
args = ["-o", f"{temp_dir}\\output2.json"]
|
||||
log = self.get_log(spider_code, args=args)
|
||||
assert (
|
||||
f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {temp_dir}\\output2.json"
|
||||
in log
|
||||
)
|
||||
|
||||
def test_args_change_settings(self):
|
||||
spider_code = """
|
||||
import scrapy
|
||||
|
||||
class MySpider(scrapy.Spider):
|
||||
name = 'myspider'
|
||||
|
||||
@classmethod
|
||||
def from_crawler(cls, crawler, *args, **kwargs):
|
||||
spider = super().from_crawler(crawler, *args, **kwargs)
|
||||
spider.settings.set("FOO", kwargs.get("foo"))
|
||||
return spider
|
||||
|
||||
async def start(self):
|
||||
self.logger.info(f"The value of FOO is {self.settings.getint('FOO')}")
|
||||
return
|
||||
yield
|
||||
"""
|
||||
args = ["-a", "foo=42"]
|
||||
log = self.get_log(spider_code, args=args)
|
||||
assert "Spider closed (finished)" in log
|
||||
assert "The value of FOO is 42" in log
|
||||
|
||||
|
||||
class TestWindowsRunSpiderCommand(TestRunSpiderCommand):
|
||||
spider_filename = "myspider.pyw"
|
||||
|
||||
def setUp(self):
|
||||
if platform.system() != "Windows":
|
||||
raise unittest.SkipTest("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):
|
||||
raise unittest.SkipTest("Already Tested in 'RunSpiderCommandTest' ")
|
||||
|
|
@ -0,0 +1,318 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from contextlib import contextmanager
|
||||
from itertools import chain
|
||||
from pathlib import Path
|
||||
from shutil import copytree
|
||||
from stat import S_IWRITE as ANYONE_WRITE_PERMISSION
|
||||
from tempfile import mkdtemp
|
||||
|
||||
import scrapy
|
||||
from scrapy.commands.startproject import IGNORE
|
||||
from tests.test_commands import TestProjectBase
|
||||
|
||||
|
||||
class TestStartprojectCommand(TestProjectBase):
|
||||
def test_startproject(self):
|
||||
p, out, err = self.proc("startproject", self.project_name)
|
||||
print(out)
|
||||
print(err, file=sys.stderr)
|
||||
assert p.returncode == 0
|
||||
|
||||
assert Path(self.proj_path, "scrapy.cfg").exists()
|
||||
assert Path(self.proj_path, "testproject").exists()
|
||||
assert Path(self.proj_mod_path, "__init__.py").exists()
|
||||
assert Path(self.proj_mod_path, "items.py").exists()
|
||||
assert Path(self.proj_mod_path, "pipelines.py").exists()
|
||||
assert Path(self.proj_mod_path, "settings.py").exists()
|
||||
assert Path(self.proj_mod_path, "spiders", "__init__.py").exists()
|
||||
|
||||
assert self.call("startproject", self.project_name) == 1
|
||||
assert self.call("startproject", "wrong---project---name") == 1
|
||||
assert self.call("startproject", "sys") == 1
|
||||
|
||||
def test_startproject_with_project_dir(self):
|
||||
project_dir = mkdtemp()
|
||||
assert self.call("startproject", self.project_name, project_dir) == 0
|
||||
|
||||
assert Path(project_dir, "scrapy.cfg").exists()
|
||||
assert Path(project_dir, "testproject").exists()
|
||||
assert Path(project_dir, self.project_name, "__init__.py").exists()
|
||||
assert Path(project_dir, self.project_name, "items.py").exists()
|
||||
assert Path(project_dir, self.project_name, "pipelines.py").exists()
|
||||
assert Path(project_dir, self.project_name, "settings.py").exists()
|
||||
assert Path(project_dir, self.project_name, "spiders", "__init__.py").exists()
|
||||
|
||||
assert self.call("startproject", self.project_name, project_dir + "2") == 0
|
||||
|
||||
assert self.call("startproject", self.project_name, project_dir) == 1
|
||||
assert self.call("startproject", self.project_name + "2", project_dir) == 1
|
||||
assert self.call("startproject", "wrong---project---name") == 1
|
||||
assert self.call("startproject", "sys") == 1
|
||||
assert self.call("startproject") == 2
|
||||
assert (
|
||||
self.call("startproject", self.project_name, project_dir, "another_params")
|
||||
== 2
|
||||
)
|
||||
|
||||
def test_existing_project_dir(self):
|
||||
project_dir = mkdtemp()
|
||||
project_name = self.project_name + "_existing"
|
||||
project_path = Path(project_dir, project_name)
|
||||
project_path.mkdir()
|
||||
|
||||
p, out, err = self.proc("startproject", project_name, cwd=project_dir)
|
||||
print(out)
|
||||
print(err, file=sys.stderr)
|
||||
assert p.returncode == 0
|
||||
|
||||
assert Path(project_path, "scrapy.cfg").exists()
|
||||
assert Path(project_path, project_name).exists()
|
||||
assert Path(project_path, project_name, "__init__.py").exists()
|
||||
assert Path(project_path, project_name, "items.py").exists()
|
||||
assert Path(project_path, project_name, "pipelines.py").exists()
|
||||
assert Path(project_path, project_name, "settings.py").exists()
|
||||
assert Path(project_path, project_name, "spiders", "__init__.py").exists()
|
||||
|
||||
|
||||
def get_permissions_dict(
|
||||
path: str | os.PathLike, renamings=None, ignore=None
|
||||
) -> dict[str, str]:
|
||||
def get_permissions(path: Path) -> str:
|
||||
return oct(path.stat().st_mode)
|
||||
|
||||
path_obj = Path(path)
|
||||
|
||||
renamings = renamings or ()
|
||||
permissions_dict = {
|
||||
".": get_permissions(path_obj),
|
||||
}
|
||||
for root, dirs, files in os.walk(path_obj):
|
||||
nodes = list(chain(dirs, files))
|
||||
if ignore:
|
||||
ignored_names = ignore(root, nodes)
|
||||
nodes = [node for node in nodes if node not in ignored_names]
|
||||
for node in nodes:
|
||||
absolute_path = Path(root, node)
|
||||
relative_path = str(absolute_path.relative_to(path))
|
||||
for search_string, replacement in renamings:
|
||||
relative_path = relative_path.replace(search_string, replacement)
|
||||
permissions = get_permissions(absolute_path)
|
||||
permissions_dict[relative_path] = permissions
|
||||
return permissions_dict
|
||||
|
||||
|
||||
class TestStartprojectTemplates(TestProjectBase):
|
||||
maxDiff = None
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.tmpl = str(Path(self.temp_path, "templates"))
|
||||
self.tmpl_proj = str(Path(self.tmpl, "project"))
|
||||
|
||||
def test_startproject_template_override(self):
|
||||
copytree(Path(scrapy.__path__[0], "templates"), self.tmpl)
|
||||
Path(self.tmpl_proj, "root_template").write_bytes(b"")
|
||||
assert Path(self.tmpl_proj, "root_template").exists()
|
||||
|
||||
args = ["--set", f"TEMPLATES_DIR={self.tmpl}"]
|
||||
p, out, err = self.proc("startproject", self.project_name, *args)
|
||||
assert (
|
||||
f"New Scrapy project '{self.project_name}', using template directory" in out
|
||||
)
|
||||
assert self.tmpl_proj in out
|
||||
assert Path(self.proj_path, "root_template").exists()
|
||||
|
||||
def test_startproject_permissions_from_writable(self):
|
||||
"""Check that generated files have the right permissions when the
|
||||
template folder has the same permissions as in the project, i.e.
|
||||
everything is writable."""
|
||||
scrapy_path = scrapy.__path__[0]
|
||||
project_template = Path(scrapy_path, "templates", "project")
|
||||
project_name = "startproject1"
|
||||
renamings = (
|
||||
("module", project_name),
|
||||
(".tmpl", ""),
|
||||
)
|
||||
expected_permissions = get_permissions_dict(
|
||||
project_template,
|
||||
renamings,
|
||||
IGNORE,
|
||||
)
|
||||
|
||||
destination = mkdtemp()
|
||||
process = subprocess.Popen(
|
||||
(
|
||||
sys.executable,
|
||||
"-m",
|
||||
"scrapy.cmdline",
|
||||
"startproject",
|
||||
project_name,
|
||||
),
|
||||
cwd=destination,
|
||||
env=self.env,
|
||||
)
|
||||
process.wait()
|
||||
|
||||
project_dir = Path(destination, project_name)
|
||||
actual_permissions = get_permissions_dict(project_dir)
|
||||
|
||||
assert actual_permissions == expected_permissions
|
||||
|
||||
def test_startproject_permissions_from_read_only(self):
|
||||
"""Check that generated files have the right permissions when the
|
||||
template folder has been made read-only, which is something that some
|
||||
systems do.
|
||||
|
||||
See https://github.com/scrapy/scrapy/pull/4604
|
||||
"""
|
||||
scrapy_path = scrapy.__path__[0]
|
||||
templates_dir = Path(scrapy_path, "templates")
|
||||
project_template = Path(templates_dir, "project")
|
||||
project_name = "startproject2"
|
||||
renamings = (
|
||||
("module", project_name),
|
||||
(".tmpl", ""),
|
||||
)
|
||||
expected_permissions = get_permissions_dict(
|
||||
project_template,
|
||||
renamings,
|
||||
IGNORE,
|
||||
)
|
||||
|
||||
def _make_read_only(path: Path):
|
||||
current_permissions = path.stat().st_mode
|
||||
path.chmod(current_permissions & ~ANYONE_WRITE_PERMISSION)
|
||||
|
||||
read_only_templates_dir = str(Path(mkdtemp()) / "templates")
|
||||
copytree(templates_dir, read_only_templates_dir)
|
||||
|
||||
for root, dirs, files in os.walk(read_only_templates_dir):
|
||||
for node in chain(dirs, files):
|
||||
_make_read_only(Path(root, node))
|
||||
|
||||
destination = mkdtemp()
|
||||
process = subprocess.Popen(
|
||||
(
|
||||
sys.executable,
|
||||
"-m",
|
||||
"scrapy.cmdline",
|
||||
"startproject",
|
||||
project_name,
|
||||
"--set",
|
||||
f"TEMPLATES_DIR={read_only_templates_dir}",
|
||||
),
|
||||
cwd=destination,
|
||||
env=self.env,
|
||||
)
|
||||
process.wait()
|
||||
|
||||
project_dir = Path(destination, project_name)
|
||||
actual_permissions = get_permissions_dict(project_dir)
|
||||
|
||||
assert actual_permissions == expected_permissions
|
||||
|
||||
def test_startproject_permissions_unchanged_in_destination(self):
|
||||
"""Check that preexisting folders and files in the destination folder
|
||||
do not see their permissions modified."""
|
||||
scrapy_path = scrapy.__path__[0]
|
||||
project_template = Path(scrapy_path, "templates", "project")
|
||||
project_name = "startproject3"
|
||||
renamings = (
|
||||
("module", project_name),
|
||||
(".tmpl", ""),
|
||||
)
|
||||
expected_permissions = get_permissions_dict(
|
||||
project_template,
|
||||
renamings,
|
||||
IGNORE,
|
||||
)
|
||||
|
||||
destination = mkdtemp()
|
||||
project_dir = Path(destination, project_name)
|
||||
|
||||
existing_nodes = {
|
||||
oct(permissions)[2:] + extension: permissions
|
||||
for extension in ("", ".d")
|
||||
for permissions in (
|
||||
0o444,
|
||||
0o555,
|
||||
0o644,
|
||||
0o666,
|
||||
0o755,
|
||||
0o777,
|
||||
)
|
||||
}
|
||||
project_dir.mkdir()
|
||||
for node, permissions in existing_nodes.items():
|
||||
path = project_dir / node
|
||||
if node.endswith(".d"):
|
||||
path.mkdir(mode=permissions)
|
||||
else:
|
||||
path.touch(mode=permissions)
|
||||
expected_permissions[node] = oct(path.stat().st_mode)
|
||||
|
||||
process = subprocess.Popen(
|
||||
(
|
||||
sys.executable,
|
||||
"-m",
|
||||
"scrapy.cmdline",
|
||||
"startproject",
|
||||
project_name,
|
||||
".",
|
||||
),
|
||||
cwd=project_dir,
|
||||
env=self.env,
|
||||
)
|
||||
process.wait()
|
||||
|
||||
actual_permissions = get_permissions_dict(project_dir)
|
||||
|
||||
assert actual_permissions == expected_permissions
|
||||
|
||||
def test_startproject_permissions_umask_022(self):
|
||||
"""Check that generated files have the right permissions when the
|
||||
system uses a umask value that causes new files to have different
|
||||
permissions than those from the template folder."""
|
||||
|
||||
@contextmanager
|
||||
def umask(new_mask):
|
||||
cur_mask = os.umask(new_mask)
|
||||
yield
|
||||
os.umask(cur_mask)
|
||||
|
||||
scrapy_path = scrapy.__path__[0]
|
||||
project_template = Path(scrapy_path, "templates", "project")
|
||||
project_name = "umaskproject"
|
||||
renamings = (
|
||||
("module", project_name),
|
||||
(".tmpl", ""),
|
||||
)
|
||||
expected_permissions = get_permissions_dict(
|
||||
project_template,
|
||||
renamings,
|
||||
IGNORE,
|
||||
)
|
||||
|
||||
with umask(0o002):
|
||||
destination = mkdtemp()
|
||||
process = subprocess.Popen(
|
||||
(
|
||||
sys.executable,
|
||||
"-m",
|
||||
"scrapy.cmdline",
|
||||
"startproject",
|
||||
project_name,
|
||||
),
|
||||
cwd=destination,
|
||||
env=self.env,
|
||||
)
|
||||
process.wait()
|
||||
|
||||
project_dir = Path(destination, project_name)
|
||||
actual_permissions = get_permissions_dict(project_dir)
|
||||
|
||||
assert actual_permissions == expected_permissions
|
||||
File diff suppressed because it is too large
Load Diff
Loading…
Reference in New Issue