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

seperated piping

Browse Source
This commit is contained in:
Edwin Eefting
2021-04-05 22:18:14 +02:00
parent 966df73d2f
commit 4910b1dfb5
5 changed files with 163 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
tests/test_executenode.py
View File

@ -73,39 +73,39 @@ class TestExecuteNode(unittest2.TestCase):
def pipe(self, nodea, nodeb):
with self.subTest("pipe data"):
output=nodea.run(["dd", "if=/dev/zero", "count=1000"], pipe=True)
output=nodea.get_pipe(["dd", "if=/dev/zero", "count=1000"])
self.assertEqual(nodeb.run(["md5sum"], inp=output), ["816df6f64deba63b029ca19d880ee10a -"])
with self.subTest("exit code both ends of pipe ok"):
output=nodea.run(["true"], pipe=True)
output=nodea.get_pipe(["true"])
nodeb.run(["true"], inp=output)
with self.subTest("error on pipe input side"):
with self.assertRaises(subprocess.CalledProcessError):
output=nodea.run(["false"], pipe=True)
output=nodea.get_pipe(["false"])
nodeb.run(["true"], inp=output)
with self.subTest("error on pipe output side "):
with self.assertRaises(subprocess.CalledProcessError):
output=nodea.run(["true"], pipe=True)
output=nodea.get_pipe(["true"])
nodeb.run(["false"], inp=output)
with self.subTest("error on both sides of pipe"):
with self.assertRaises(subprocess.CalledProcessError):
output=nodea.run(["false"], pipe=True)
output=nodea.get_pipe(["false"])
nodeb.run(["false"], inp=output)
with self.subTest("check stderr on pipe output side"):
output=nodea.run(["true"], pipe=True)
output=nodea.get_pipe(["true"])
(stdout, stderr)=nodeb.run(["ls", "nonexistingfile"], inp=output, return_stderr=True, valid_exitcodes=[0,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)
output=nodea.get_pipe(["ls", "nonexistingfile"])
(stdout, stderr)=nodeb.run(["true"], inp=output, return_stderr=True, valid_exitcodes=[0,2])
self.assertEqual(stdout,[])
self.assertEqual(stderr,[] )
self.assertEqual(stderr,[])