Add infrastucture for asserting a series of related values within a single test.

tests/support.py → tests/support/__init__.py

@@ -3,7 +3,6 @@
 #
 # --- 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.
@@ -27,22 +26,18 @@
 
 """Supporting functions for the unit test framework"""
 
-from collections import defaultdict
 import difflib
 import errno
 import io
 import logging
 import os
-import re
 import shutil
 import stat
 import sys
 from unittest import TestCase, main as testmain
 
-TEST_BASE = os.path.dirname(__file__)
-TEST_FIXTURE_DIR = os.path.join(TEST_BASE, "fixture")
-TEST_OUTPUT_DIR = os.path.join(TEST_BASE, "output")
-MIG_BASE = os.path.realpath(os.path.join(TEST_BASE, ".."))
+from tests.support.suppconst import MIG_BASE, \
+    TEST_BASE, TEST_FIXTURE_DIR, TEST_OUTPUT_DIR
 PY2 = sys.version_info[0] == 2
 
 # force defaults to a local environment
@@ -63,6 +58,13 @@
         shutil.rmtree(TEST_OUTPUT_DIR)
         os.mkdir(TEST_OUTPUT_DIR)
 
+# 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
+
+
 # Basic global logging configuration for testing
 
 
@@ -80,111 +82,6 @@ def write(self, message):
 logging.captureWarnings(True)
 
 
-class FakeLogger:
-    """An output capturing logger suitable for being passed to the
-    majority of MiG code by presenting an API compatible interface
-    with the common logger module.
-
-    An instance of this class is made available to test cases which
-    can pass it down into function calls and subsequenently make
-    assertions against any output strings hat were recorded during
-    execution while also avoiding noise hitting the console.
-    """
-
-    RE_UNCLOSEDFILE = re.compile(
-        'unclosed file <.*? name=\'(?P<location>.*?)\'( .*?)?>')
-
-    def __init__(self):
-        self.channels_dict = defaultdict(list)
-        self.forgive_by_channel = defaultdict(lambda: False)
-        self.unclosed_by_file = defaultdict(list)
-
-    def _append_as(self, channel, line):
-        self.channels_dict[channel].append(line)
-
-    def check_empty_and_reset(self):
-        channels_dict = self.channels_dict
-        forgive_by_channel = self.forgive_by_channel
-        unclosed_by_file = self.unclosed_by_file
-
-        # reset the record of any logged messages
-        self.channels_dict = defaultdict(list)
-        self.forgive_by_channel = defaultdict(lambda: False)
-        self.unclosed_by_file = defaultdict(list)
-
-        # complain loudly (and in detail) in the case of unclosed files
-        if len(unclosed_by_file) > 0:
-            messages = '\n'.join({' --> %s: line=%s, file=%s' % (fname, lineno, outname)
-                                  for fname, (lineno, outname) in unclosed_by_file.items()})
-            raise RuntimeError('unclosed files encountered:\n%s' % (messages,))
-
-        if channels_dict['error'] and not forgive_by_channel['error']:
-            raise RuntimeError('errors reported to logger:\n%s' % '\n'.join(channels_dict['error']))
-
-
-    def forgive_errors(self):
-        self.forgive_by_channel['error'] = True
-
-    # logger interface
-
-    def debug(self, line):
-        self._append_as('debug', line)
-
-    def error(self, line):
-        self._append_as('error', line)
-
-    def info(self, line):
-        self._append_as('info', line)
-
-    def warning(self, line):
-        self._append_as('warning', line)
-
-    def write(self, message):
-        channel, namespace, specifics = message.split(':', 2)
-
-        # ignore everything except warnings sent by the python runtime
-        if not (channel == 'WARNING' and namespace == 'py.warnings'):
-            return
-
-        filename_and_datatuple = FakeLogger.identify_unclosed_file(specifics)
-        if filename_and_datatuple is not None:
-            self.unclosed_by_file.update((filename_and_datatuple,))
-
-    @staticmethod
-    def identify_unclosed_file(specifics):
-        filename, lineno, exc_name, message = specifics.split(':', 3)
-
-        exc_name = exc_name.lstrip()
-        if exc_name != 'ResourceWarning':
-            return
-
-        matched = FakeLogger.RE_UNCLOSEDFILE.match(message.lstrip())
-        if matched is None:
-            return
-
-        relative_testfile = os.path.relpath(filename, start=MIG_BASE)
-        relative_outputfile = os.path.relpath(
-            matched.groups('location')[0], start=TEST_BASE)
-        return (relative_testfile, (lineno, relative_outputfile))
-
-
-class FakeConfiguration:
-    """A simple helper to pretend we have a real Configuration object with any
-    required attributes explicitly passed.
-    Automatically attaches a FakeLogger instance if no logger is provided in
-    kwargs.
-    """
-
-    def __init__(self, **kwargs):
-        """Initialise instance attributes to be any named args provided and a
-        FakeLogger instance attached if not provided.
-        """
-        self.__dict__.update(kwargs)
-        if not 'logger' in self.__dict__:
-            dummy_logger = FakeLogger()
-            self.__dict__.update({'logger': dummy_logger})
-
-
 class MigTestCase(TestCase):
     """Embellished base class for MiG test cases. Provides additional commonly
     used assertions as well as some basics for the standardised and idiomatic
@@ -197,6 +94,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
@@ -211,6 +109,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)
 
@@ -231,6 +134,9 @@ def after_each(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]
@@ -242,6 +148,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):
@@ -303,7 +215,8 @@ def cleanpath(relative_path, test_case, ensure_dir=False):
         try:
             os.mkdir(tmp_path)
         except FileExistsError:
-            raise AssertionError("ABORT: use of unclean output path: %s" % relative_path)
+            raise AssertionError(
+                "ABORT: use of unclean output path: %s" % relative_path)
     test_case._cleanup_paths.add(tmp_path)
     return tmp_path
 

tests/support/assertover.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/configsupp.py

@@ -0,0 +1,46 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+#
+# --- BEGIN_HEADER ---
+#
+# 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 ---
+#
+
+"""Configuration related details within the test support library."""
+
+from tests.support.loggersupp import FakeLogger
+
+
+class FakeConfiguration:
+    """A simple helper to pretend we have a real Configuration object with any
+    required attributes explicitly passed.
+    Automatically attaches a FakeLogger instance if no logger is provided in
+    kwargs.
+    """
+
+    def __init__(self, **kwargs):
+        """Initialise instance attributes to be any named args provided and a
+        FakeLogger instance attached if not provided.
+        """
+        self.__dict__.update(kwargs)
+        if not 'logger' in self.__dict__:
+            dummy_logger = FakeLogger()
+            self.__dict__.update({'logger': dummy_logger})