Merge branch 'tickets/DM-47829'

Author: kfindeisen

python/activator/middleware_interface.py

@@ -1367,6 +1367,9 @@ def _check_permanent_changes(self, where: str) -> bool:
             data_id = data_ids.pop()
             apdb = lsst.dax.apdb.Apdb.from_uri(self._apdb_config)
             return apdb.containsVisitDetector(data_id["visit"], self.visit.detector)
+        elif not data_ids:
+            # Engineering exposures don't produce visits, but they also can't write to the APDB.
+            return False
         else:
             # Don't know how this could happen, so won't try to handle it gracefully.
             _log.warning("Unexpected visit ids: %s. Assuming APDB modified.", data_ids)

tests/test_middleware_interface.py

@@ -51,7 +51,7 @@
 
 from activator.caching import DatasetCache
 from activator.config import PipelinesConfig
-from activator.exception import NonRetriableError
+from activator.exception import NonRetriableError, NoGoodPipelinesError, PipelineExecutionError
 from activator.visit import FannedOutVisit
 from activator.middleware_interface import get_central_butler, flush_local_repo, make_local_repo, \
     _get_sasquatch_dispatcher, MiddlewareInterface, \
@@ -71,6 +71,9 @@
                                             "${PROMPT_PROCESSING_DIR}/tests/data/SingleFrame.yaml",
                                             ],
                               }])
+pipelines_minimal = PipelinesConfig([{"survey": "SURVEY",
+                                      "pipelines": ["${PROMPT_PROCESSING_DIR}/tests/data/ISR.yaml", ],
+                                      }])
 pre_pipelines_empty = PipelinesConfig([{"survey": "SURVEY", "pipelines": None}])
 pre_pipelines_full = PipelinesConfig([{"survey": "SURVEY",
                                        "pipelines": ["${PROMPT_PROCESSING_DIR}/tests/data/Preprocess.yaml",
@@ -132,6 +135,60 @@ def fake_file_data(filename, dimensions, instrument, visit):
     return data_id, file_data
 
 
+# TODO: merge this into fake_file_data after DM-46152
+def fake_eng_data(filename, dimensions, instrument, visit):
+    """Return file data for a mock non-science file to be ingested.
+
+    Parameters
+    ----------
+    filename : `str`
+        Full path to the file to mock. Can be a non-existant file.
+    dimensions : `lsst.daf.butler.DimensionsUniverse`
+        The full set of dimensions for this butler.
+    instrument : `lsst.obs.base.Instrument`
+        The instrument the file is supposed to be from.
+    visit : `FannedOutVisit`
+        Group of snaps from one detector to be processed.
+
+    Returns
+    -------
+    data_id, file_data, : `DataCoordinate`, `RawFileData`
+        The id and descriptor for the mock file.
+    """
+    exposure_id = int(visit.groupId)
+    data_id = DataCoordinate.standardize({"exposure": exposure_id,
+                                          "detector": visit.detector,
+                                          "instrument": instrument.getName()},
+                                         universe=dimensions)
+
+    start_time = astropy.time.Time("2024-06-17T22:06:15", scale="tai")
+    day_obs = 20240617
+    obs_info = astro_metadata_translator.makeObservationInfo(
+        instrument=instrument.getName(),
+        datetime_begin=start_time,
+        datetime_end=start_time + 30*u.second,
+        exposure_id=exposure_id,
+        exposure_group=visit.groupId,
+        visit_id=None,
+        boresight_rotation_angle=None,
+        boresight_rotation_coord=None,
+        tracking_radec=None,
+        observation_id=visit.groupId,
+        physical_filter=filter,
+        exposure_time=30.0*u.second,
+        observation_type="goofing off",
+        observing_day=day_obs,
+        group_counter_start=exposure_id,
+        group_counter_end=exposure_id,
+    )
+    dataset_info = RawFileDatasetInfo(data_id, obs_info)
+    file_data = RawFileData([dataset_info],
+                            lsst.resources.ResourcePath(filename),
+                            FitsImageFormatter,
+                            instrument)
+    return data_id, file_data
+
+
 class MiddlewareInterfaceTest(unittest.TestCase):
     """Test the MiddlewareInterface class with faked data.
     """
@@ -360,7 +417,7 @@ def test_prep_butler_nofilter(self):
     def test_prep_butler_notemplates(self):
         """Test that prep_butler can handle pipeline configs without templates.
         """
-        self.interface.main_pipelines = pre_pipelines_empty
+        self.interface.main_pipelines = pipelines_minimal
         with unittest.mock.patch("activator.middleware_interface.MiddlewareInterface._run_preprocessing") \
                 as mock_pre, \
                 self.assertNoLogs(level="ERROR"):
@@ -475,10 +532,10 @@ def test_ingest_image_fails_missing_file(self):
         self.interface.prep_butler()  # Ensure raw collections exist.
         filename = "nonexistentImage.fits"
         filepath = os.path.join(self.input_data, filename)
-        data_id, file_data = fake_file_data(filepath,
-                                            self.interface.butler.dimensions,
-                                            self.interface.instrument,
-                                            self.next_visit)
+        _, file_data = fake_file_data(filepath,
+                                      self.interface.butler.dimensions,
+                                      self.interface.instrument,
+                                      self.next_visit)
         with unittest.mock.patch.object(self.interface.rawIngestTask, "extractMetadata") as mock, \
                 self.assertRaisesRegex(FileNotFoundError, "Resource at .* does not exist"):
             mock.return_value = file_data
@@ -498,14 +555,29 @@ def _prepare_run_pipeline(self):
 
         filename = "fakeRawImage.fits"
         filepath = os.path.join(self.input_data, filename)
-        data_id, file_data = fake_file_data(filepath,
-                                            self.interface.butler.dimensions,
-                                            self.interface.instrument,
-                                            self.next_visit)
+        _, file_data = fake_file_data(filepath,
+                                      self.interface.butler.dimensions,
+                                      self.interface.instrument,
+                                      self.next_visit)
         with unittest.mock.patch.object(self.interface.rawIngestTask, "extractMetadata") as mock:
             mock.return_value = file_data
             self.interface.ingest_image(filename)
-
+        # Dummy "engineering" visit to test non-science handling
+        eng_visit = dataclasses.replace(self.next_visit,
+                                        groupId="42",
+                                        coordinateSystem=FannedOutVisit.CoordSys.NONE,
+                                        position=[0.0, 0.0],
+                                        rotationSystem=FannedOutVisit.RotSys.NONE,
+                                        cameraAngle=0.0,
+                                        dome=FannedOutVisit.Dome.CLOSED,
+                                        )
+        _, eng_data = fake_eng_data(filepath,
+                                    self.interface.butler.dimensions,
+                                    self.interface.instrument,
+                                    eng_visit)
+        with unittest.mock.patch.object(self.interface.rawIngestTask, "extractMetadata") as mock:
+            mock.return_value = eng_data
+            self.interface.ingest_image(filename)
         # TODO: add any preprocessing outputs the main pipeline depends on (DM-43418?)
 
     def test_run_pipeline(self):
@@ -594,7 +666,7 @@ def test_run_pipeline_fallback_2failof2(self):
         expected = ""
 
         self._prepare_run_pipeline()
-        with self.assertRaises(RuntimeError):
+        with self.assertRaises(NoGoodPipelinesError):
             self._check_run_pipeline_fallback(lambda: self.interface.run_pipeline({1}),
                                               pipe_list, graph_list, expected)
 
@@ -650,10 +722,10 @@ def test_run_pipeline_bad_visits(self):
         self.interface.prep_butler()
         filename = "fakeRawImage.fits"
         filepath = os.path.join(self.input_data, filename)
-        data_id, file_data = fake_file_data(filepath,
-                                            self.interface.butler.dimensions,
-                                            self.interface.instrument,
-                                            self.next_visit)
+        _, file_data = fake_file_data(filepath,
+                                      self.interface.butler.dimensions,
+                                      self.interface.instrument,
+                                      self.next_visit)
         with unittest.mock.patch.object(self.interface.rawIngestTask, "extractMetadata") as mock:
             mock.return_value = file_data
             self.interface.ingest_image(filename)
@@ -671,9 +743,9 @@ def test_run_pipeline_early_exception(self):
              unittest.mock.patch("activator.middleware_interface.SeparablePipelineExecutor.run_pipeline") \
                 as mock_run, \
              unittest.mock.patch("lsst.dax.apdb.ApdbSql.containsVisitDetector") as mock_query:
-            mock_run.side_effect = RuntimeError("The pipeline doesn't like you.")
+            mock_run.side_effect = ValueError("Error: not computable")
             mock_query.return_value = False
-            with self.assertRaises(RuntimeError):
+            with self.assertRaises(PipelineExecutionError):
                 self.interface.run_pipeline({1})
 
     def test_run_pipeline_late_exception(self):
@@ -686,7 +758,7 @@ def test_run_pipeline_late_exception(self):
              unittest.mock.patch("activator.middleware_interface.SeparablePipelineExecutor.run_pipeline") \
                 as mock_run, \
              unittest.mock.patch("lsst.dax.apdb.ApdbSql.containsVisitDetector") as mock_query:
-            mock_run.side_effect = RuntimeError("The pipeline doesn't like you.")
+            mock_run.side_effect = ValueError("Error: not computable")
             mock_query.return_value = True
             with self.assertRaises(NonRetriableError):
                 self.interface.run_pipeline({1})
@@ -701,11 +773,28 @@ def test_run_pipeline_cascading_exception(self):
              unittest.mock.patch("activator.middleware_interface.SeparablePipelineExecutor.run_pipeline") \
                 as mock_run, \
              unittest.mock.patch("lsst.dax.apdb.ApdbSql.containsVisitDetector") as mock_query:
-            mock_run.side_effect = RuntimeError("The pipeline doesn't like you.")
+            mock_run.side_effect = ValueError("Error: not computable")
             mock_query.side_effect = psycopg2.OperationalError("Database? What database?")
             with self.assertRaises(NonRetriableError):
                 self.interface.run_pipeline({1})
 
+    def test_run_pipeline_early_exception_novisit(self):
+        """Test behavior when execution fails in ISR-only processing.
+        """
+        self._prepare_run_pipeline()
+
+        with unittest.mock.patch(
+                "activator.middleware_interface.SeparablePipelineExecutor.pre_execute_qgraph"), \
+             unittest.mock.patch("activator.middleware_interface.SeparablePipelineExecutor.run_pipeline") \
+                as mock_run, \
+             unittest.mock.patch.object(self.interface, "main_pipelines", pipelines_minimal), \
+             unittest.mock.patch("lsst.dax.apdb.ApdbSql.containsVisitDetector") as mock_query:
+            mock_run.side_effect = ValueError("Error: not computable")
+            mock_query.return_value = False
+            with self.assertRaises(PipelineExecutionError):
+                # Engineering run; see _prepare_run_pipeline
+                self.interface.run_pipeline({42})
+
     def test_run_preprocessing_empty(self):
         """Test that running the preprocessiing pipeline does nothing if no
         pipelines configured.
@@ -767,7 +856,7 @@ def test_run_preprocessing_fallback_2failof2(self):
         expected = ""
 
         self._prepare_run_preprocessing()
-        with self.assertRaises(RuntimeError):
+        with self.assertRaises(NoGoodPipelinesError):
             self._check_run_pipeline_fallback(self.interface._run_preprocessing,
                                               pipe_list, graph_list, expected)