1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
from __future__ import annotations
from typing import TYPE_CHECKING
from tests.test_commands import TestProjectBase
from tests.utils.cmdline import proc
if TYPE_CHECKING:
from collections.abc import Iterable
from pathlib import Path
class TestCrawlCommand(TestProjectBase):
def crawl(
self, code: str, proj_path: Path, args: Iterable[str] = ()
) -> tuple[int, str, str]:
(proj_path / self.project_name / "spiders" / "myspider.py").write_text(
code, encoding="utf-8"
)
return proc("crawl", "myspider", *args, cwd=proj_path)
def get_log(self, code: str, proj_path: Path, args: Iterable[str] = ()) -> str:
_, _, stderr = self.crawl(code, proj_path, args=args)
return stderr
def test_no_output(self, proj_path: Path) -> None:
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, proj_path)
assert "[myspider] DEBUG: It works!" in log
assert (
"Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
in log
)
assert "Spider closed (finished)" in log
def test_output(self, proj_path: Path) -> None:
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, proj_path, args=args)
assert "[myspider] DEBUG: FEEDS: {'example.json': {'format': 'json'}}" in log
def test_overwrite_output(self, proj_path: Path) -> None:
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
"""
j = proj_path / "example.json"
j.write_text("not empty", encoding="utf-8")
args = ["-O", "example.json"]
log = self.get_log(spider_code, proj_path, args=args)
assert (
'[myspider] DEBUG: FEEDS: {"example.json": {"format": "json", "overwrite": true}}'
in log
)
with j.open(encoding="utf-8") as f2:
first_line = f2.readline()
assert first_line != "not empty"
def test_output_and_overwrite_output(self, proj_path: Path) -> None:
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, proj_path, args=args)
assert (
"error: Please use only one of -o/--output and -O/--overwrite-output" in log
)
def test_default_reactor(self, proj_path: Path) -> None:
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, proj_path, args=("-s", "TWISTED_REACTOR="))
assert "[myspider] DEBUG: It works!" in log
assert (
"Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
not in log
)
assert "Spider closed (finished)" in log
|
