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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
|
import atexit
import os
import sys
import unittest
from .impl_aep_asyncio import run
from .impl_aep_asyncio import loop
this_dir = os.path.dirname(os.path.abspath(__file__))
test_script = os.path.join(
this_dir,
'fixtures',
'execute_process',
'stdout_stderr_ordering.py')
test_script_quoted = '"%s"' % test_script if ' ' in test_script else test_script
python = sys.executable
# This atexit handler ensures the loop is closed after all tests were run.
@atexit.register
def close_loop():
if not loop.is_closed():
loop.close()
class TestProcessUtilsAsyncExecuteProcess(unittest.TestCase):
def test_async_execute_process_no_emulation_shell_false_combined(self):
stdout, stderr, retcode = loop.run_until_complete(run(
[python, test_script]))
stdout, stderr = stdout.decode(), stderr.decode()
self.assertIn('out 1', stdout)
self.assertIn('err 1', stdout)
self.assertIn('out 2', stdout)
self.assertEqual('', stderr)
self.assertEqual(0, retcode)
def test_async_execute_process_no_emulation_shell_true_combined(self):
stdout, stderr, retcode = loop.run_until_complete(run(
[python, test_script_quoted], shell=True))
stdout, stderr = stdout.decode(), stderr.decode()
self.assertIn('out 1', stdout)
self.assertIn('err 1', stdout)
self.assertIn('out 2', stdout)
self.assertEqual('', stderr)
self.assertEqual(0, retcode)
def test_async_execute_process_no_emulation_shell_false(self):
stdout, stderr, retcode = loop.run_until_complete(run(
[python, test_script], stderr_to_stdout=False))
stdout, stderr = stdout.decode(), stderr.decode()
self.assertIn('out 1', stdout)
self.assertIn('err 1', stderr)
self.assertIn('out 2', stdout)
self.assertEqual(0, retcode)
def test_async_execute_process_no_emulation_shell_true(self):
stdout, stderr, retcode = loop.run_until_complete(run(
[python, test_script_quoted], stderr_to_stdout=False, shell=True))
stdout, stderr = stdout.decode(), stderr.decode()
self.assertIn('out 1', stdout)
self.assertIn('err 1', stderr)
self.assertIn('out 2', stdout)
self.assertEqual(0, retcode)
def test_async_execute_process_with_emulation_shell_false_combined(self):
stdout, stderr, retcode = loop.run_until_complete(run(
[python, test_script], emulate_tty=True))
stdout, stderr = stdout.decode(), stderr.decode()
self.assertIn('out 1', stdout)
self.assertIn('err 1', stdout)
self.assertIn('out 2', stdout)
self.assertEqual('', stderr)
self.assertEqual(0, retcode)
def test_async_execute_process_with_emulation_shell_true_combined(self):
stdout, stderr, retcode = loop.run_until_complete(run(
[python, test_script_quoted], emulate_tty=True, shell=True))
stdout, stderr = stdout.decode(), stderr.decode()
self.assertIn('out 1', stdout)
self.assertIn('err 1', stdout)
self.assertIn('out 2', stdout)
self.assertEqual('', stderr)
self.assertEqual(0, retcode)
def test_async_execute_process_with_emulation_shell_false(self):
stdout, stderr, retcode = loop.run_until_complete(run(
[python, test_script], emulate_tty=True, stderr_to_stdout=False))
stdout, stderr = stdout.decode(), stderr.decode()
self.assertIn('out 1', stdout)
self.assertIn('err 1', stderr)
self.assertIn('out 2', stdout)
self.assertEqual(0, retcode)
def test_async_execute_process_with_emulation_shell_true(self):
stdout, stderr, retcode = loop.run_until_complete(run(
[python, test_script_quoted], emulate_tty=True, stderr_to_stdout=False,
shell=True))
stdout, stderr = stdout.decode(), stderr.decode()
self.assertIn('out 1', stdout)
self.assertIn('err 1', stderr)
self.assertIn('out 2', stdout)
self.assertEqual(0, retcode)
|
