scrapy / scrapy
1 7
import sys
2 7
import os
3

4 7
from twisted.internet import defer, protocol
5

6

7 7
class ProcessTest:
8

9 7
    command = None
10 7
    prefix = [sys.executable, '-m', 'scrapy.cmdline']
11 7
    cwd = os.getcwd()  # trial chdirs to temp dir
12

13 7
    def execute(self, args, check_code=True, settings=None):
14 7
        from twisted.internet import reactor
15 7
        env = os.environ.copy()
16 7
        if settings is not None:
17 0
            env['SCRAPY_SETTINGS_MODULE'] = settings
18 7
        cmd = self.prefix + [self.command] + list(args)
19 7
        pp = TestProcessProtocol()
20 7
        pp.deferred.addBoth(self._process_finished, cmd, check_code)
21 7
        reactor.spawnProcess(pp, cmd[0], cmd, env=env, path=self.cwd)
22 7
        return pp.deferred
23

24 7
    def _process_finished(self, pp, cmd, check_code):
25 7
        if pp.exitcode and check_code:
26 0
            msg = f"process {cmd} exit with code {pp.exitcode}"
27 0
            msg += f"\n>>> stdout <<<\n{pp.out}"
28 0
            msg += "\n"
29 0
            msg += f"\n>>> stderr <<<\n{pp.err}"
30 0
            raise RuntimeError(msg)
31 7
        return pp.exitcode, pp.out, pp.err
32

33

34 7
class TestProcessProtocol(protocol.ProcessProtocol):
35

36 7
    def __init__(self):
37 7
        self.deferred = defer.Deferred()
38 7
        self.out = b''
39 7
        self.err = b''
40 7
        self.exitcode = None
41

42 7
    def outReceived(self, data):
43 7
        self.out += data
44

45 7
    def errReceived(self, data):
46 7
        self.err += data
47

48 7
    def processEnded(self, status):
49 7
        self.exitcode = status.value.exitCode
50 7
        self.deferred.callback(self)

Read our documentation on viewing source code .

Loading