Merge remote-tracking branch 'origin/master' into edge

Author: jonasbardino

tests/support/__init__.py

@@ -62,6 +62,7 @@
 
 # Exports to expose at the top level from the support library.
 
+from tests.support.assertover import AssertOver
 from tests.support.configsupp import FakeConfiguration
 from tests.support.loggersupp import FakeLogger
 
@@ -96,6 +97,7 @@ class MigTestCase(TestCase):
 
     def __init__(self, *args):
         super(MigTestCase, self).__init__(*args)
+        self._cleanup_checks = list()
         self._cleanup_paths = set()
         self._logger = None
         self._skip_logging = False
@@ -112,6 +114,11 @@ def tearDown(self):
 
         if not self._skip_logging:
             self._logger.check_empty_and_reset()
+
+        for check_callable in self._cleanup_checks:
+            check_callable.__call__()
+        self._cleanup_checks = list()
+
         if self._logger is not None:
             self._reset_logging(stream=BLACKHOLE_STREAM)
 
@@ -134,6 +141,9 @@ def before_each(self):
         """Before each test action hook"""
         pass
 
+    def _register_check(self, check_callable):
+        self._cleanup_checks.append(check_callable)
+
     def _reset_logging(self, stream):
         root_logger = logging.getLogger()
         root_handler = root_logger.handlers[0]
@@ -146,6 +156,12 @@ def logger(self):
             self._logger = FakeLogger()
         return self._logger
 
+    def assert_over(self, values=None, _AssertOver=AssertOver):
+        assert_over = _AssertOver(values=values, testcase=self)
+        check_callable = assert_over.to_check_callable()
+        self._register_check(check_callable)
+        return assert_over
+
     # custom assertions available for common use
 
     def assertFileContentIdentical(self, file_actual, file_expected):

tests/support/assertover.py

@@ -0,0 +1,106 @@
+# -*- coding: utf-8 -*-
+#
+# --- BEGIN_HEADER ---
+#
+# assertover - iterating assert helper functions for unit testing
+# Copyright (C) 2003-2024  The MiG Project by the Science HPC Center at UCPH
+#
+# This file is part of MiG.
+#
+# MiG is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# MiG is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+#
+# -- END_HEADER ---
+#
+
+"""Infrastruture to support assertion over a range of values"""
+
+
+class NoBlockError(AssertionError):
+    """Decorate AssertionError for our own convenience"""
+    pass
+
+
+class NoCasesError(AssertionError):
+    """Decorate AssertionError for our own convenience"""
+    pass
+
+
+class AssertOver:
+    """Iteration helper to repeat the same overall assertion structure for a
+    given sequence of test input.
+    """
+
+    def __init__(self, values=None, testcase=None):
+        self._attempts = None
+        self._consulted = False
+        self._ended = False
+        self._started = False
+        self._testcase = testcase
+        self._values = iter(values)
+
+    def __call__(self, block):
+        self._attempts = []
+
+        try:
+            while True:
+                block_value = next(self._values)
+                attempt_info = self._execute_block(block, block_value)
+                self.record_attempt(attempt_info)
+        except StopIteration:
+            pass
+
+        self._ended = True
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        if self._attempts is None:
+            raise NoBlockError()
+
+        if len(self._attempts) == 0:
+            raise NoCasesError()
+
+        if not any(self._attempts):
+            return True
+
+        value_lines = ["- <%r> : %s" % (attempt[0], str(attempt[1])) for
+                       attempt in self._attempts if attempt]
+        raise AssertionError("assertions raised for the following values:\n%s"
+                             % '\n'.join(value_lines))
+
+    def record_attempt(self, attempt_info):
+        """Record the result of a test attempt"""
+        return self._attempts.append(attempt_info)
+
+    def to_check_callable(self):
+        def raise_unless_consulted():
+            if not self._consulted:
+                raise AssertionError(
+                    "no examiniation made of assertion of multiple values")
+        return raise_unless_consulted
+
+    def assert_success(self):
+        """Assertion with registration in consulted"""
+        self._consulted = True
+        assert not any(self._attempts)
+
+    @classmethod
+    def _execute_block(cls, block, block_value):
+        try:
+            block.__call__(block_value)
+            return None
+        except Exception as blockexc:
+            return (block_value, blockexc,)

tests/test_support.py

@@ -3,10 +3,42 @@
 import sys
 import unittest
 
-from tests.support import MigTestCase, PY2, testmain, temppath
+from tests.support import MigTestCase, PY2, testmain, temppath, \
+                    AssertOver
+
+
+class InstrumentedAssertOver(AssertOver):
+    def __init__(self, *args, **kwargs):
+        AssertOver.__init__(self, *args, **kwargs)
+        self._check_callable = None
+        self._check_callable_called = False
+
+    def get_check_callable(self):
+        return self._check_callable
+
+    def has_check_callable(self):
+        return self._check_callable is not None
+
+    def was_check_callable_called(self):
+        return self._check_callable_called
+
+    def to_check_callable(self):
+        _check_callable = AssertOver.to_check_callable(self)
+        def _wrapped_check_callable():
+            self._check_callable_called = True
+            _check_callable()
+        self._check_callable = _wrapped_check_callable
+        return _wrapped_check_callable
 
 
 class SupportTestCase(MigTestCase):
+    def _class_attribute(self, name, **kwargs):
+        cls = type(self)
+        if 'value' in kwargs:
+            setattr(cls, name, kwargs['value'])
+        else:
+            return getattr(cls, name, None)
+
     @unittest.skipIf(PY2, "Python 3 only")
     def test_unclosed_files_are_recorded(self):
         tmp_path = temppath("support-unclosed", self)
@@ -31,6 +63,30 @@ def test_unclosed_files_are_reset(self):
         except:
             self.assertTrue(False, "should not be reachable")
 
+    def test_when_asserting_over_multiple_values(self):
+        def assert_is_int(value):
+            assert isinstance(value, int)
+
+        attempt_wrapper = self.assert_over(values=(1, 2, 3), _AssertOver=InstrumentedAssertOver)
+
+        # record the wrapper on the test case so the subsequent test can assert against it
+        self._class_attribute('surviving_attempt_wrapper', value=attempt_wrapper)
+
+        with attempt_wrapper as attempt:
+            attempt(assert_is_int)
+        attempt_wrapper.assert_success()
+
+        self.assertTrue(attempt_wrapper.has_check_callable())
+        # cleanup was recorded
+        self.assertIn(attempt_wrapper.get_check_callable(), self._cleanup_checks)
+
+    def test_when_asserting_over_multiple_values_after(self):
+        # test name is purposefully after ..._recorded in sort order
+        # such that we can check the check function was called correctly
+
+        attempt_wrapper = self._class_attribute('surviving_attempt_wrapper')
+        self.assertTrue(attempt_wrapper.was_check_callable_called())
+
 
 if __name__ == '__main__':
     testmain()

tests/test_tests_support_assertover.py

@@ -0,0 +1,73 @@
+# -*- coding: utf-8 -*-
+#
+# --- BEGIN_HEADER ---
+#
+# test_tests_support_assertover - unit test of the corresponding tests module
+# Copyright (C) 2003-2024  The MiG Project by the Science HPC Center at UCPH
+#
+# This file is part of MiG.
+#
+# MiG is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# MiG is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301,
+# USA.
+#
+# --- END_HEADER ---
+#
+
+"""Unit tests for the tests module pointed to in the filename"""
+
+import unittest
+
+from tests.support import AssertOver
+from tests.support.assertover import NoBlockError, NoCasesError
+
+
+def assert_a_thing(value):
+    """A simple assert helper to test with"""
+    assert value.endswith(' thing'), "must end with a thing"
+
+
+class TestsSupportAssertOver(unittest.TestCase):
+    """Coverage of AssertOver helper"""
+
+    def test_none_failing(self):
+        saw_raise = False
+        try:
+            with AssertOver(values=('some thing', 'other thing')) as value_block:
+                value_block(lambda _: assert_a_thing(_))
+        except Exception as exc:
+            saw_raise = True
+        self.assertFalse(saw_raise)
+
+    def test_three_total_two_failing(self):
+        with self.assertRaises(AssertionError) as raised:
+            with AssertOver(values=('some thing', 'other stuff', 'foobar')) as value_block:
+                value_block(lambda _: assert_a_thing(_))
+
+        theexception = raised.exception
+        self.assertEqual(str(theexception), """assertions raised for the following values:
+- <'other stuff'> : must end with a thing
+- <'foobar'> : must end with a thing""")
+
+    def test_no_cases(self):
+        with self.assertRaises(AssertionError) as raised:
+            with AssertOver(values=()) as value_block:
+                value_block(lambda _: assert_a_thing(_))
+
+        theexception = raised.exception
+        self.assertIsInstance(theexception, NoCasesError)
+
+
+if __name__ == '__main__':
+    unittest.main()