diff options
Diffstat (limited to 'xtesting')
| -rw-r--r-- | xtesting/core/unit.py | 65 | ||||
| -rw-r--r-- | xtesting/tests/unit/core/test_unit.py | 254 |
2 files changed, 262 insertions, 57 deletions
diff --git a/xtesting/core/unit.py b/xtesting/core/unit.py index 27773679..f874d01f 100644 --- a/xtesting/core/unit.py +++ b/xtesting/core/unit.py @@ -12,9 +12,13 @@ from __future__ import division import logging +import os +import shutil +import subprocess import time import unittest +from subunit.run import SubunitTestRunner import six from xtesting.core import testcase @@ -30,8 +34,46 @@ class Suite(testcase.TestCase): def __init__(self, **kwargs): super(Suite, self).__init__(**kwargs) + self.res_dir = "/var/lib/xtesting/results/{}".format(self.case_name) self.suite = None + @classmethod + def generate_stats(cls, stream): + """Generate stats from subunit stream + + Raises: + Exception + """ + stream.seek(0) + stats = subprocess.Popen( + ['subunit-stats'], stdin=subprocess.PIPE, stdout=subprocess.PIPE) + output, _ = stats.communicate(stream.read()) + cls.__logger.info("\n\n%s", output) + + def generate_xunit(self, stream): + """Generate junit report from subunit stream + + Raises: + Exception + """ + stream.seek(0) + with open("{}/results.xml".format(self.res_dir), "w") as xml: + stats = subprocess.Popen( + ['subunit2junitxml'], stdin=subprocess.PIPE, + stdout=subprocess.PIPE) + output, _ = stats.communicate(stream.read()) + xml.write(output) + + def generate_html(self, stream): + """Generate html report from subunit stream + + Raises: + CalledProcessError + """ + cmd = ['subunit2html', stream, '{}/results.html'.format(self.res_dir)] + output = subprocess.check_output(cmd) + self.__logger.debug("\n%s\n\n%s", ' '.join(cmd), output) + def run(self, **kwargs): """Run the test suite. @@ -53,8 +95,8 @@ class Suite(testcase.TestCase): Args: kwargs: Arbitrary keyword arguments. - Returns: - TestCase.EX_OK if any TestSuite has been run, + Return: + TestCase.EX_OK if any TestSuite has been run TestCase.EX_RUN_ERROR otherwise. """ try: @@ -69,16 +111,22 @@ class Suite(testcase.TestCase): try: assert self.suite self.start_time = time.time() + if not os.path.isdir(self.res_dir): + os.makedirs(self.res_dir) stream = six.StringIO() - result = unittest.TextTestRunner( - stream=stream, verbosity=2).run(self.suite) - self.__logger.debug("\n\n%s", stream.getvalue()) + result = SubunitTestRunner( + stream=stream, verbosity=2).run(self.suite).decorated + self.generate_stats(stream) + self.generate_xunit(stream) + with open('{}/subunit_stream'.format(self.res_dir), 'w') as subfd: + stream.seek(0) + shutil.copyfileobj(stream, subfd) + self.generate_html('{}/subunit_stream'.format(self.res_dir)) self.stop_time = time.time() self.details = { "testsRun": result.testsRun, "failures": len(result.failures), - "errors": len(result.errors), - "stream": stream.getvalue()} + "errors": len(result.errors)} self.result = 100 * ( (result.testsRun - (len(result.failures) + len(result.errors))) / @@ -90,3 +138,6 @@ class Suite(testcase.TestCase): except ZeroDivisionError: self.__logger.error("No test has been run") return testcase.TestCase.EX_RUN_ERROR + except Exception: # pylint: disable=broad-except + self.__logger.exception("something wrong occurs") + return testcase.TestCase.EX_RUN_ERROR diff --git a/xtesting/tests/unit/core/test_unit.py b/xtesting/tests/unit/core/test_unit.py index 8afe0bde..20fd6959 100644 --- a/xtesting/tests/unit/core/test_unit.py +++ b/xtesting/tests/unit/core/test_unit.py @@ -8,67 +8,211 @@ # pylint: disable=missing-docstring import logging +import subprocess import unittest import mock +import six from xtesting.core import unit from xtesting.core import testcase -class PyTestSuiteRunnerTesting(unittest.TestCase): +class SuiteTesting(unittest.TestCase): def setUp(self): - self.psrunner = unit.Suite() + self.psrunner = unit.Suite(case_name="unit") self.psrunner.suite = "foo" + @mock.patch('subprocess.Popen', side_effect=Exception) + def test_generate_stats_ko(self, *args): + stream = six.StringIO() + with self.assertRaises(Exception): + self.psrunner.generate_stats(stream) + args[0].assert_called_once_with( + ['subunit-stats'], stdin=subprocess.PIPE, stdout=subprocess.PIPE) + + @mock.patch('subprocess.Popen', + return_value=mock.Mock( + communicate=mock.Mock(return_value=("foo", "bar")))) + def test_generate_stats_ok(self, *args): + stream = six.StringIO() + self.psrunner.generate_stats(stream) + args[0].assert_called_once_with( + ['subunit-stats'], stdin=subprocess.PIPE, stdout=subprocess.PIPE) + + @mock.patch('six.moves.builtins.open', mock.mock_open()) + @mock.patch('subprocess.Popen', side_effect=Exception) + def test_generate_xunit_ko(self, *args): + stream = six.StringIO() + with self.assertRaises(Exception), \ + mock.patch('six.moves.builtins.open', + mock.mock_open()) as mock_open: + self.psrunner.generate_xunit(stream) + args[0].assert_called_once_with( + ['subunit2junitxml'], stdin=subprocess.PIPE, + stdout=subprocess.PIPE) + mock_open.assert_called_once_with( + '{}/results.xml'.format(self.psrunner.res_dir), 'w') + + @mock.patch('subprocess.Popen', + return_value=mock.Mock( + communicate=mock.Mock(return_value=("foo", "bar")))) + def test_generate_xunit_ok(self, *args): + stream = six.StringIO() + with mock.patch('six.moves.builtins.open', + mock.mock_open()) as mock_open: + self.psrunner.generate_xunit(stream) + args[0].assert_called_once_with( + ['subunit2junitxml'], stdin=subprocess.PIPE, + stdout=subprocess.PIPE) + mock_open.assert_called_once_with( + '{}/results.xml'.format(self.psrunner.res_dir), 'w') + + @mock.patch('subprocess.check_output', side_effect=Exception) + def test_generate_html_ko(self, *args): + stream = "foo" + with self.assertRaises(Exception): + self.psrunner.generate_html(stream) + args[0].assert_called_once_with( + ['subunit2html', stream, + '{}/results.html'.format(self.psrunner.res_dir)]) + + @mock.patch('subprocess.check_output') + def test_generate_html_ok(self, *args): + stream = "foo" + self.psrunner.generate_html(stream) + args[0].assert_called_once_with( + ['subunit2html', stream, + '{}/results.html'.format(self.psrunner.res_dir)]) + + @mock.patch('xtesting.core.unit.Suite.generate_html') + @mock.patch('xtesting.core.unit.Suite.generate_xunit') + @mock.patch('xtesting.core.unit.Suite.generate_stats') @mock.patch('unittest.TestLoader') - def _test_run(self, mock_class=None, result=mock.Mock(), - status=testcase.TestCase.EX_OK): - with mock.patch('xtesting.core.unit.unittest.TextTestRunner.run', - return_value=result): + @mock.patch('subunit.run.SubunitTestRunner.run') + def _test_run(self, mock_result, status, result, *args): + args[0].return_value = mock_result + with mock.patch('six.moves.builtins.open', mock.mock_open()) as m_open: self.assertEqual(self.psrunner.run(), status) - mock_class.assert_not_called() + m_open.assert_called_once_with( + '{}/subunit_stream'.format(self.psrunner.res_dir), 'w') + self.assertEqual(self.psrunner.is_successful(), result) + args[0].assert_called_once_with(self.psrunner.suite) + args[1].assert_not_called() + args[2].assert_called_once_with(mock.ANY) + args[3].assert_called_once_with(mock.ANY) + args[4].assert_called_once_with( + '{}/subunit_stream'.format(self.psrunner.res_dir)) + + @mock.patch('xtesting.core.unit.Suite.generate_html') + @mock.patch('xtesting.core.unit.Suite.generate_xunit') + @mock.patch('xtesting.core.unit.Suite.generate_stats') + @mock.patch('unittest.TestLoader') + @mock.patch('subunit.run.SubunitTestRunner.run') + def _test_run_name(self, name, mock_result, status, result, *args): + args[0].return_value = mock_result + with mock.patch('six.moves.builtins.open', mock.mock_open()) as m_open: + self.assertEqual(self.psrunner.run(name=name), status) + m_open.assert_called_once_with( + '{}/subunit_stream'.format(self.psrunner.res_dir), 'w') + self.assertEqual(self.psrunner.is_successful(), result) + args[0].assert_called_once_with(self.psrunner.suite) + args[1].assert_called_once_with() + args[2].assert_called_once_with(mock.ANY) + args[3].assert_called_once_with(mock.ANY) + args[4].assert_called_once_with( + '{}/subunit_stream'.format(self.psrunner.res_dir)) + + @mock.patch('xtesting.core.unit.Suite.generate_html') + @mock.patch('xtesting.core.unit.Suite.generate_xunit') + @mock.patch('xtesting.core.unit.Suite.generate_stats') + @mock.patch('unittest.TestLoader') + @mock.patch('subunit.run.SubunitTestRunner.run') + @mock.patch('os.path.isdir', return_value=True) + def _test_run_exc(self, exc, *args): + args[1].return_value = mock.Mock( + decorated=mock.Mock( + testsRun=50, errors=[], failures=[])) + args[3].side_effect = exc + with mock.patch('six.moves.builtins.open', + mock.mock_open()) as m_open: + self.assertEqual( + self.psrunner.run(), testcase.TestCase.EX_RUN_ERROR) + m_open.assert_not_called() + self.assertEqual( + self.psrunner.is_successful(), + testcase.TestCase.EX_TESTCASE_FAILED) + args[0].assert_called_once_with(self.psrunner.res_dir) + args[1].assert_called_once_with(self.psrunner.suite) + args[2].assert_not_called() + args[3].assert_called_once_with(mock.ANY) + args[4].assert_not_called() + args[5].assert_not_called() def test_check_suite_null(self): self.assertEqual(unit.Suite().suite, None) self.psrunner.suite = None - self._test_run(result=mock.Mock(), - status=testcase.TestCase.EX_RUN_ERROR) - - def test_run_no_ut(self): - mock_result = mock.Mock(testsRun=0, errors=[], failures=[]) - self._test_run(result=mock_result, - status=testcase.TestCase.EX_RUN_ERROR) + self.assertEqual(self.psrunner.run(), testcase.TestCase.EX_RUN_ERROR) + + @mock.patch('os.path.isdir', return_value=True) + def test_run_no_ut(self, *args): + mock_result = mock.Mock( + decorated=mock.Mock(testsRun=0, errors=[], failures=[])) + self._test_run( + mock_result, testcase.TestCase.EX_RUN_ERROR, + testcase.TestCase.EX_TESTCASE_FAILED) self.assertEqual(self.psrunner.result, 0) - self.assertEqual(self.psrunner.details, - {'errors': 0, 'failures': 0, 'stream': '', - 'testsRun': 0}) - self.assertEqual(self.psrunner.is_successful(), - testcase.TestCase.EX_TESTCASE_FAILED) + self.assertEqual( + self.psrunner.details, + {'errors': 0, 'failures': 0, 'testsRun': 0}) + args[0].assert_called_once_with(self.psrunner.res_dir) - def test_run_result_ko(self): + @mock.patch('os.path.isdir', return_value=True) + def test_run_result_ko(self, *args): self.psrunner.criteria = 100 - mock_result = mock.Mock(testsRun=50, errors=[('test1', 'error_msg1')], - failures=[('test2', 'failure_msg1')]) - self._test_run(result=mock_result) + mock_result = mock.Mock( + decorated=mock.Mock( + testsRun=50, errors=[('test1', 'error_msg1')], + failures=[('test2', 'failure_msg1')])) + self._test_run( + mock_result, testcase.TestCase.EX_OK, + testcase.TestCase.EX_TESTCASE_FAILED) self.assertEqual(self.psrunner.result, 96) - self.assertEqual(self.psrunner.details, - {'errors': 1, 'failures': 1, 'stream': '', - 'testsRun': 50}) - self.assertEqual(self.psrunner.is_successful(), - testcase.TestCase.EX_TESTCASE_FAILED) - - def test_run_result_ok(self): - mock_result = mock.Mock(testsRun=50, errors=[], - failures=[]) - self._test_run(result=mock_result) + self.assertEqual( + self.psrunner.details, + {'errors': 1, 'failures': 1, 'testsRun': 50}) + args[0].assert_called_once_with(self.psrunner.res_dir) + + @mock.patch('os.path.isdir', return_value=True) + def test_run_result_ok_1(self, *args): + mock_result = mock.Mock( + decorated=mock.Mock( + testsRun=50, errors=[], failures=[])) + self._test_run( + mock_result, testcase.TestCase.EX_OK, + testcase.TestCase.EX_OK) self.assertEqual(self.psrunner.result, 100) - self.assertEqual(self.psrunner.details, - {'errors': 0, 'failures': 0, 'stream': '', - 'testsRun': 50}) - self.assertEqual(self.psrunner.is_successful(), - testcase.TestCase.EX_OK) + self.assertEqual( + self.psrunner.details, + {'errors': 0, 'failures': 0, 'testsRun': 50}) + args[0].assert_called_once_with(self.psrunner.res_dir) + + @mock.patch('os.makedirs') + @mock.patch('os.path.isdir', return_value=False) + def test_run_result_ok_2(self, *args): + mock_result = mock.Mock( + decorated=mock.Mock( + testsRun=50, errors=[], failures=[])) + self._test_run( + mock_result, testcase.TestCase.EX_OK, + testcase.TestCase.EX_OK) + self.assertEqual(self.psrunner.result, 100) + self.assertEqual( + self.psrunner.details, + {'errors': 0, 'failures': 0, 'testsRun': 50}) + args[0].assert_called_once_with(self.psrunner.res_dir) + args[1].assert_called_once_with(self.psrunner.res_dir) @mock.patch('unittest.TestLoader') def test_run_name_exc(self, mock_class=None): @@ -79,18 +223,28 @@ class PyTestSuiteRunnerTesting(unittest.TestCase): mock_class.assert_called_once_with() mock_obj.assert_called_once_with() - @mock.patch('unittest.TestLoader') - def test_run_name(self, mock_class=None): - mock_result = mock.Mock(testsRun=50, errors=[], - failures=[]) - mock_obj = mock.Mock() - mock_class.side_effect = mock_obj - with mock.patch('xtesting.core.unit.unittest.TextTestRunner.run', - return_value=mock_result): - self.assertEqual(self.psrunner.run(name='foo'), - testcase.TestCase.EX_OK) - mock_class.assert_called_once_with() - mock_obj.assert_called_once_with() + @mock.patch('os.path.isdir', return_value=True) + def test_run_name(self, *args): + mock_result = mock.Mock( + decorated=mock.Mock( + testsRun=50, errors=[], failures=[])) + self._test_run_name( + "foo", mock_result, testcase.TestCase.EX_OK, + testcase.TestCase.EX_OK) + self.assertEqual(self.psrunner.result, 100) + self.assertEqual( + self.psrunner.details, + {'errors': 0, 'failures': 0, 'testsRun': 50}) + args[0].assert_called_once_with(self.psrunner.res_dir) + + def test_run_exc1(self): + self._test_run_exc(AssertionError) + + def test_run_exc2(self): + self._test_run_exc(ZeroDivisionError) + + def test_run_exc3(self): + self._test_run_exc(Exception) if __name__ == "__main__": |