Coverage for tests/test_single_quantum_executor.py: 16%
83 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-21 01:24 -0800
1# This file is part of pipe_base.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <https://www.gnu.org/licenses/>.
28from __future__ import annotations
30import os
31import time
32import unittest
34import lsst.pipe.base.automatic_connection_constants as acc
35from lsst.pipe.base.resource_usage import QuantumResourceUsage
36from lsst.pipe.base.single_quantum_executor import SingleQuantumExecutor
37from lsst.pipe.base.tests.mocks import InMemoryRepo
39TESTDIR = os.path.abspath(os.path.dirname(__file__))
42class SingleQuantumExecutorTestCase(unittest.TestCase):
43 """Tests for SingleQuantumExecutor implementation."""
45 def test_simple_execute(self) -> None:
46 """Run execute() method in simplest setup."""
47 helper = InMemoryRepo("base.yaml")
48 helper.add_task()
49 qgraph = helper.make_quantum_graph_builder().build(attach_datastore_records=False)
50 executor, butler = helper.make_single_quantum_executor()
51 nQuanta = 1
52 nodes = list(qgraph)
53 self.assertEqual(len(nodes), nQuanta)
54 node = nodes[0]
55 t1 = time.time()
56 executor.execute(node.task_node, node.quantum)
57 t2 = time.time()
58 # There must be one dataset of task's output connection
59 self.assertEqual(len(butler.get_datasets("dataset_auto1")), 1)
60 # Test that we can construct resource usage information from the
61 # metadata.
62 (md,) = butler.get_datasets(acc.METADATA_OUTPUT_TEMPLATE.format(label="task_auto1")).values()
63 ru = QuantumResourceUsage.from_task_metadata(md)
64 self.assertIsNotNone(ru)
65 self.assertGreater(ru.memory, 0)
66 self.assertGreater(ru.prep_time, 0)
67 self.assertGreater(ru.init_time, 0)
68 self.assertGreater(ru.run_time, 0)
69 self.assertGreater(ru.run_time_cpu, 0)
70 self.assertGreater(ru.total_time, 0)
71 self.assertLess(ru.total_time, t2 - t1)
73 def test_skip_existing_execute(self) -> None:
74 """Run execute() method twice, with skip_existing_in."""
75 helper = InMemoryRepo("base.yaml")
76 helper.add_task()
77 qgraph = helper.make_quantum_graph_builder().build(attach_datastore_records=False)
78 executor, butler = helper.make_single_quantum_executor()
79 nQuanta = 1
80 nodes = list(qgraph)
81 self.assertEqual(len(nodes), nQuanta)
82 node = nodes[0]
83 executor.execute(node.task_node, node.quantum)
85 outputs1 = butler.get_datasets("dataset_auto1")
86 self.assertEqual(len(outputs1), 1)
87 ref1, obj1 = outputs1.popitem()
89 # Re-run it with skip_existing, it should not run. Note that if it did
90 # run (and called 'butler.put') that would raise an exception.
91 executor = SingleQuantumExecutor(limited_butler_factory=butler.factory, skip_existing=True)
92 executor.execute(node.task_node, node.quantum)
94 outputs2 = butler.get_datasets("dataset_auto1")
95 self.assertEqual(len(outputs2), 1)
96 ref2, obj2 = outputs2.popitem()
97 self.assertEqual(ref1, ref2)
98 # Objects should be the same (but not identities, because the butler
99 # will copy them).
100 self.assertEqual(obj1, obj2)
102 def test_clobber_outputs_execute(self) -> None:
103 """Run execute() method twice, with clobber_outputs."""
104 helper = InMemoryRepo("base.yaml")
105 helper.add_task()
106 qgraph = helper.make_quantum_graph_builder().build(attach_datastore_records=False)
107 executor, butler = helper.make_single_quantum_executor()
108 nQuanta = 1
109 nodes = list(qgraph)
110 self.assertEqual(len(nodes), nQuanta)
111 node = nodes[0]
112 executor.execute(node.task_node, node.quantum)
114 outputs1 = butler.get_datasets("dataset_auto1")
115 self.assertEqual(len(outputs1), 1)
116 ref1, obj1 = outputs1.popitem()
118 # Remove the dataset ourself, and replace it with something
119 # different so we can check later whether it got replaced.
120 butler.pruneDatasets([ref1], disassociate=False, unstore=True, purge=False)
121 obj1.quantum = None
122 butler.put(obj1, ref1)
124 # Re-run it with clobber_outputs and skip_existing, it should not
125 # clobber but should skip instead.
126 executor = SingleQuantumExecutor(
127 limited_butler_factory=butler.factory, skip_existing=True, clobber_outputs=True
128 )
129 executor.execute(node.task_node, node.quantum)
130 outputs2 = butler.get_datasets("dataset_auto1")
131 self.assertEqual(len(outputs2), 1)
132 ref2, obj2 = outputs2.popitem()
133 self.assertEqual(ref1, ref2)
134 self.assertEqual(obj1, obj2)
136 # Re-run it with clobber_outputs but without skip_existing_in, it
137 # should clobber.
138 executor = SingleQuantumExecutor(limited_butler_factory=butler.factory, clobber_outputs=True)
139 executor.execute(node.task_node, node.quantum)
140 outputs3 = butler.get_datasets("dataset_auto1")
141 self.assertEqual(len(outputs3), 1)
142 ref3, obj3 = outputs3.popitem()
143 self.assertEqual(ref1, ref3)
144 self.assertNotEqual(obj1, obj3)
147if __name__ == "__main__":
148 unittest.main()