| import unittest |
| import subprocess |
| import os |
| |
| from nose.plugins.skip import SkipTest |
| |
| from util.commands import run_cmd, get_output, run_cmd_periodic_poll, TERMINATED_PROCESS_MSG |
| |
| |
| def has_urandom(): |
| return os.path.exists('/dev/urandom') |
| |
| |
| class TestRunCmd(unittest.TestCase): |
| def testSimple(self): |
| self.assertEquals(run_cmd(['true']), 0) |
| |
| def testFailure(self): |
| self.assertRaises(subprocess.CalledProcessError, run_cmd, ['false']) |
| |
| |
| class TestRunCmdPeriodicPoll(unittest.TestCase): |
| def testSimple(self): |
| self.assertEquals(run_cmd_periodic_poll(['true']), 0) |
| |
| def testFailure(self): |
| self.assertRaises(subprocess.CalledProcessError, run_cmd_periodic_poll, |
| ['false']) |
| |
| def testSuccess2secs(self): |
| self.assertEquals( |
| run_cmd_periodic_poll(['bash', '-c', 'sleep 2 && true']), |
| 0) |
| |
| def testFailure2secs(self): |
| self.assertRaises( |
| subprocess.CalledProcessError, run_cmd_periodic_poll, |
| ['bash', '-c', 'sleep 2 && false']) |
| |
| def testSuccess3secsWith2secsPoll(self): |
| self.assertEquals( |
| run_cmd_periodic_poll(['bash', '-c', 'sleep 3 && true'], |
| warning_interval=2), |
| 0) |
| |
| def testTerminatedByTimeout(self): |
| self.assertRaises( |
| subprocess.CalledProcessError, run_cmd_periodic_poll, |
| ['bash', '-c', 'sleep 1 && true'], warning_interval=1, timeout=0.5) |
| |
| def testSuccessCallback(self): |
| self.callback_called = 0 |
| |
| def callback(start_time, eapsed, proc): |
| self.callback_called += 1 |
| |
| run_cmd_periodic_poll(['bash', '-c', 'sleep 5 && true'], |
| warning_interval=2, warning_callback=callback), |
| self.assertEqual(self.callback_called, 2) |
| |
| def testFailureCallback(self): |
| self.callback_called = 0 |
| |
| def callback(start_time, eapsed, proc): |
| self.callback_called += 1 |
| |
| self.assertRaises( |
| subprocess.CalledProcessError, run_cmd_periodic_poll, |
| ['bash', '-c', 'sleep 5 && false'], warning_interval=2, |
| warning_callback=callback) |
| self.assertEqual(self.callback_called, 2) |
| |
| |
| class TestGetOutput(unittest.TestCase): |
| def testOutput(self): |
| output = get_output(['echo', 'hello']) |
| self.assertEquals(output, 'hello\n') |
| |
| def testStdErr(self): |
| output = get_output( |
| ['bash', '-c', 'echo hello 1>&2'], include_stderr=True) |
| self.assertEquals(output, 'hello\n') |
| |
| def testNoStdErr(self): |
| output = get_output(['bash', '-c', 'echo hello 1>&2']) |
| self.assertEquals(output, '') |
| |
| def testBadOutput(self): |
| self.assertRaises(subprocess.CalledProcessError, get_output, ['false']) |
| |
| def testOutputAttachedToError(self): |
| """Older versions of CalledProcessError don't attach 'output' to |
| themselves. This test is to ensure that get_output always does.""" |
| try: |
| get_output(['bash', '-c', 'echo hello && false']) |
| except subprocess.CalledProcessError, e: |
| self.assertEquals(e.output, 'hello\n') |
| else: |
| self.fail("get_output did not raise CalledProcessError") |
| |
| def testOutputTimeout(self): |
| try: |
| get_output(['bash', '-c', 'sleep 5'], |
| timeout=1) |
| except subprocess.CalledProcessError, e: |
| self.assertTrue(e.output.startswith(TERMINATED_PROCESS_MSG)) |
| else: |
| self.fail("get_output did not raise CalledProcessError") |
| |
| def testTerminateAndGetOutputGetStdErrorToo(self): |
| text_s = "this is just a test" |
| text_e = "this should go to the stderr" |
| cmd = ['bash', '-c', 'echo "%s" && echo "%s" 1>&2 && sleep 1 && true' % (text_s, text_e)] |
| try: |
| get_output(cmd, include_stderr=True, timeout=0.5) |
| except subprocess.CalledProcessError as error: |
| self.assertTrue(text_s in error.output) |
| self.assertTrue(text_e in error.output) |
| else: |
| self.fail("get_output did not raise CalledProcessError") |
| |
| def testTerminateAndGetOutput(self): |
| text_s = "this is just a test" |
| text_e = "this should go to the stderr" |
| cmd = ['bash', '-c', 'echo "%s" && echo "%s" 1>&2 && sleep 1 && true' % (text_s, text_e)] |
| try: |
| get_output(cmd, timeout=0.5) |
| except subprocess.CalledProcessError as error: |
| self.assertTrue(text_s in error.output) |
| self.assertTrue(text_e not in error.output) |
| else: |
| self.fail("get_output did not raise CalledProcessError") |
| |
| def testInsaneOutputStreamTimeout(self): |
| if not has_urandom(): |
| raise SkipTest |
| text_s = "this is just a test" |
| # create a huge output, check that text_s is in the raised exception |
| # output. |
| # tested with ~400 MB output: decreasing the size of the output to make |
| # tests faster (terminates happens on time, but proc.communicate() can |
| # take few seconds) |
| cmd = ['bash', '-c', 'echo "%s" && dd if=/dev/urandom bs=4096 count=1000 2>&1 && sleep 2 && true' % text_s] |
| try: |
| get_output(cmd, include_stderr=False, timeout=1.0) |
| except subprocess.CalledProcessError as error: |
| self.assertTrue(text_s in error.output) |
| else: |
| self.fail("get_output did not raise CalledProcessError") |
