DM-15096: Cannot reuse association database with ApPipeTask reruns

python/lsst/ap/verify/ap_verify.py

@@ -30,7 +30,6 @@
 __all__ = ["runApVerify", "runIngestion"]
 
 import argparse
-import os
 import re
 
 import lsst.log
@@ -153,11 +152,10 @@ def _measureFinalProperties(metricsJob, metadata, workspace, args):
         All command-line arguments passed to this program, including those
         supported by `lsst.ap.verify.pipeline_driver.ApPipeParser`.
     """
-    # TODO: remove this function's dependency on pipeline_driver (DM-13555)
     measurements = []
     measurements.extend(measureFromMetadata(metadata))
     measurements.extend(measureFromButlerRepo(workspace.outputRepo, args.dataId))
-    measurements.extend(measureFromL1DbSqlite(os.path.join(workspace.outputRepo, "association.db")))
+    measurements.extend(measureFromL1DbSqlite(workspace.dbLocation))
 
     for measurement in measurements:
         metricsJob.measurements.insert(measurement)

python/lsst/ap/verify/pipeline_driver.py

@@ -39,6 +39,7 @@
 
 import lsst.log
 import lsst.daf.persistence as dafPersist
+from lsst.ap.association import AssociationDBSqliteTask
 import lsst.ap.pipe as apPipe
 from lsst.verify import Job
 
@@ -242,6 +243,9 @@ def _getConfig(workspace):
     packageDir = lsst.utils.getPackageDir(mapper.getPackageName())
 
     config = apPipe.ApPipeTask.ConfigClass()
+    # Equivalent to task-level default for ap_verify
+    config.associator.level1_db.retarget(AssociationDBSqliteTask)
+    config.associator.level1_db.db_name = workspace.dbLocation
     for path in [
         os.path.join(packageDir, 'config'),
         os.path.join(packageDir, 'config', mapper.getCameraName()),

python/lsst/ap/verify/workspace.py

@@ -94,6 +94,16 @@ def outputRepo(self):
         """
         return os.path.join(self._location, 'output')
 
+    @property
+    def dbLocation(self):
+        """The default location of the source association database to be
+        created or updated by the pipeline (`str`, read-only).
+
+        Shall be a filename to a database file suitable
+        for `AssociationDBSqliteTask`.
+        """
+        return os.path.join(self._location, 'association.db')
+
     @property
     def workButler(self):
         """A Butler that can produce pipeline inputs and outputs

tests/test_driver.py

@@ -52,6 +52,22 @@ def wrapper(self, *args, **kwargs):
     return wrapper
 
 
+class InitRecordingMock(unittest.mock.MagicMock):
+    """A MagicMock for classes that records requests for objects of that class.
+
+    Because ``__init__`` cannot be mocked directly, the calls cannot be
+    identified with the usual ``object.method`` syntax. Instead, filter the
+    object's calls for a ``name`` attribute equal to ``__init__``.
+    """
+    def __call__(self, *args, **kwargs):
+        # super() unsafe because MagicMock does not guarantee support
+        instance = unittest.mock.MagicMock.__call__(self, *args, **kwargs)
+        initCall = unittest.mock.call(*args, **kwargs)
+        initCall.name = "__init__"
+        instance.mock_calls.append(initCall)
+        return instance
+
+
 class PipelineDriverTestSuite(lsst.utils.tests.TestCase):
     def setUp(self):
         self._testDir = tempfile.mkdtemp()
@@ -102,7 +118,7 @@ def setUpMockPatch(self, target, **kwargs):
         self.addCleanup(patcher.stop)
         return mock
 
-    # Mock up ApPipeTask to avoid doing any processing. _getConfig patch may be unneccessary after DM-13602
+    # Mock up ApPipeTask to avoid doing any processing.
     @unittest.mock.patch("lsst.ap.verify.pipeline_driver._getConfig", return_value=None)
     @patchApPipe
     def testRunApPipeReturn(self, _mockConfig, mockClass):
@@ -115,7 +131,7 @@ def testRunApPipeReturn(self, _mockConfig, mockClass):
         self.assertEqual(len(metadata.paramNames(topLevelOnly=False)), 1)
         self.assertEqual(metadata.getScalar("lsst.ap.pipe.ccdProcessor.cycleCount"), 42)
 
-    # Mock up ApPipeTask to avoid doing any processing. _getConfig patch may be unneccessary after DM-13602
+    # Mock up ApPipeTask to avoid doing any processing.
     @unittest.mock.patch("lsst.ap.verify.pipeline_driver._getConfig", return_value=None)
     @patchApPipe
     def testRunApPipeSteps(self, _mockConfig, mockClass):
@@ -152,7 +168,7 @@ def testUpdateMetricsReal(self):
 
         self.assertEqual(self.job.measurements, self.subtaskJob.measurements)
 
-    # Mock up ApPipeTask to avoid doing any processing. _getConfig patch may be unneccessary after DM-13602
+    # Mock up ApPipeTask to avoid doing any processing.
     @unittest.mock.patch("lsst.ap.verify.pipeline_driver._getConfig", return_value=None)
     @patchApPipe
     def testUpdateMetricsOnError(self, _mockConfig, mockClass):
@@ -179,16 +195,9 @@ def testRunApPipeCustomConfig(self):
         configFile = os.path.join(self.workspace.configDir, "apPipe.py")
         with open(configFile, "w") as f:
             # Illegal value; would never be set by a real config
-            f.write("config.differencer.doWriteSources = False")
-
-        class InitRecordingMock(unittest.mock.MagicMock):
-            def __call__(self, *args, **kwargs):
-                # super() unsafe because MagicMock does not guarantee support
-                instance = unittest.mock.MagicMock.__call__(self, *args, **kwargs)
-                initCall = unittest.mock.call(*args, **kwargs)
-                initCall.name = "__init__"
-                instance.mock_calls.append(initCall)
-                return instance
+            f.write("config.differencer.doWriteSources = False\n")
+            f.write("config.associator.level1_db.db_name = ':memory:'\n")
+
         task = self.setUpMockPatch("lsst.ap.pipe.ApPipeTask",
                                    spec=True,
                                    new_callable=InitRecordingMock,
@@ -202,6 +211,24 @@ def __call__(self, *args, **kwargs):
             self.assertIn("config", kwargs)
             taskConfig = kwargs["config"]
             self.assertFalse(taskConfig.differencer.doWriteSources)
+            self.assertNotEqual(taskConfig.associator.level1_db.db_name, self.workspace.dbLocation)
+
+    def testRunApPipeWorkspaceDb(self):
+        """Test that runApPipe places a database in the workspace location by default.
+        """
+        task = self.setUpMockPatch("lsst.ap.pipe.ApPipeTask",
+                                   spec=True,
+                                   new_callable=InitRecordingMock,
+                                   _DefaultName=ApPipeTask._DefaultName,
+                                   ConfigClass=ApPipeTask.ConfigClass).return_value
+
+        pipeline_driver.runApPipe(self.job, self.workspace, self.apPipeArgs)
+        initCalls = (c for c in task.mock_calls if c.name == "__init__")
+        for call in initCalls:
+            kwargs = call[2]
+            self.assertIn("config", kwargs)
+            taskConfig = kwargs["config"]
+            self.assertEqual(taskConfig.associator.level1_db.db_name, self.workspace.dbLocation)
 
 
 class MemoryTester(lsst.utils.tests.MemoryTestCase):

tests/test_workspace.py

@@ -48,6 +48,14 @@ def _assertInDir(self, path, baseDir):
         ancestor = os.path.commonprefix([_canonPath, _canonDir])
         self.assertEqual(ancestor, _canonDir)
 
+    def _assertNotInDir(self, path, baseDir):
+        """Test that ``path`` is a subpath of ``baseDir``.
+        """
+        _canonPath = os.path.abspath(os.path.realpath(path))
+        _canonDir = os.path.abspath(os.path.realpath(baseDir))
+        ancestor = os.path.commonprefix([_canonPath, _canonDir])
+        self.assertNotEqual(ancestor, _canonDir)
+
     def testMakeDir(self):
         """Verify that a Workspace creates the workspace directory if it does not exist.
         """
@@ -61,18 +69,35 @@ def testMakeDir(self):
         finally:
             shutil.rmtree(newPath, ignore_errors=True)
 
+    @staticmethod
+    def _allRepos(workspace):
+        """An iterator over all repos exposed by a Workspace.
+        """
+        yield workspace.dataRepo
+        yield workspace.calibRepo
+        yield workspace.templateRepo
+        yield workspace.outputRepo
+
     def testDirectories(self):
         """Verify that a Workspace creates repositories in the target directory.
 
         The exact repository locations are not tested, as they are likely to change.
         """
         root = self._testWorkspace
         self._assertInDir(self._testbed.configDir, root)
-        # Workspace spec allows these to be URIs or paths, whatever the Butler accepts
-        self._assertInDir(url2pathname(self._testbed.dataRepo), root)
-        self._assertInDir(url2pathname(self._testbed.calibRepo), root)
-        self._assertInDir(url2pathname(self._testbed.templateRepo), root)
-        self._assertInDir(url2pathname(self._testbed.outputRepo), root)
+        for repo in WorkspaceTestSuite._allRepos(self._testbed):
+            # Workspace spec allows these to be URIs or paths, whatever the Butler accepts
+            self._assertInDir(url2pathname(repo), root)
+
+    def testDatabase(self):
+        """Verify that a Workspace requests a database file in the target
+        directory, but not in any repository.
+        """
+        root = self._testWorkspace
+        self._assertInDir(self._testbed.dbLocation, root)
+        for repo in WorkspaceTestSuite._allRepos(self._testbed):
+            # Workspace spec allows these to be URIs or paths, whatever the Butler accepts
+            self._assertNotInDir(self._testbed.dbLocation, url2pathname(repo))
 
 
 class MemoryTester(lsst.utils.tests.MemoryTestCase):