Add a simple unit test for AssembleCellCoaddTask

tests/test_assemble_cell_coadd.py

@@ -0,0 +1,138 @@
+# This file is part of drp_tasks.
+#
+# LSST Data Management System
+# This product includes software developed by the
+# LSST Project (http://www.lsst.org/).
+# See COPYRIGHT file at the top of the source tree.
+#
+# This program 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 3 of the License, or
+# (at your option) any later version.
+#
+# This program 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 LSST License Statement and
+# the GNU General Public License along with this program.  If not,
+# see <https://www.lsstcorp.org/LegalNotices/>.
+#
+"""Test AssembleCoaddTask and its variants.
+"""
+import unittest
+
+import lsst.pipe.base as pipeBase
+import lsst.utils.tests
+from assemble_coadd_test_utils import MockCoaddTestData, makeMockSkyInfo
+from lsst.drp.tasks.assemble_cell_coadd import AssembleCellCoaddConfig, AssembleCellCoaddTask
+
+__all__ = [
+    "MockAssembleCellCoaddConfig",
+    "MockAssembleCellCoaddTask",
+]
+
+
+class MockAssembleCellCoaddConfig(AssembleCellCoaddConfig):
+    pass
+
+
+class MockAssembleCellCoaddTask(AssembleCellCoaddTask):
+    """Lightly modified version of `AssembleCoaddTask` for use with unit tests.
+
+    The modifications bypass the usual middleware for loading data and setting
+    up the Task, and instead supply in-memory mock data references to the `run`
+    method so that the coaddition algorithms can be tested without a Butler.
+    """
+
+    ConfigClass = MockAssembleCellCoaddConfig
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.makeSubtask("input_recorder")
+        self.makeSubtask("interpolate_coadd")
+        self.makeSubtask("scale_zero_point")
+
+    def processResults(self, *args, **kwargs):
+        "This should be tested separately."
+        pass
+
+    def runQuantum(self, mockSkyInfo, warpRefList, *args):
+        """Modified interface for testing coaddition algorithms without a
+        Butler.
+
+        Parameters
+        ----------
+        mockSkyInfo : `lsst.pipe.base.Struct`
+            A simple container that supplies a bounding box and WCS in the
+            same format as the output of
+            `lsst.pipe.tasks.CoaddBaseTask.getSkyInfo`
+        warpRefList : `list` of `lsst.pipe.tasks.MockExposureReference`
+            Data references to the test exposures that will be coadded,
+            using the Gen 3 API.
+
+        Returns
+        -------
+        retStruct : `lsst.pipe.base.Struct`
+            The coadded exposure and associated metadata.
+        """
+
+        self.common = pipeBase.Struct(
+            units=None,
+            wcs=mockSkyInfo.wcs,
+            band="i",
+            identifiers=pipeBase.Struct(skymap=None, tract=0, patch=42, band="i"),
+        )
+
+        retStruct = self.run(
+            warpRefList,
+            mockSkyInfo,
+        )
+        return retStruct
+
+
+class AssembleCellCoaddTestCase(lsst.utils.tests.TestCase):
+    """Tests of AssembleCellCoaddTask.
+
+    These tests bypass the middleware used for accessing data and managing Task
+    execution.
+    """
+
+    def setUp(self):
+        patch = 42
+        tract = 0
+        testData = MockCoaddTestData(fluxRange=1e4)
+        exposures = {}
+        matchedExposures = {}
+        for expId in range(100, 110):
+            exposures[expId], matchedExposures[expId] = testData.makeTestImage(expId)
+        self.dataRefList = testData.makeDataRefList(
+            exposures, matchedExposures, "direct", patch=patch, tract=tract
+        )
+        self.skyInfo = makeMockSkyInfo(testData.bbox, testData.wcs, patch=patch)
+
+    def checkRun(self, assembleTask):
+        """Check that the task runs successfully."""
+        result = assembleTask.runQuantum(self.skyInfo, self.dataRefList)
+
+        # Check that we produced an exposure.
+        self.assertTrue(result.multipleCellCoadd is not None)
+
+    def testAssembleBasic(self):
+        config = MockAssembleCellCoaddConfig()
+        assembleTask = MockAssembleCellCoaddTask(config=config)
+        self.checkRun(assembleTask)
+
+
+class MyMemoryTestCase(lsst.utils.tests.MemoryTestCase):
+    pass
+
+
+def setup_module(module):
+    lsst.utils.tests.init()
+
+
+if __name__ == "__main__":
+    lsst.utils.tests.init()
+    unittest.main()