Added regex convenience to Iris tests AssertLogs().

lib/iris/fileformats/netcdf.py

@@ -52,7 +52,7 @@
 DEBUG = False
 
 # Configure the logger.
-logger = iris.config.get_logger(__name__, fmt="[%(cls)s.%(funcName)s]")
+logger = iris.config.get_logger(__name__)
 
 # Pyke CF related file names.
 _PYKE_RULE_BASE = "fc_rules_cf"

lib/iris/tests/__init__.py

@@ -608,11 +608,15 @@ def assertWarnsRegexp(self, expected_regexp=""):
         self.assertTrue(matches, msg)
 
     @contextlib.contextmanager
-    def assertLogs(self, logger=None, level=None):
+    def assertLogs(self, logger=None, level=None, msg_regex=None):
         """
         An extended version of the usual :meth:`unittest.TestCase.assertLogs`,
         which also exercises the logger's message formatting.
 
+        Also adds the ``msg_regex`` kwarg:
+        If used, check that the result is a single message of the specified
+        level, and that it matches this regex.
+
         The inherited version of this method temporarily *replaces* the logger
         in order to capture log records generated within the context.
         However, in doing so it prevents any messages from being formatted
@@ -632,6 +636,13 @@ def assertLogs(self, logger=None, level=None):
             for handler in assertlogging_context.logger.handlers:
                 handler.format(record)
 
+        # Check message, if requested.
+        if msg_regex:
+            self.assertEqual(len(watcher.records), 1)
+            rec = watcher.records[0]
+            self.assertEqual(level, rec.levelname)
+            self.assertRegex(rec.msg, msg_regex)
+
     @contextlib.contextmanager
     def assertNoWarningsRegexp(self, expected_regexp=""):
         # Check that no warning matching the given expression is raised.

lib/iris/tests/integration/experimental/test_ugrid_load.py

@@ -155,36 +155,26 @@ def create_synthetic_file(self, **create_kwargs):
 
     def test_mesh_bad_topology_dimension(self):
         # Check that the load generates a suitable warning.
-        with self.assertLogs(logger) as log:
+        log_regex = r"topology_dimension.* ignoring"
+        with self.assertLogs(logger, level="WARNING", msg_regex=log_regex):
             template = "minimal_bad_topology_dim"
             dim_line = "mesh_var:topology_dimension = 1 ;"  # which is wrong !
             cube = self.create_synthetic_test_cube(
                 template=template, subs=dict(TOPOLOGY_DIM_DEFINITION=dim_line)
             )
-        # Check we got just one message, and its content
-        self.assertEqual(len(log.records), 1)
-        rec = log.records[0]
-        self.assertEqual(rec.levelname, "WARNING")
-        re_msg = r"topology_dimension.* ignoring"
-        self.assertRegex(rec.msg, re_msg)
 
         # Check that the result has topology-dimension of 2 (not 1).
         self.assertEqual(cube.mesh.topology_dimension, 2)
 
     def test_mesh_no_topology_dimension(self):
         # Check that the load generates a suitable warning.
-        with self.assertLogs(logger) as log:
+        log_regex = r"Mesh variable.* has no 'topology_dimension'"
+        with self.assertLogs(logger, level="WARNING", msg_regex=log_regex):
             template = "minimal_bad_topology_dim"
             dim_line = ""  # don't create ANY topology_dimension property
             cube = self.create_synthetic_test_cube(
                 template=template, subs=dict(TOPOLOGY_DIM_DEFINITION=dim_line)
             )
-        # Check we got just one message, and its content
-        self.assertEqual(len(log.records), 1)
-        rec = log.records[0]
-        self.assertEqual(rec.levelname, "WARNING")
-        re_msg = r"Mesh variable.* has no 'topology_dimension'"
-        self.assertRegex(rec.msg, re_msg)
 
         # Check that the result has the correct topology-dimension value.
         self.assertEqual(cube.mesh.topology_dimension, 2)

lib/iris/tests/unit/experimental/ugrid/test_CFUGridAuxiliaryCoordinateVariable.py

@@ -213,40 +213,26 @@ def test_warn(self):
             "ref_source": ref_source,
         }
 
-        def check_log(expected_levelname, log_regex):
-            rec = log.records[-1]
-            self.assertEqual(expected_levelname, rec.levelname)
-            self.assertRegex(rec.msg, log_regex)
+        # The warn kwarg and expected corresponding log level.
+        warn_and_level = {True: "WARNING", False: "DEBUG"}
 
         # Missing warning.
-        with self.assertLogs(logger, level="DEBUG") as log:
-            log_regex = rf"Missing CF-netCDF auxiliary coordinate variable {subject_name}.*"
-
-            result = CFUGridAuxiliaryCoordinateVariable.identify(
-                vars_all, warn=False
-            )
-            check_log("DEBUG", log_regex)
-            self.assertDictEqual({}, result)
-
-            # Default is warn=True
-            result = CFUGridAuxiliaryCoordinateVariable.identify(vars_all)
-            check_log("WARNING", log_regex)
-            self.assertDictEqual({}, result)
+        log_regex = rf"Missing CF-netCDF auxiliary coordinate variable {subject_name}.*"
+        for warn, level in warn_and_level.items():
+            with self.assertLogs(logger, level=level, msg_regex=log_regex):
+                result = CFUGridAuxiliaryCoordinateVariable.identify(
+                    vars_all, warn=warn
+                )
+                self.assertDictEqual({}, result)
 
         # String variable warning.
-        with self.assertLogs(logger, level="DEBUG") as log:
-            log_regex = r".*is a CF-netCDF label variable.*"
-            vars_all[subject_name] = netcdf_ugrid_variable(
-                subject_name, "", np.bytes_
-            )
-
-            result = CFUGridAuxiliaryCoordinateVariable.identify(
-                vars_all, warn=False
-            )
-            check_log("DEBUG", log_regex)
-            self.assertDictEqual({}, result)
-
-            # Default is warn=True
-            result = CFUGridAuxiliaryCoordinateVariable.identify(vars_all)
-            check_log("WARNING", log_regex)
-            self.assertDictEqual({}, result)
+        log_regex = r".*is a CF-netCDF label variable.*"
+        for warn, level in warn_and_level.items():
+            with self.assertLogs(logger, level=level, msg_regex=log_regex):
+                vars_all[subject_name] = netcdf_ugrid_variable(
+                    subject_name, "", np.bytes_
+                )
+                result = CFUGridAuxiliaryCoordinateVariable.identify(
+                    vars_all, warn=warn
+                )
+                self.assertDictEqual({}, result)

lib/iris/tests/unit/experimental/ugrid/test_CFUGridConnectivityVariable.py

@@ -202,38 +202,26 @@ def test_warn(self):
             "ref_source": ref_source,
         }
 
-        def check_log(expected_levelname, log_regex):
-            rec = log.records[-1]
-            self.assertEqual(expected_levelname, rec.levelname)
-            self.assertRegex(rec.msg, log_regex)
+        # The warn kwarg and expected corresponding log level.
+        warn_and_level = {True: "WARNING", False: "DEBUG"}
 
         # Missing warning.
-        with self.assertLogs(logger, level="DEBUG") as log:
-            log_regex = (
-                rf"Missing CF-UGRID connectivity variable {subject_name}.*"
-            )
-
-            result = CFUGridConnectivityVariable.identify(vars_all, warn=False)
-            check_log("DEBUG", log_regex)
-            self.assertDictEqual({}, result)
-
-            # Default is warn=True
-            result = CFUGridConnectivityVariable.identify(vars_all)
-            check_log("WARNING", log_regex)
-            self.assertDictEqual({}, result)
+        log_regex = rf"Missing CF-UGRID connectivity variable {subject_name}.*"
+        for warn, level in warn_and_level.items():
+            with self.assertLogs(logger, level=level, msg_regex=log_regex):
+                result = CFUGridConnectivityVariable.identify(
+                    vars_all, warn=warn
+                )
+                self.assertDictEqual({}, result)
 
         # String variable warning.
-        with self.assertLogs(logger, level="DEBUG") as log:
-            log_regex = r".*is a CF-netCDF label variable.*"
-            vars_all[subject_name] = netcdf_ugrid_variable(
-                subject_name, "", np.bytes_
-            )
-
-            result = CFUGridConnectivityVariable.identify(vars_all, warn=False)
-            check_log("DEBUG", log_regex)
-            self.assertDictEqual({}, result)
-
-            # Default is warn=True
-            result = CFUGridConnectivityVariable.identify(vars_all)
-            check_log("WARNING", log_regex)
-            self.assertDictEqual({}, result)
+        log_regex = r".*is a CF-netCDF label variable.*"
+        for warn, level in warn_and_level.items():
+            with self.assertLogs(logger, level=level, msg_regex=log_regex):
+                vars_all[subject_name] = netcdf_ugrid_variable(
+                    subject_name, "", np.bytes_
+                )
+                result = CFUGridConnectivityVariable.identify(
+                    vars_all, warn=warn
+                )
+                self.assertDictEqual({}, result)

lib/iris/tests/unit/experimental/ugrid/test_CFUGridMeshVariable.py

@@ -195,36 +195,22 @@ def test_warn(self):
             "ref_source": ref_source,
         }
 
-        def check_log(expected_levelname, log_regex):
-            rec = log.records[-1]
-            self.assertEqual(expected_levelname, rec.levelname)
-            self.assertRegex(rec.msg, log_regex)
+        # The warn kwarg and expected corresponding log level.
+        warn_and_level = {True: "WARNING", False: "DEBUG"}
 
         # Missing warning.
-        with self.assertLogs(logger, level="DEBUG") as log:
-            log_regex = rf"Missing CF-UGRID mesh variable {subject_name}.*"
-
-            result = CFUGridMeshVariable.identify(vars_all, warn=False)
-            check_log("DEBUG", log_regex)
-            self.assertDictEqual({}, result)
-
-            # Default is warn=True
-            result = CFUGridMeshVariable.identify(vars_all)
-            check_log("WARNING", log_regex)
-            self.assertDictEqual({}, result)
+        log_regex = rf"Missing CF-UGRID mesh variable {subject_name}.*"
+        for warn, level in warn_and_level.items():
+            with self.assertLogs(logger, level=level, msg_regex=log_regex):
+                result = CFUGridMeshVariable.identify(vars_all, warn=warn)
+                self.assertDictEqual({}, result)
 
         # String variable warning.
-        with self.assertLogs(logger, level="DEBUG") as log:
-            log_regex = r".*is a CF-netCDF label variable.*"
-            vars_all[subject_name] = netcdf_ugrid_variable(
-                subject_name, "", np.bytes_
-            )
-
-            result = CFUGridMeshVariable.identify(vars_all, warn=False)
-            check_log("DEBUG", log_regex)
-            self.assertDictEqual({}, result)
-
-            # Default is warn=True
-            result = CFUGridMeshVariable.identify(vars_all)
-            check_log("WARNING", log_regex)
-            self.assertDictEqual({}, result)
+        log_regex = r".*is a CF-netCDF label variable.*"
+        for warn, level in warn_and_level.items():
+            with self.assertLogs(logger, level=level, msg_regex=log_regex):
+                vars_all[subject_name] = netcdf_ugrid_variable(
+                    subject_name, "", np.bytes_
+                )
+                result = CFUGridMeshVariable.identify(vars_all, warn=warn)
+                self.assertDictEqual({}, result)

lib/iris/tests/unit/experimental/ugrid/test_Mesh.py

@@ -69,11 +69,6 @@ def setUpClass(cls):
             [[0, 1], [1, 2], [2, 0]], cf_role="boundary_node_connectivity"
         )
 
-    def check_log(self, log, expected_levelname, log_regex):
-        rec = log.records[-1]
-        self.assertEqual(expected_levelname, rec.levelname)
-        self.assertRegex(rec.msg, log_regex)
-
 
 class TestProperties1D(TestMeshCommon):
     # Tests that can re-use a single instance for greater efficiency.
@@ -197,9 +192,9 @@ def test_connectivities_locations(self):
         for kwargs in negative_kwargs:
             self.assertEqual([], func(**kwargs))
 
-        with self.assertLogs(ugrid.logger, level="DEBUG") as log:
+        log_regex = r".*filter for non-existent.*"
+        with self.assertLogs(ugrid.logger, level="DEBUG", msg_regex=log_regex):
             self.assertEqual([], func(contains_face=True))
-            self.check_log(log, "DEBUG", r".*filter for non-existent.*")
 
     def test_coord(self):
         # See Mesh.coords tests for thorough coverage of cases.
@@ -269,9 +264,9 @@ def test_coords_locations(self):
             expected = [all_expected[k] for k in expected if k in all_expected]
             self.assertEqual(expected, func(**kwargs))
 
-        with self.assertLogs(ugrid.logger, level="DEBUG") as log:
+        log_regex = r".*filter non-existent.*"
+        with self.assertLogs(ugrid.logger, level="DEBUG", msg_regex=log_regex):
             self.assertEqual([], func(include_faces=True))
-            self.check_log(log, "DEBUG", r".*filter non-existent.*")
 
     def test_edge_dimension(self):
         self.assertEqual(
@@ -623,9 +618,9 @@ def test_add_connectivities_invalid(self):
         )
 
         face_node = self.FACE_NODE
-        with self.assertLogs(ugrid.logger, level="DEBUG") as log:
+        log_regex = r"Not adding connectivity.*"
+        with self.assertLogs(ugrid.logger, level="DEBUG", msg_regex=log_regex):
             self.mesh.add_connectivities(face_node)
-            self.check_log(log, "DEBUG", r"Not adding connectivity.*")
 
     def test_add_coords(self):
         # ADD coords.
@@ -730,9 +725,9 @@ def test_dimension_names(self):
         default = ugrid.Mesh1DNames("Mesh1d_node", "Mesh1d_edge")
         self.assertEqual(default, self.mesh.dimension_names())
 
-        with self.assertLogs(ugrid.logger, level="DEBUG") as log:
+        log_regex = r"Not setting face_dimension.*"
+        with self.assertLogs(ugrid.logger, level="DEBUG", msg_regex=log_regex):
             self.mesh.dimension_names("foo", "bar", "baz")
-            self.check_log(log, "DEBUG", r"Not setting face_dimension.*")
         self.assertEqual(
             ugrid.Mesh1DNames("foo", "bar"), self.mesh.dimension_names()
         )
@@ -751,9 +746,9 @@ def test_edge_dimension_set(self):
         self.assertEqual("foo", self.mesh.edge_dimension)
 
     def test_face_dimension_set(self):
-        with self.assertLogs(ugrid.logger, level="DEBUG") as log:
+        log_regex = r"Not setting face_dimension.*"
+        with self.assertLogs(ugrid.logger, level="DEBUG", msg_regex=log_regex):
             self.mesh.face_dimension = "foo"
-            self.check_log(log, "DEBUG", r"Not setting face_dimension.*")
         self.assertIsNone(self.mesh.face_dimension)
 
     def test_node_dimension_set(self):
@@ -795,10 +790,12 @@ def test_remove_connectivities(self):
             {"contains_edge": False, "contains_node": False},
         )
 
+        log_regex = r"Ignoring request to remove.*"
         for kwargs in positive_kwargs:
-            with self.assertLogs(ugrid.logger, level="DEBUG") as log:
+            with self.assertLogs(
+                ugrid.logger, level="DEBUG", msg_regex=log_regex
+            ):
                 self.mesh.remove_connectivities(**kwargs)
-                self.check_log(log, "DEBUG", r"Ignoring request to remove.*")
             self.assertEqual(self.EDGE_NODE, self.mesh.edge_node_connectivity)
         for kwargs in negative_kwargs:
             with self.assertLogs(ugrid.logger, level="DEBUG") as log:
@@ -830,10 +827,12 @@ def test_remove_coords(self):
             {"attributes": {"test": 2}},
         )
 
+        log_regex = r"Ignoring request to remove.*"
         for kwargs in positive_kwargs:
-            with self.assertLogs(ugrid.logger, level="DEBUG") as log:
+            with self.assertLogs(
+                ugrid.logger, level="DEBUG", msg_regex=log_regex
+            ):
                 self.mesh.remove_coords(**kwargs)
-                self.check_log(log, "DEBUG", r"Ignoring request to remove.*")
             self.assertEqual(self.NODE_LON, self.mesh.node_coords.node_x)
         for kwargs in negative_kwargs:
             with self.assertLogs(ugrid.logger, level="DEBUG") as log:
@@ -950,9 +949,9 @@ def test_add_connectivities_invalid(self):
         fake_cf_role = tests.mock.Mock(
             __class__=ugrid.Connectivity, cf_role="foo"
         )
-        with self.assertLogs(ugrid.logger, level="DEBUG") as log:
+        log_regex = r"Not adding connectivity.*"
+        with self.assertLogs(ugrid.logger, level="DEBUG", msg_regex=log_regex):
             self.mesh.add_connectivities(fake_cf_role)
-            self.check_log(log, "DEBUG", r"Not adding connectivity.*")
 
     def test_add_coords_face(self):
         # ADD coords.

lib/iris/tests/unit/fileformats/netcdf/test_load_cubes.py

@@ -314,9 +314,6 @@ def test_missing_mesh(self):
         _ = list(load_cubes(nc_path))
 
         with PARSE_UGRID_ON_LOAD.context():
-            with self.assertLogs(logger, level="DEBUG") as log:
+            log_regex = r"File does not contain mesh.*"
+            with self.assertLogs(logger, level="DEBUG", msg_regex=log_regex):
                 _ = list(load_cubes(nc_path))
-                rec = log.records[0]
-                self.assertEqual("DEBUG", rec.levelname)
-                re_msg = r"File does not contain mesh.*"
-                self.assertRegex(rec.msg, re_msg)