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