patrick/zfs_autobackup
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

better handling of piped exit codes

Browse Source
This commit is contained in:
Edwin Eefting
2021-04-22 01:12:41 +02:00
parent 8ec5ed2f4f
commit 401a3f73cc
6 changed files with 54 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
tests/test_cmdpipe.py
View File

@ -9,26 +9,24 @@ class TestCmdPipe(unittest2.TestCase):
p=CmdPipe(readonly=False, inp=None)
err=[]
out=[]
p.add(["ls", "-d", "/", "/", "/nonexistent"], stderr_handler=lambda line: err.append(line))
p.add(["ls", "-d", "/", "/", "/nonexistent"], stderr_handler=lambda line: err.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2))
executed=p.execute(stdout_handler=lambda line: out.append(line))
self.assertEqual(err, ["ls: cannot access '/nonexistent': No such file or directory"])
self.assertEqual(out, ["/","/"])
self.assertTrue(executed)
self.assertEqual(p.items[0]['process'].returncode,2)
def test_input(self):
"""test stdinput"""
p=CmdPipe(readonly=False, inp="test")
err=[]
out=[]
p.add(["echo", "test"], stderr_handler=lambda line: err.append(line))
p.add(["echo", "test"], stderr_handler=lambda line: err.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,0))
executed=p.execute(stdout_handler=lambda line: out.append(line))
self.assertEqual(err, [])
self.assertEqual(out, ["test"])
self.assertTrue(executed)
self.assertEqual(p.items[0]['process'].returncode,0)
def test_pipe(self):
"""test piped"""
@ -37,9 +35,9 @@ class TestCmdPipe(unittest2.TestCase):
err2=[]
err3=[]
out=[]
p.add(["echo", "test"], stderr_handler=lambda line: err1.append(line))
p.add(["tr", "e", "E"], stderr_handler=lambda line: err2.append(line))
p.add(["tr", "t", "T"], stderr_handler=lambda line: err3.append(line))
p.add(["echo", "test"], stderr_handler=lambda line: err1.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,0))
p.add(["tr", "e", "E"], stderr_handler=lambda line: err2.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,0))
p.add(["tr", "t", "T"], stderr_handler=lambda line: err3.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,0))
executed=p.execute(stdout_handler=lambda line: out.append(line))
self.assertEqual(err1, [])
@ -47,9 +45,6 @@ class TestCmdPipe(unittest2.TestCase):
self.assertEqual(err3, [])
self.assertEqual(out, ["TEsT"])
self.assertTrue(executed)
self.assertEqual(p.items[0]['process'].returncode,0)
self.assertEqual(p.items[1]['process'].returncode,0)
self.assertEqual(p.items[2]['process'].returncode,0)
#test str representation as well
self.assertEqual(str(p), "(echo test) | (tr e E) | (tr t T)")
@ -61,9 +56,9 @@ class TestCmdPipe(unittest2.TestCase):
err2=[]
err3=[]
out=[]
p.add(["ls", "/nonexistent1"], stderr_handler=lambda line: err1.append(line))
p.add(["ls", "/nonexistent2"], stderr_handler=lambda line: err2.append(line))
p.add(["ls", "/nonexistent3"], stderr_handler=lambda line: err3.append(line))
p.add(["ls", "/nonexistent1"], stderr_handler=lambda line: err1.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2))
p.add(["ls", "/nonexistent2"], stderr_handler=lambda line: err2.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2))
p.add(["ls", "/nonexistent3"], stderr_handler=lambda line: err3.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2))
executed=p.execute(stdout_handler=lambda line: out.append(line))
self.assertEqual(err1, ["ls: cannot access '/nonexistent1': No such file or directory"])
@ -71,9 +66,6 @@ class TestCmdPipe(unittest2.TestCase):
self.assertEqual(err3, ["ls: cannot access '/nonexistent3': No such file or directory"])
self.assertEqual(out, [])
self.assertTrue(executed)
self.assertEqual(p.items[0]['process'].returncode,2)
self.assertEqual(p.items[1]['process'].returncode,2)
self.assertEqual(p.items[2]['process'].returncode,2)
def test_exitcode(self):
"""test piped exitcodes """
@ -82,9 +74,9 @@ class TestCmdPipe(unittest2.TestCase):
err2=[]
err3=[]
out=[]
p.add(["bash", "-c", "exit 1"], stderr_handler=lambda line: err1.append(line))
p.add(["bash", "-c", "exit 2"], stderr_handler=lambda line: err2.append(line))
p.add(["bash", "-c", "exit 3"], stderr_handler=lambda line: err3.append(line))
p.add(["bash", "-c", "exit 1"], stderr_handler=lambda line: err1.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,1))
p.add(["bash", "-c", "exit 2"], stderr_handler=lambda line: err2.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2))
p.add(["bash", "-c", "exit 3"], stderr_handler=lambda line: err3.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,3))
executed=p.execute(stdout_handler=lambda line: out.append(line))
self.assertEqual(err1, [])
@ -92,9 +84,6 @@ class TestCmdPipe(unittest2.TestCase):
self.assertEqual(err3, [])
self.assertEqual(out, [])
self.assertTrue(executed)
self.assertEqual(p.items[0]['process'].returncode,1)
self.assertEqual(p.items[1]['process'].returncode,2)
self.assertEqual(p.items[2]['process'].returncode,3)
def test_readonly_execute(self):
"""everything readonly, just should execute"""

22
tests/test_executenode.py
View File

@ -1,5 +1,5 @@
from basetest import *
from zfs_autobackup.ExecuteNode import ExecuteNode
from zfs_autobackup.ExecuteNode import *
print("THIS TEST REQUIRES SSH TO LOCALHOST")
@ -15,7 +15,7 @@ class TestExecuteNode(unittest2.TestCase):
self.assertEqual(node.run(["echo","test"]), ["test"])
with self.subTest("error exit code"):
with self.assertRaises(subprocess.CalledProcessError):
with self.assertRaises(ExecuteError):
node.run(["false"])
#
@ -81,29 +81,33 @@ class TestExecuteNode(unittest2.TestCase):
nodeb.run(["true"], inp=output)
with self.subTest("error on pipe input side"):
with self.assertRaises(subprocess.CalledProcessError):
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(subprocess.CalledProcessError):
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(subprocess.CalledProcessError):
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)
(stdout, stderr)=nodeb.run(["ls", "nonexistingfile"], inp=output, return_stderr=True, valid_exitcodes=[0,2])
output=nodea.run(["true"], pipe=True, valid_exitcodes=[0])
(stdout, stderr)=nodeb.run(["ls", "nonexistingfile"], inp=output, return_stderr=True, valid_exitcodes=[2])
self.assertEqual(stdout,[])
self.assertRegex(stderr[0], "nonexistingfile" )
with self.subTest("check stderr on pipe input side (should be only printed)"):
output=nodea.run(["ls", "nonexistingfile"], pipe=True)
(stdout, stderr)=nodeb.run(["true"], inp=output, return_stderr=True, valid_exitcodes=[0,2])
output=nodea.run(["ls", "nonexistingfile"], pipe=True, valid_exitcodes=[2])
(stdout, stderr)=nodeb.run(["true"], inp=output, return_stderr=True, valid_exitcodes=[0])
self.assertEqual(stdout,[])
self.assertEqual(stderr,[])