third-party-mirrors/zfs_autobackup
1
0
Fork 1
mirror of https://github.com/psy0rz/zfs_autobackup.git synced 2025-04-13 22:47:12 +03:00
Code Issues Actions Packages Projects Releases Wiki Activity

subtests

Browse Source
This commit is contained in:
Edwin Eefting 2020-05-12 18:28:16 +02:00
parent c864e5ffad
commit 2fa95f098b
1 changed files with 48 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
test_executenode.py
View File

@ -16,35 +16,41 @@ class TestExecuteNode(unittest.TestCase):
def basics(self, node ):
#single line
self.assertEqual(node.run(["echo","test"]), ["test"])
with self.subTest("simple echo"):
self.assertEqual(node.run(["echo","test"]), ["test"])
#error exit code
with self.assertRaises(subprocess.CalledProcessError):
node.run(["false"])
with self.subTest("error exit code"):
with self.assertRaises(subprocess.CalledProcessError):
node.run(["false"])
#multiline without tabsplit
self.assertEqual(node.run(["echo","l1c1\tl1c2\nl2c1\tl2c2"], tab_split=False), ["l1c1\tl1c2", "l2c1\tl2c2"])
#
with self.subTest("multiline without tabsplit"):
self.assertEqual(node.run(["echo","l1c1\tl1c2\nl2c1\tl2c2"], tab_split=False), ["l1c1\tl1c2", "l2c1\tl2c2"])
#multiline tabsplit
self.assertEqual(node.run(["echo","l1c1\tl1c2\nl2c1\tl2c2"], tab_split=True), [['l1c1', 'l1c2'], ['l2c1', 'l2c2']])
with self.subTest("multiline tabsplit"):
self.assertEqual(node.run(["echo","l1c1\tl1c2\nl2c1\tl2c2"], tab_split=True), [['l1c1', 'l1c2'], ['l2c1', 'l2c2']])
#escaping test (shouldnt be a problem locally, single quotes can be a problem remote via ssh)
s="><`'\"@&$()$bla\\/.*!#test _+-={}[]|"
self.assertEqual(node.run(["echo",s]), [s])
with self.subTest("escape test"):
s="><`'\"@&$()$bla\\/.*!#test _+-={}[]|"
self.assertEqual(node.run(["echo",s]), [s])
#return std err as well, trigger stderr by listing something non existing
(stdout, stderr)=node.run(["ls", "nonexistingfile"], return_stderr=True, valid_exitcodes=[2])
self.assertEqual(stdout,[])
self.assertRegex(stderr[0],"nonexistingfile")
with self.subTest("stderr return"):
(stdout, stderr)=node.run(["ls", "nonexistingfile"], return_stderr=True, valid_exitcodes=[2])
self.assertEqual(stdout,[])
self.assertRegex(stderr[0],"nonexistingfile")
#slow command, make sure things dont exit too early
start_time=time.time()
self.assertEqual(node.run(["sleep","1"]), [])
self.assertGreaterEqual(time.time()-start_time,1)
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
self.assertEqual(node.run(["cat"], input="test"), ["test"])
with self.subTest("stdin input string"):
self.assertEqual(node.run(["cat"], input="test"), ["test"])
def test_basics_local(self):
@ -67,27 +73,40 @@ class TestExecuteNode(unittest.TestCase):
################
def pipe(self, nodea, nodeb):
output=nodea.run(["dd", "if=/dev/zero", "count=1000"], pipe=True)
self.assertEqual(nodeb.run(["md5sum"], input=output), ["816df6f64deba63b029ca19d880ee10a -"])
with self.subTest("pipe data"):
output=nodea.run(["dd", "if=/dev/zero", "count=1000"], pipe=True)
self.assertEqual(nodeb.run(["md5sum"], input=output), ["816df6f64deba63b029ca19d880ee10a -"])
#both ok
output=nodea.run(["true"], pipe=True)
nodeb.run(["true"], input=output)
with self.subTest("exit code both ok"):
output=nodea.run(["true"], pipe=True)
nodeb.run(["true"], input=output)
#remote error
with self.assertRaises(subprocess.CalledProcessError):
output=nodea.run(["false"], pipe=True)
nodeb.run(["true"], input=output)
with self.subTest("node a error"):
with self.assertRaises(subprocess.CalledProcessError):
output=nodea.run(["false"], pipe=True)
nodeb.run(["true"], input=output)
#local error
with self.assertRaises(subprocess.CalledProcessError):
output=nodea.run(["true"], pipe=True)
nodeb.run(["false"], input=output)
with self.subTest("node b error"):
with self.assertRaises(subprocess.CalledProcessError):
output=nodea.run(["true"], pipe=True)
nodeb.run(["false"], input=output)
#both error
with self.assertRaises(subprocess.CalledProcessError):
with self.subTest("both nodes error"):
with self.assertRaises(subprocess.CalledProcessError):
output=nodea.run(["false"], pipe=True)
nodeb.run(["false"], input=output)
#capture local stderr
with self.subTest("pipe local stderr output"):
output=nodea.run(["true"], pipe=True)
nodeb.run(["false"], input=output)
(stdout, stderr)=nodeb.run(["ls", "nonexistingfile"], input=output, return_stderr=True, valid_exitcodes=[2])
self.assertEqual(stdout,[])
self.assertRegex(stderr[0],"nonexistingfile")