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