1import sys
2import os
3
4from twisted.internet import defer, protocol
5
6
7class ProcessTest:
8
9    command = None
10    prefix = [sys.executable, '-m', 'scrapy.cmdline']
11    cwd = os.getcwd()  # trial chdirs to temp dir
12
13    def execute(self, args, check_code=True, settings=None):
14        from twisted.internet import reactor
15        env = os.environ.copy()
16        if settings is not None:
17            env['SCRAPY_SETTINGS_MODULE'] = settings
18        cmd = self.prefix + [self.command] + list(args)
19        pp = TestProcessProtocol()
20        pp.deferred.addBoth(self._process_finished, cmd, check_code)
21        reactor.spawnProcess(pp, cmd[0], cmd, env=env, path=self.cwd)
22        return pp.deferred
23
24    def _process_finished(self, pp, cmd, check_code):
25        if pp.exitcode and check_code:
26            msg = f"process {cmd} exit with code {pp.exitcode}"
27            msg += f"\n>>> stdout <<<\n{pp.out}"
28            msg += "\n"
29            msg += f"\n>>> stderr <<<\n{pp.err}"
30            raise RuntimeError(msg)
31        return pp.exitcode, pp.out, pp.err
32
33
34class TestProcessProtocol(protocol.ProcessProtocol):
35
36    def __init__(self):
37        self.deferred = defer.Deferred()
38        self.out = b''
39        self.err = b''
40        self.exitcode = None
41
42    def outReceived(self, data):
43        self.out += data
44
45    def errReceived(self, data):
46        self.err += data
47
48    def processEnded(self, status):
49        self.exitcode = status.value.exitCode
50        self.deferred.callback(self)
51