Add tests of runQuantum exception handling

tests/test_calibrateImage.py

@@ -545,6 +545,54 @@ def test_lintConnections(self):
         Connections = CalibrateImageTask.ConfigClass.ConnectionsClass
         lsst.pipe.base.testUtils.lintConnections(Connections)
 
+    def test_runQuantum_exception(self):
+        """Test handling of exceptions handling in runQuantum.
+        """
+        task = CalibrateImageTask()
+        lsst.pipe.base.testUtils.assertValidInitOutput(task)
+
+        quantum = lsst.pipe.base.testUtils.makeQuantum(
+            task, self.butler, self.visit_id,
+            {"exposures": [self.exposure0_id],
+             "astrometry_ref_cat": [self.htm_id],
+             "photometry_ref_cat": [self.htm_id],
+             # outputs
+             "exposure": self.visit_id,
+             "stars": self.visit_id,
+             "background": self.visit_id,
+             "psf_stars": self.visit_id,
+             "applied_photo_calib": self.visit_id,
+             "initial_pvi_background": self.visit_id,
+             "astrometry_matches": self.visit_id,
+             "photometry_matches": self.visit_id,
+             })
+
+        # A generic exception should raise directly.
+        msg = "mocked run exception"
+        with (
+            mock.patch.object(task, "run", side_effect=ValueError(msg)),
+            self.assertRaisesRegex(ValueError, "mocked run exception")
+        ):
+            lsst.pipe.base.testUtils.runTestQuantum(task, self.butler, quantum, mockRun=False)
+
+        class TestError(lsst.pipe.base.RepeatableQuantumError):
+            @property
+            def metadata(self):
+                return {"something": 12345}
+
+        msg = "mocked run exception"
+        with (
+            mock.patch.object(task, "run", side_effect=TestError(msg)),
+            self.assertRaisesRegex(ValueError, "mocked run exception"),
+            lsst.log.UsePythonLogging(),  # so that assertLogs works with lsst.log
+        ):
+            with self.assertLogs("lsst.calibrateImage", level="WARNING") as cm:
+                lsst.pipe.base.testUtils.runTestQuantum(task,
+                                                        self.butler,
+                                                        quantum,
+                                                        mockRun=False)
+        self.assertIn(msg, "\n".join(cm.output))
+        self.assertIn("writing available datasets", "\n".join(cm.output))
 
 def setup_module(module):
     lsst.utils.tests.init()