aboutsummaryrefslogtreecommitdiffstats
path: root/xtesting
diff options
context:
space:
mode:
Diffstat (limited to 'xtesting')
-rw-r--r--xtesting/core/unit.py65
-rw-r--r--xtesting/tests/unit/core/test_unit.py254
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__":