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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
|
from basetest import *
from zfs_autobackup.ExecuteNode import *
print("THIS TEST REQUIRES SSH TO LOCALHOST")
class TestExecuteNode(unittest2.TestCase):
# def setUp(self):
# return super().setUp()
def basics(self, node ):
with self.subTest("simple echo"):
self.assertEqual(node.run(["echo","test"]), ["test"])
with self.subTest("error exit code"):
with self.assertRaises(ExecuteError):
node.run(["false"])
#
with self.subTest("multiline without tabsplit"):
self.assertEqual(node.run(["echo","l1c1\tl1c2\nl2c1\tl2c2"], tab_split=False), ["l1c1\tl1c2", "l2c1\tl2c2"])
#multiline tabsplit
with self.subTest("multiline tabsplit"):
self.assertEqual(node.run(["echo","l1c1\tl1c2\nl2c1\tl2c2"], tab_split=True), [['l1c1', 'l1c2'], ['l2c1', 'l2c2']])
#escaping test
with self.subTest("escape test"):
s="><`'\"@&$()$bla\\/.* !#test _+-={}[]|${bla} $bla"
self.assertEqual(node.run(["echo",s]), [s])
#return std err as well, trigger stderr by listing something non existing
with self.subTest("stderr return"):
(stdout, stderr)=node.run(["sh", "-c", "echo bla >&2"], return_stderr=True, valid_exitcodes=[0])
self.assertEqual(stdout,[])
self.assertRegex(stderr[0],"bla")
#slow command, make sure things dont exit too early
with self.subTest("early exit test"):
start_time=time.time()
self.assertEqual(node.run(["sleep","1"]), [])
self.assertGreaterEqual(time.time()-start_time,1)
#input a string and check it via cat
with self.subTest("stdin input string"):
self.assertEqual(node.run(["cat"], inp="test"), ["test"])
#command that wants input, while we dont have input, shouldnt hang forever.
with self.subTest("stdin process with inp=None (shouldn't hang)"):
self.assertEqual(node.run(["cat"]), [])
# let the system do the piping with an unescaped |:
with self.subTest("system piping test"):
#first make sure the actual | character is still properly escaped:
self.assertEqual(node.run(["echo","|"]), ["|"])
#now pipe
self.assertEqual(node.run(["echo", "abc", node.PIPE, "tr", "a", "A" ]), ["Abc"])
def test_basics_local(self):
node=ExecuteNode(debug_output=True)
self.basics(node)
def test_basics_remote(self):
node=ExecuteNode(ssh_to="localhost", debug_output=True)
self.basics(node)
################
def test_readonly(self):
node=ExecuteNode(debug_output=True, readonly=True)
self.assertEqual(node.run(["echo","test"], readonly=False), [])
self.assertEqual(node.run(["echo","test"], readonly=True), ["test"])
################
def pipe(self, nodea, nodeb):
with self.subTest("pipe data"):
output=nodea.run(["dd", "if=/dev/zero", "count=1000"],pipe=True)
self.assertEqual(nodeb.run(["md5sum"], inp=output), ["816df6f64deba63b029ca19d880ee10a -"])
with self.subTest("exit code both ends of pipe ok"):
output=nodea.run(["true"], pipe=True)
nodeb.run(["true"], inp=output)
with self.subTest("error on pipe input side"):
with self.assertRaises(ExecuteError):
output=nodea.run(["false"], pipe=True)
nodeb.run(["true"], inp=output)
with self.subTest("error on both sides, ignore exit codes"):
output=nodea.run(["false"], pipe=True, valid_exitcodes=[])
nodeb.run(["false"], inp=output, valid_exitcodes=[])
with self.subTest("error on pipe output side "):
with self.assertRaises(ExecuteError):
output=nodea.run(["true"], pipe=True)
nodeb.run(["false"], inp=output)
with self.subTest("error on both sides of pipe"):
with self.assertRaises(ExecuteError):
output=nodea.run(["false"], pipe=True)
nodeb.run(["false"], inp=output)
with self.subTest("check stderr on pipe output side"):
output=nodea.run(["true"], pipe=True, valid_exitcodes=[0])
(stdout, stderr)=nodeb.run(["sh", "-c", "echo bla >&2"], inp=output, return_stderr=True, valid_exitcodes=[0])
self.assertEqual(stdout,[])
self.assertRegex(stderr[0], "bla" )
with self.subTest("check stderr on pipe input side (should be only printed)"):
output=nodea.run(["sh", "-c", "echo bla >&2"], pipe=True, valid_exitcodes=[0])
(stdout, stderr)=nodeb.run(["true"], inp=output, return_stderr=True, valid_exitcodes=[0])
self.assertEqual(stdout,[])
self.assertEqual(stderr,[])
def test_pipe_local_local(self):
nodea=ExecuteNode(debug_output=True)
nodeb=ExecuteNode(debug_output=True)
self.pipe(nodea, nodeb)
def test_pipe_remote_remote(self):
nodea=ExecuteNode(ssh_to="localhost", debug_output=True)
nodeb=ExecuteNode(ssh_to="localhost", debug_output=True)
self.pipe(nodea, nodeb)
def test_pipe_local_remote(self):
nodea=ExecuteNode(debug_output=True)
nodeb=ExecuteNode(ssh_to="localhost", debug_output=True)
self.pipe(nodea, nodeb)
def test_pipe_remote_local(self):
nodea=ExecuteNode(ssh_to="localhost", debug_output=True)
nodeb=ExecuteNode(debug_output=True)
self.pipe(nodea, nodeb)
def test_cwd(self):
nodea=ExecuteNode(ssh_to="localhost", debug_output=True)
nodeb=ExecuteNode(debug_output=True)
#change to a directory with a space and execute a system pipe, check if all piped commands are executed in correct directory.
shelltest("mkdir '/tmp/space test' 2>/dev/null; true")
self.assertEqual(nodea.run(cmd=["pwd", ExecuteNode.PIPE, "cat"], cwd="/tmp/space test"), ["/tmp/space test"])
self.assertEqual(nodea.run(cmd=["cat", ExecuteNode.PIPE, "pwd"], cwd="/tmp/space test"), ["/tmp/space test"])
self.assertEqual(nodeb.run(cmd=["pwd", ExecuteNode.PIPE, "cat"], cwd="/tmp/space test"), ["/tmp/space test"])
self.assertEqual(nodeb.run(cmd=["cat", ExecuteNode.PIPE, "pwd"], cwd="/tmp/space test"), ["/tmp/space test"])
def test_script_handlers(self):
def test(node):
results = []
node.script(lines=["echo line1", "echo line2 1>&2", "exit 123"],
stdout_handler=lambda line: results.append(line),
stderr_handler=lambda line: results.append(line),
exit_handler=lambda exit_code: results.append(exit_code),
valid_exitcodes=[123]
)
self.assertEqual(results, ["line1", "line2", 123 ])
with self.subTest("remote"):
test(ExecuteNode(ssh_to="localhost", debug_output=True))
#
with self.subTest("local"):
test(ExecuteNode(debug_output=True))
def test_script_defaults(self):
result=[]
nodea=ExecuteNode(debug_output=True)
nodea.script(lines=["echo test"], stdout_handler=lambda line: result.append(line))
self.assertEqual(result, ["test"])
def test_script_pipe(self):
result=[]
nodea=ExecuteNode()
cmd_pipe=nodea.script(lines=["echo test"], pipe=True)
nodea.script(lines=["tr e E"], inp=cmd_pipe,stdout_handler=lambda line: result.append(line))
self.assertEqual(result, ["tEst"])
def test_mixed(self):
#should be able to mix run() and script()
node=ExecuteNode()
result=[]
pipe=node.run(["echo", "test"], pipe=True)
node.script(["tr e E"], inp=pipe, stdout_handler=lambda line: result.append(line))
self.assertEqual(result, ["tEst"])
|
