Provide infrastructure for assertion of related values as a single case.

In out desired use of the test suite is a requirement to allow writing
a test that is the same assertion agianst a series of related values.
As an example, imagine an operation on chunks and wishing to assert
multiple sizes.

This is obviously something to be extremely careful in promoting given
it is in a fair amount of conflict with the notion of separate test
cases, which experience says more "generally" leads to better outcomes.
However, used judiciously this can be useful, and this is far better
supported by a mechanism designed to support it: namely, that _all_
failures are reported up front rather than the first failing assertion
cancelling the test block and the ending up in a fix-and-rerun cycle.

Add a facility and associated assertion to the test support library
that allows executing an operation and its assertion against a series
of values, buffers any exceptions that occurs and enforce that either
a no exception or all exception outcome is asserted by end of test.

Include a mechanism to the support library so arbitrary objects can do
checks and use this in multi-value assertions to ensure a result is
validated by its containing case.

Author: albu-diku

tests/_support/__init__.py

@@ -0,0 +1,93 @@
+# -*- coding: utf-8 -*-
+#
+# --- BEGIN_HEADER ---
+#
+# support - 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):
+    pass
+
+class NoCasesError(AssertionError):
+    pass
+
+class AssertOver:
+    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):
+        return self._attempts.append(attempt_info)
+
+    def to_check_callable(self):
+        def raise_unless_consuted():
+            if not self._consulted:
+                raise AssertionError("no examiniation made of assertion of multiple values")
+        return raise_unless_consuted
+
+    def assert_success(self):
+        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/_support/assertover.py

@@ -63,6 +63,10 @@
         shutil.rmtree(TEST_OUTPUT_DIR)
         os.mkdir(TEST_OUTPUT_DIR)
 
+
+from tests._support.assertover import AssertOver
+
+
 # Basic global logging configuration for testing
 
 
@@ -184,6 +188,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
@@ -196,6 +201,11 @@ def setUp(self):
     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)
 
@@ -213,6 +223,9 @@ def tearDown(self):
     def before_each(self):
         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]
@@ -224,6 +237,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.py

@@ -3,10 +3,42 @@
 import sys
 import unittest
 
-from support import MigTestCase, PY2, testmain, temppath
+from 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_support.py

@@ -0,0 +1,41 @@
+import unittest
+
+from tests.support import AssertOver
+from tests._support.assertover import NoBlockError, NoCasesError
+
+
+def assert_a_thing(value):
+    assert value.endswith(' thing'), "must end with a thing"
+
+
+class TestsSupportAssertOver(unittest.TestCase):
+    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()