process_test.py 文件源码

def test_capture(self):
    """Test capturing output."""
    self.popen.returncode = 0
    self.popen.communicate.return_value = ('Test', None)
    self.assertEqual(
        (0, 'Test'),
        process.call('test', cwd='path', env={'NEW': '2'}, capture=True))

    self.mock.Popen.assert_called_once_with(
        'test', shell=True, cwd='path', env={'TEST': '1', 'NEW': '2'},
        stdout=subprocess.PIPE, preexec_fn=os.setsid)
    self.popen.communicate.assert_called_once_with()
    self.mock.store_last_pid.assert_called_once_with(123)
    self.assert_exact_calls(self.mock.kill_last_pid, [mock.call()] * 2)
    self.mock.Thread.assert_called_once_with(
        target=process.kill_when_timeout,
        args=(self.popen, process.DEFAULT_TIMEOUT))
    self.mock.Thread.return_value.start.assert_called_once_with()