Merge pull request #114 from lsst/tickets/DM-28101

DM-28101: miscellaneous usability improvements

python/lsst/ctrl/mpexec/cli/script/qgraph.py

@@ -131,6 +131,9 @@ def qgraph(pipelineObj, qgraph, qgraph_id, qgraph_node_id, skip_existing, save_q
     f = CmdLineFwk()
     qgraph = f.makeGraph(pipelineObj, args)
 
+    if qgraph is None:
+        raise RuntimeError("QuantumGraph is empty.")
+
     # optionally dump some info.
     if show:
         f.showInfo(args, pipelineObj, qgraph)

python/lsst/ctrl/mpexec/cmdLineFwk.py

@@ -213,7 +213,23 @@ def check(self, args: argparse.Namespace):
         """
         assert not (args.extend_run and args.replace_run), "In mutually-exclusive group in ArgumentParser."
         if self.inputs and self.output is not None and self.output.exists:
-            raise ValueError("Cannot use --output with existing collection with --inputs.")
+            # Passing the same inputs that were used to initialize the output
+            # collection is allowed; this means they must _end_ with the same
+            # collections, because we push new runs to the front of the chain.
+            for c1, c2 in zip(self.inputs[::-1], self.output.chain[::-1]):
+                if c1 != c2:
+                    raise ValueError(
+                        f"Output CHAINED collection {self.output.name!r} exists, but it ends with "
+                        "a different sequence of input collections than those given: "
+                        f"{c1!r} != {c2!r} in inputs={self.inputs} vs "
+                        f"{self.output.name}={self.output.chain}."
+                    )
+            if len(self.inputs) > len(self.output.chain):
+                nNew = len(self.inputs) - len(self.output.chain)
+                raise ValueError(
+                    f"Cannot add new input collections {self.inputs[:nNew]} after "
+                    "output collection is first created."
+                )
         if args.extend_run and self.outputRun is None:
             raise ValueError("Cannot --extend-run when no output collection is given.")
         if args.extend_run and not self.outputRun.exists:

tests/test_cmdLineFwk.py

@@ -533,8 +533,8 @@ def testSimpleQGraphReplaceRun(self):
         refs = butler.registry.queryDatasets(..., collections="output/run1")
         self.assertEqual(len(list(refs)), n_outputs)
 
-        # re-run with --replace-run (--inputs is not compatible)
-        args.input = None
+        # re-run with --replace-run (--inputs is ignored, as long as it hasn't
+        # changed)
         args.replace_run = True
         args.output_run = "output/run2"
         fwk.runPipeline(copy.deepcopy(qgraph), taskFactory, args)
@@ -552,7 +552,6 @@ def testSimpleQGraphReplaceRun(self):
         self.assertEqual(len(list(refs)), n_outputs)
 
         # re-run with --replace-run and --prune-replaced=unstore
-        args.input = None
         args.replace_run = True
         args.prune_replaced = "unstore"
         args.output_run = "output/run3"
@@ -575,6 +574,8 @@ def testSimpleQGraphReplaceRun(self):
             butler.get(refs[0], collections="output/run2")
 
         # re-run with --replace-run and --prune-replaced=purge
+        # This time also remove --input; passing the same inputs that we
+        # started with and not passing inputs at all should be equivalent.
         args.input = None
         args.replace_run = True
         args.prune_replaced = "purge"
@@ -590,6 +591,27 @@ def testSimpleQGraphReplaceRun(self):
         refs = butler.registry.queryDatasets(..., collections="output/run4")
         self.assertEqual(len(list(refs)), n_outputs)
 
+        # Trying to run again with inputs that aren't exactly what we started
+        # with is an error, and the kind that should not modify the data repo.
+        with self.assertRaises(ValueError):
+            args.input = ["test", "output/run2"]
+            args.prune_replaced = None
+            args.replace_run = True
+            args.output_run = "output/run5"
+            fwk.runPipeline(copy.deepcopy(qgraph), taskFactory, args)
+        butler.registry.refresh()
+        collections = set(butler.registry.queryCollections(...))
+        self.assertEqual(collections, {"test", "output", "output/run1", "output/run2", "output/run4"})
+        with self.assertRaises(ValueError):
+            args.input = ["output/run2", "test"]
+            args.prune_replaced = None
+            args.replace_run = True
+            args.output_run = "output/run6"
+            fwk.runPipeline(copy.deepcopy(qgraph), taskFactory, args)
+        butler.registry.refresh()
+        collections = set(butler.registry.queryCollections(...))
+        self.assertEqual(collections, {"test", "output", "output/run1", "output/run2", "output/run4"})
+
     def testSubgraph(self):
         """Test successfull execution of trivial quantum graph.
         """