Coverage for tests/test_separable_pipeline_executor.py: 100%

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/>. 

27 

28 

29import os 

30import tempfile 

31import unittest 

32 

33import lsst.daf.butler 

34import lsst.daf.butler.tests as butlerTests 

35import lsst.pex.config 

36import lsst.utils.tests 

37from lsst.daf.butler.registry import RegistryDefaults 

38from lsst.pipe.base import ( 

39 Instrument, 

40 Pipeline, 

41 PipelineGraph, 

42 QuantumAttemptStatus, 

43 QuantumGraph, 

44 QuantumSuccessCaveats, 

45 TaskMetadata, 

46) 

47from lsst.pipe.base.all_dimensions_quantum_graph_builder import AllDimensionsQuantumGraphBuilder 

48from lsst.pipe.base.automatic_connection_constants import ( 

49 PACKAGES_INIT_OUTPUT_NAME, 

50 PROVENANCE_DATASET_TYPE_NAME, 

51 PROVENANCE_STORAGE_CLASS, 

52) 

53from lsst.pipe.base.mp_graph_executor import MPGraphExecutorError 

54from lsst.pipe.base.quantum_graph import ProvenanceQuantumGraph 

55from lsst.pipe.base.quantum_graph_builder import OutputExistsError 

56from lsst.pipe.base.separable_pipeline_executor import SeparablePipelineExecutor 

57from lsst.pipe.base.tests.mocks import ( 

58 DirectButlerRepo, 

59 DynamicTestPipelineTaskConfig, 

60) 

61from lsst.resources import ResourcePath 

62from lsst.utils.packages import Packages 

63 

64TESTDIR = os.path.abspath(os.path.dirname(__file__)) 

65 

66 

67class SeparablePipelineExecutorTests(lsst.utils.tests.TestCase): 

68 """Test the SeparablePipelineExecutor API with a trivial task.""" 

69 

70 pipeline_file = os.path.join(TESTDIR, "pipelines", "pipeline_separable.yaml") 

71 

72 def setUp(self): 

73 repodir = tempfile.TemporaryDirectory() 

74 # TemporaryDirectory warns on leaks; addCleanup also keeps it from 

75 # getting garbage-collected. 

76 self.addCleanup(tempfile.TemporaryDirectory.cleanup, repodir) 

77 

78 # standalone parameter forces the returned config to also include 

79 # the information from the search paths. 

80 config = lsst.daf.butler.Butler.makeRepo( 

81 repodir.name, standalone=True, searchPaths=[os.path.join(TESTDIR, "config")] 

82 ) 

83 butler = lsst.daf.butler.Butler.from_config(config, writeable=True) 

84 self.enterContext(butler) 

85 output = "fake" 

86 output_run = f"{output}/{Instrument.makeCollectionTimestamp()}" 

87 butler.registry.registerCollection(output_run, lsst.daf.butler.CollectionType.RUN) 

88 butler.registry.registerCollection(output, lsst.daf.butler.CollectionType.CHAINED) 

89 butler.registry.setCollectionChain(output, [output_run]) 

90 butler.registry.defaults = RegistryDefaults(collections=[output], run=output_run) 

91 self.butler = butler 

92 

93 butlerTests.addDatasetType(self.butler, "input", set(), "StructuredDataDict") 

94 butlerTests.addDatasetType(self.butler, "intermediate", set(), "StructuredDataDict") 

95 butlerTests.addDatasetType(self.butler, "a_log", set(), "ButlerLogRecords") 

96 butlerTests.addDatasetType(self.butler, "a_metadata", set(), "TaskMetadata") 

97 butlerTests.addDatasetType(self.butler, "a_config", set(), "Config") 

98 provenance_dataset_type = butlerTests.addDatasetType( 

99 self.butler, PROVENANCE_DATASET_TYPE_NAME, set(), PROVENANCE_STORAGE_CLASS 

100 ) 

101 self.provenance_ref = lsst.daf.butler.DatasetRef( 

102 provenance_dataset_type, 

103 lsst.daf.butler.DataCoordinate.make_empty(self.butler.dimensions), 

104 run=butler.run, 

105 ) 

106 

107 def build_empty_quantum_graph(self) -> None: 

108 pipeline_graph = PipelineGraph(universe=self.butler.dimensions) 

109 pipeline_graph.resolve(self.butler.registry) 

110 builder = AllDimensionsQuantumGraphBuilder(pipeline_graph, self.butler) 

111 return builder.finish(attach_datastore_records=False).assemble() 

112 

113 def check_provenance_fullgraph(self): 

114 provenance_qg = self.butler.get(self.provenance_ref) 

115 empty_data_id = lsst.daf.butler.DataCoordinate.make_empty(self.butler.dimensions) 

116 self.assertCountEqual(provenance_qg.quanta_by_task.keys(), {"a", "b"}) 

117 self.assertCountEqual( 

118 provenance_qg.datasets_by_type.keys(), 

119 { 

120 "input", 

121 "intermediate", 

122 "output", 

123 "a_config", 

124 "b_config", 

125 "a_metadata", 

126 "b_metadata", 

127 "a_log", 

128 "b_log", 

129 }, 

130 ) 

131 a_id = provenance_qg.quanta_by_task["a"][empty_data_id] 

132 b_id = provenance_qg.quanta_by_task["b"][empty_data_id] 

133 input_id = provenance_qg.datasets_by_type["input"][empty_data_id] 

134 intermediate_id = provenance_qg.datasets_by_type["intermediate"][empty_data_id] 

135 output_id = provenance_qg.datasets_by_type["output"][empty_data_id] 

136 a_metadata_id = provenance_qg.datasets_by_type["a_metadata"][empty_data_id] 

137 a_log_id = provenance_qg.datasets_by_type["a_log"][empty_data_id] 

138 b_metadata_id = provenance_qg.datasets_by_type["b_metadata"][empty_data_id] 

139 b_log_id = provenance_qg.datasets_by_type["b_log"][empty_data_id] 

140 self.assertEqual( 

141 provenance_qg.bipartite_xgraph.nodes[a_id]["status"], QuantumAttemptStatus.SUCCESSFUL 

142 ) 

143 self.assertEqual( 

144 provenance_qg.bipartite_xgraph.nodes[b_id]["status"], QuantumAttemptStatus.SUCCESSFUL 

145 ) 

146 self.assertEqual(list(provenance_qg.bipartite_xgraph.predecessors(a_id)), [input_id]) 

147 self.assertEqual( 

148 list(provenance_qg.bipartite_xgraph.successors(a_id)), [intermediate_id, a_metadata_id, a_log_id] 

149 ) 

150 self.assertEqual(list(provenance_qg.bipartite_xgraph.predecessors(b_id)), [intermediate_id]) 

151 self.assertEqual( 

152 list(provenance_qg.bipartite_xgraph.successors(b_id)), [output_id, b_metadata_id, b_log_id] 

153 ) 

154 for datasets_by_data_id in provenance_qg.datasets_by_type.values(): 

155 for dataset_id in datasets_by_data_id.values(): 

156 self.assertTrue(provenance_qg.bipartite_xgraph.nodes[dataset_id]["produced"]) 

157 logs_pqg = self.butler.get( 

158 self.provenance_ref.makeComponentRef("logs"), 

159 parameters={"quanta": (a_id,), "datasets": (b_log_id,)}, 

160 ) 

161 self.assertEqual(list(logs_pqg[a_id][-1]), list(self.butler.get("a_log", empty_data_id))) 

162 self.assertEqual(list(logs_pqg[b_log_id][-1]), list(self.butler.get("b_log", empty_data_id))) 

163 metadata_pqg = self.butler.get( 

164 self.provenance_ref.makeComponentRef("metadata"), 

165 parameters={"quanta": (b_id,), "datasets": (a_metadata_id,)}, 

166 ) 

167 self.assertEqual(metadata_pqg[a_metadata_id][-1], self.butler.get("a_metadata", empty_data_id)) 

168 self.assertEqual(metadata_pqg[b_id][-1], self.butler.get("b_metadata", empty_data_id)) 

169 self.assertIsInstance(self.butler.get("run_provenance.packages"), Packages) 

170 

171 def check_provenance_emptygraph(self): 

172 provenance_qg = self.butler.get(self.provenance_ref) 

173 self.assertFalse(provenance_qg.bipartite_xgraph) 

174 self.assertFalse(any(provenance_qg.quanta_by_task.values())) 

175 self.assertFalse(any(provenance_qg.datasets_by_type.values())) 

176 self.assertIsInstance(self.butler.get("run_provenance.packages"), Packages) 

177 

178 def test_pre_execute_qgraph_old(self): 

179 # Too hard to make a quantum graph from scratch. 

180 executor = SeparablePipelineExecutor(self.butler) 

181 pipeline = Pipeline.fromFile(self.pipeline_file) 

182 self.butler.put({"zero": 0}, "input") 

183 graph = executor.make_quantum_graph(pipeline) 

184 

185 butlerTests.addDatasetType(self.butler, "output", set(), "StructuredDataDict") 

186 butlerTests.addDatasetType(self.butler, "b_config", set(), "Config") 

187 butlerTests.addDatasetType(self.butler, "b_log", set(), "ButlerLogRecords") 

188 butlerTests.addDatasetType(self.butler, "b_metadata", set(), "TaskMetadata") 

189 butlerTests.addDatasetType(self.butler, PACKAGES_INIT_OUTPUT_NAME, set(), "Packages") 

190 

191 executor.pre_execute_qgraph( 

192 graph, 

193 register_dataset_types=False, 

194 save_init_outputs=False, 

195 save_versions=False, 

196 ) 

197 self.assertFalse(self.butler.exists("a_config", {}, collections=[self.butler.run])) 

198 self.assertFalse(self.butler.exists(PACKAGES_INIT_OUTPUT_NAME, {})) 

199 

200 def test_pre_execute_qgraph(self): 

201 # Too hard to make a quantum graph from scratch. 

202 executor = SeparablePipelineExecutor(self.butler) 

203 pipeline = Pipeline.fromFile(self.pipeline_file) 

204 self.butler.put({"zero": 0}, "input") 

205 graph = executor.build_quantum_graph(pipeline) 

206 

207 butlerTests.addDatasetType(self.butler, "output", set(), "StructuredDataDict") 

208 butlerTests.addDatasetType(self.butler, "b_config", set(), "Config") 

209 butlerTests.addDatasetType(self.butler, "b_log", set(), "ButlerLogRecords") 

210 butlerTests.addDatasetType(self.butler, "b_metadata", set(), "TaskMetadata") 

211 butlerTests.addDatasetType(self.butler, PACKAGES_INIT_OUTPUT_NAME, set(), "Packages") 

212 

213 executor.pre_execute_qgraph( 

214 graph, 

215 register_dataset_types=False, 

216 save_init_outputs=False, 

217 save_versions=False, 

218 ) 

219 self.assertFalse(self.butler.exists("a_config", {}, collections=[self.butler.run])) 

220 self.assertFalse(self.butler.exists(PACKAGES_INIT_OUTPUT_NAME, {})) 

221 

222 def test_pre_execute_qgraph_unconnected_old(self): 

223 # Unconnected graph; see 

224 # test_make_quantum_graph_nowhere_skippartial_clobber. 

225 executor = SeparablePipelineExecutor( 

226 self.butler, 

227 skip_existing_in=[self.butler.run], 

228 clobber_output=True, 

229 ) 

230 pipeline = Pipeline.fromFile(self.pipeline_file) 

231 self.butler.put({"zero": 0}, "input") 

232 self.butler.put({"zero": 0}, "intermediate") 

233 graph = executor.make_quantum_graph(pipeline) 

234 

235 butlerTests.addDatasetType(self.butler, "output", set(), "StructuredDataDict") 

236 butlerTests.addDatasetType(self.butler, "b_config", set(), "Config") 

237 butlerTests.addDatasetType(self.butler, "b_log", set(), "ButlerLogRecords") 

238 butlerTests.addDatasetType(self.butler, "b_metadata", set(), "TaskMetadata") 

239 butlerTests.addDatasetType(self.butler, PACKAGES_INIT_OUTPUT_NAME, set(), "Packages") 

240 

241 executor.pre_execute_qgraph( 

242 graph, 

243 register_dataset_types=False, 

244 save_init_outputs=False, 

245 save_versions=False, 

246 ) 

247 self.assertFalse(self.butler.exists("a_config", {}, collections=[self.butler.run])) 

248 self.assertFalse(self.butler.exists(PACKAGES_INIT_OUTPUT_NAME, {})) 

249 

250 def test_pre_execute_qgraph_unconnected(self): 

251 # Unconnected graph; see 

252 # test_make_quantum_graph_nowhere_skippartial_clobber. 

253 executor = SeparablePipelineExecutor( 

254 self.butler, 

255 skip_existing_in=[self.butler.run], 

256 clobber_output=True, 

257 ) 

258 pipeline = Pipeline.fromFile(self.pipeline_file) 

259 self.butler.put({"zero": 0}, "input") 

260 self.butler.put({"zero": 0}, "intermediate") 

261 graph = executor.build_quantum_graph(pipeline) 

262 

263 butlerTests.addDatasetType(self.butler, "output", set(), "StructuredDataDict") 

264 butlerTests.addDatasetType(self.butler, "b_config", set(), "Config") 

265 butlerTests.addDatasetType(self.butler, "b_log", set(), "ButlerLogRecords") 

266 butlerTests.addDatasetType(self.butler, "b_metadata", set(), "TaskMetadata") 

267 butlerTests.addDatasetType(self.butler, PACKAGES_INIT_OUTPUT_NAME, set(), "Packages") 

268 

269 executor.pre_execute_qgraph( 

270 graph, 

271 register_dataset_types=False, 

272 save_init_outputs=False, 

273 save_versions=False, 

274 ) 

275 self.assertFalse(self.butler.exists("a_config", {}, collections=[self.butler.run])) 

276 self.assertFalse(self.butler.exists(PACKAGES_INIT_OUTPUT_NAME, {})) 

277 

278 def test_pre_execute_qgraph_empty_old(self): 

279 executor = SeparablePipelineExecutor(self.butler) 

280 graph = QuantumGraph({}, universe=self.butler.dimensions) 

281 

282 butlerTests.addDatasetType(self.butler, "output", set(), "StructuredDataDict") 

283 butlerTests.addDatasetType(self.butler, "b_config", set(), "Config") 

284 butlerTests.addDatasetType(self.butler, "b_log", set(), "ButlerLogRecords") 

285 butlerTests.addDatasetType(self.butler, "b_metadata", set(), "TaskMetadata") 

286 butlerTests.addDatasetType(self.butler, PACKAGES_INIT_OUTPUT_NAME, set(), "Packages") 

287 

288 executor.pre_execute_qgraph( 

289 graph, 

290 register_dataset_types=False, 

291 save_init_outputs=False, 

292 save_versions=False, 

293 ) 

294 self.assertFalse(self.butler.exists("a_config", {}, collections=[self.butler.run])) 

295 self.assertFalse(self.butler.exists(PACKAGES_INIT_OUTPUT_NAME, {})) 

296 

297 def test_pre_execute_qgraph_empty(self): 

298 executor = SeparablePipelineExecutor(self.butler) 

299 graph = self.build_empty_quantum_graph() 

300 

301 butlerTests.addDatasetType(self.butler, "output", set(), "StructuredDataDict") 

302 butlerTests.addDatasetType(self.butler, "b_config", set(), "Config") 

303 butlerTests.addDatasetType(self.butler, "b_log", set(), "ButlerLogRecords") 

304 butlerTests.addDatasetType(self.butler, "b_metadata", set(), "TaskMetadata") 

305 butlerTests.addDatasetType(self.butler, PACKAGES_INIT_OUTPUT_NAME, set(), "Packages") 

306 

307 executor.pre_execute_qgraph( 

308 graph, 

309 register_dataset_types=False, 

310 save_init_outputs=False, 

311 save_versions=False, 

312 ) 

313 self.assertFalse(self.butler.exists("a_config", {}, collections=[self.butler.run])) 

314 self.assertFalse(self.butler.exists(PACKAGES_INIT_OUTPUT_NAME, {})) 

315 

316 def test_pre_execute_qgraph_register_old(self): 

317 executor = SeparablePipelineExecutor(self.butler) 

318 pipeline = Pipeline.fromFile(self.pipeline_file) 

319 self.butler.put({"zero": 0}, "input") 

320 graph = executor.make_quantum_graph(pipeline) 

321 

322 executor.pre_execute_qgraph( 

323 graph, 

324 register_dataset_types=True, 

325 save_init_outputs=False, 

326 save_versions=False, 

327 ) 

328 self.assertEqual({d.name for d in self.butler.registry.queryDatasetTypes("output")}, {"output"}) 

329 self.assertEqual( 

330 {d.name for d in self.butler.registry.queryDatasetTypes("b_*")}, 

331 {"b_config", "b_log", "b_metadata"}, 

332 ) 

333 self.assertFalse(self.butler.exists("a_config", {}, collections=[self.butler.run])) 

334 self.assertFalse(self.butler.exists(PACKAGES_INIT_OUTPUT_NAME, {})) 

335 

336 def test_pre_execute_qgraph_register(self): 

337 executor = SeparablePipelineExecutor(self.butler) 

338 pipeline = Pipeline.fromFile(self.pipeline_file) 

339 self.butler.put({"zero": 0}, "input") 

340 graph = executor.build_quantum_graph(pipeline) 

341 

342 executor.pre_execute_qgraph( 

343 graph, 

344 register_dataset_types=True, 

345 save_init_outputs=False, 

346 save_versions=False, 

347 ) 

348 self.assertEqual({d.name for d in self.butler.registry.queryDatasetTypes("output")}, {"output"}) 

349 self.assertEqual( 

350 {d.name for d in self.butler.registry.queryDatasetTypes("b_*")}, 

351 {"b_config", "b_log", "b_metadata"}, 

352 ) 

353 self.assertFalse(self.butler.exists("a_config", {}, collections=[self.butler.run])) 

354 self.assertFalse(self.butler.exists(PACKAGES_INIT_OUTPUT_NAME, {})) 

355 

356 def test_pre_execute_qgraph_init_outputs_old(self): 

357 # Too hard to make a quantum graph from scratch. 

358 executor = SeparablePipelineExecutor(self.butler) 

359 pipeline = Pipeline.fromFile(self.pipeline_file) 

360 self.butler.put({"zero": 0}, "input") 

361 graph = executor.make_quantum_graph(pipeline) 

362 

363 butlerTests.addDatasetType(self.butler, "output", set(), "StructuredDataDict") 

364 butlerTests.addDatasetType(self.butler, "b_config", set(), "Config") 

365 butlerTests.addDatasetType(self.butler, "b_log", set(), "ButlerLogRecords") 

366 butlerTests.addDatasetType(self.butler, "b_metadata", set(), "TaskMetadata") 

367 butlerTests.addDatasetType(self.butler, PACKAGES_INIT_OUTPUT_NAME, set(), "Packages") 

368 

369 executor.pre_execute_qgraph( 

370 graph, 

371 register_dataset_types=False, 

372 save_init_outputs=True, 

373 save_versions=False, 

374 ) 

375 self.assertTrue(self.butler.exists("a_config", {}, collections=[self.butler.run])) 

376 self.assertFalse(self.butler.exists(PACKAGES_INIT_OUTPUT_NAME, {})) 

377 

378 def test_pre_execute_qgraph_init_outputs(self): 

379 # Too hard to make a quantum graph from scratch. 

380 executor = SeparablePipelineExecutor(self.butler) 

381 pipeline = Pipeline.fromFile(self.pipeline_file) 

382 self.butler.put({"zero": 0}, "input") 

383 graph = executor.build_quantum_graph(pipeline) 

384 

385 butlerTests.addDatasetType(self.butler, "output", set(), "StructuredDataDict") 

386 butlerTests.addDatasetType(self.butler, "b_config", set(), "Config") 

387 butlerTests.addDatasetType(self.butler, "b_log", set(), "ButlerLogRecords") 

388 butlerTests.addDatasetType(self.butler, "b_metadata", set(), "TaskMetadata") 

389 butlerTests.addDatasetType(self.butler, PACKAGES_INIT_OUTPUT_NAME, set(), "Packages") 

390 

391 executor.pre_execute_qgraph( 

392 graph, 

393 register_dataset_types=False, 

394 save_init_outputs=True, 

395 save_versions=False, 

396 ) 

397 self.assertTrue(self.butler.exists("a_config", {}, collections=[self.butler.run])) 

398 self.assertFalse(self.butler.exists(PACKAGES_INIT_OUTPUT_NAME, {})) 

399 

400 def test_pre_execute_qgraph_versions_old(self): 

401 executor = SeparablePipelineExecutor(self.butler) 

402 pipeline = Pipeline.fromFile(self.pipeline_file) 

403 self.butler.put({"zero": 0}, "input") 

404 graph = executor.make_quantum_graph(pipeline) 

405 

406 butlerTests.addDatasetType(self.butler, "output", set(), "StructuredDataDict") 

407 butlerTests.addDatasetType(self.butler, "b_config", set(), "Config") 

408 butlerTests.addDatasetType(self.butler, "b_log", set(), "ButlerLogRecords") 

409 butlerTests.addDatasetType(self.butler, "b_metadata", set(), "TaskMetadata") 

410 butlerTests.addDatasetType(self.butler, PACKAGES_INIT_OUTPUT_NAME, set(), "Packages") 

411 

412 executor.pre_execute_qgraph( 

413 graph, 

414 register_dataset_types=False, 

415 save_init_outputs=True, 

416 save_versions=True, 

417 ) 

418 self.assertTrue(self.butler.exists("a_config", {}, collections=[self.butler.run])) 

419 self.assertTrue(self.butler.exists(PACKAGES_INIT_OUTPUT_NAME, {})) 

420 

421 def test_pre_execute_qgraph_versions(self): 

422 executor = SeparablePipelineExecutor(self.butler) 

423 pipeline = Pipeline.fromFile(self.pipeline_file) 

424 self.butler.put({"zero": 0}, "input") 

425 graph = executor.build_quantum_graph(pipeline) 

426 

427 butlerTests.addDatasetType(self.butler, "output", set(), "StructuredDataDict") 

428 butlerTests.addDatasetType(self.butler, "b_config", set(), "Config") 

429 butlerTests.addDatasetType(self.butler, "b_log", set(), "ButlerLogRecords") 

430 butlerTests.addDatasetType(self.butler, "b_metadata", set(), "TaskMetadata") 

431 butlerTests.addDatasetType(self.butler, PACKAGES_INIT_OUTPUT_NAME, set(), "Packages") 

432 

433 executor.pre_execute_qgraph( 

434 graph, 

435 register_dataset_types=False, 

436 save_init_outputs=True, 

437 save_versions=True, 

438 ) 

439 self.assertTrue(self.butler.exists("a_config", {}, collections=[self.butler.run])) 

440 self.assertTrue(self.butler.exists(PACKAGES_INIT_OUTPUT_NAME, {})) 

441 

442 def test_init_badinput(self): 

443 with lsst.daf.butler.Butler.from_config(butler=self.butler, collections=[], run="foo") as butler: 

444 with self.assertRaises(ValueError): 

445 SeparablePipelineExecutor(butler) 

446 

447 def test_init_badoutput(self): 

448 with lsst.daf.butler.Butler.from_config(butler=self.butler, collections=["foo"]) as butler: 

449 with self.assertRaises(ValueError): 

450 SeparablePipelineExecutor(butler) 

451 

452 def test_make_pipeline_full(self): 

453 executor = SeparablePipelineExecutor(self.butler) 

454 for uri in [ 

455 self.pipeline_file, 

456 ResourcePath(self.pipeline_file), 

457 ResourcePath(self.pipeline_file).geturl(), 

458 ]: 

459 pipeline_graph = executor.make_pipeline(uri).to_graph() 

460 self.assertEqual(set(pipeline_graph.tasks), {"a", "b"}) 

461 

462 def test_make_pipeline_subset(self): 

463 executor = SeparablePipelineExecutor(self.butler) 

464 path = self.pipeline_file + "#a" 

465 for uri in [ 

466 path, 

467 ResourcePath(path), 

468 ResourcePath(path).geturl(), 

469 ]: 

470 pipeline_graph = executor.make_pipeline(uri).to_graph() 

471 self.assertEqual(set(pipeline_graph.tasks), {"a"}) 

472 

473 def test_build_quantum_graph_nowhere_noskip_noclobber(self): 

474 executor = SeparablePipelineExecutor(self.butler, skip_existing_in=None, clobber_output=False) 

475 pipeline = Pipeline.fromFile(self.pipeline_file) 

476 

477 self.butler.put({"zero": 0}, "input") 

478 

479 graph = executor.make_quantum_graph(pipeline) 

480 self.assertTrue(graph.isConnected) 

481 self.assertEqual(len(graph), 2) 

482 self.assertEqual({q.taskDef.label for q in graph.inputQuanta}, {"a"}) 

483 self.assertEqual({q.taskDef.label for q in graph.outputQuanta}, {"b"}) 

484 

485 def test_make_quantum_graph_nowhere_noskip_noclobber(self): 

486 executor = SeparablePipelineExecutor(self.butler, skip_existing_in=None, clobber_output=False) 

487 pipeline = Pipeline.fromFile(self.pipeline_file) 

488 

489 self.butler.put({"zero": 0}, "input") 

490 

491 graph = executor.build_quantum_graph(pipeline) 

492 self.assertEqual(len(graph), 2) 

493 self.assertEqual(graph.quanta_by_task.keys(), {"a", "b"}) 

494 

495 def test_make_quantum_graph_nowhere_noskip_noclobber_conflict(self): 

496 executor = SeparablePipelineExecutor(self.butler, skip_existing_in=None, clobber_output=False) 

497 pipeline = Pipeline.fromFile(self.pipeline_file) 

498 

499 self.butler.put({"zero": 0}, "input") 

500 self.butler.put({"zero": 0}, "intermediate") 

501 self.butler.put(lsst.daf.butler.ButlerLogRecords.from_records([]), "a_log") 

502 self.butler.put(TaskMetadata(), "a_metadata") 

503 

504 with self.assertRaises(OutputExistsError): 

505 executor.build_quantum_graph(pipeline) 

506 

507 # TODO: need more complex task and Butler to test 

508 # make_quantum_graph(where=...) 

509 

510 def test_build_quantum_graph_nowhere_skipnone_noclobber(self): 

511 executor = SeparablePipelineExecutor( 

512 self.butler, 

513 skip_existing_in=[self.butler.run], 

514 clobber_output=False, 

515 ) 

516 pipeline = Pipeline.fromFile(self.pipeline_file) 

517 

518 self.butler.put({"zero": 0}, "input") 

519 

520 graph = executor.make_quantum_graph(pipeline) 

521 self.assertTrue(graph.isConnected) 

522 self.assertEqual(len(graph), 2) 

523 self.assertEqual({q.taskDef.label for q in graph.inputQuanta}, {"a"}) 

524 self.assertEqual({q.taskDef.label for q in graph.outputQuanta}, {"b"}) 

525 

526 def test_make_quantum_graph_nowhere_skipnone_noclobber(self): 

527 executor = SeparablePipelineExecutor( 

528 self.butler, 

529 skip_existing_in=[self.butler.run], 

530 clobber_output=False, 

531 ) 

532 pipeline = Pipeline.fromFile(self.pipeline_file) 

533 self.butler.put({"zero": 0}, "input") 

534 graph = executor.build_quantum_graph(pipeline) 

535 self.assertEqual(len(graph), 2) 

536 self.assertEqual(graph.quanta_by_task.keys(), {"a", "b"}) 

537 

538 def test_build_quantum_graph_nowhere_skiptotal_noclobber(self): 

539 executor = SeparablePipelineExecutor( 

540 self.butler, 

541 skip_existing_in=[self.butler.run], 

542 clobber_output=False, 

543 ) 

544 pipeline = Pipeline.fromFile(self.pipeline_file) 

545 

546 self.butler.put({"zero": 0}, "input") 

547 self.butler.put({"zero": 0}, "intermediate") 

548 self.butler.put(lsst.daf.butler.ButlerLogRecords.from_records([]), "a_log") 

549 self.butler.put(TaskMetadata(), "a_metadata") 

550 self.butler.put(lsst.pex.config.Config(), "a_config") 

551 

552 graph = executor.make_quantum_graph(pipeline) 

553 self.assertTrue(graph.isConnected) 

554 self.assertEqual(len(graph), 1) 

555 self.assertEqual({q.taskDef.label for q in graph.inputQuanta}, {"b"}) 

556 self.assertEqual({q.taskDef.label for q in graph.outputQuanta}, {"b"}) 

557 

558 def test_make_quantum_graph_nowhere_skiptotal_noclobber(self): 

559 executor = SeparablePipelineExecutor( 

560 self.butler, 

561 skip_existing_in=[self.butler.run], 

562 clobber_output=False, 

563 ) 

564 pipeline = Pipeline.fromFile(self.pipeline_file) 

565 

566 self.butler.put({"zero": 0}, "input") 

567 self.butler.put({"zero": 0}, "intermediate") 

568 self.butler.put(lsst.daf.butler.ButlerLogRecords.from_records([]), "a_log") 

569 self.butler.put(TaskMetadata(), "a_metadata") 

570 self.butler.put(lsst.pex.config.Config(), "a_config") 

571 

572 graph = executor.build_quantum_graph(pipeline) 

573 self.assertEqual(len(graph), 1) 

574 self.assertEqual(graph.header.n_task_quanta["a"], 0) 

575 self.assertEqual(graph.header.n_task_quanta["b"], 1) 

576 

577 def test_make_quantum_graph_nowhere_skippartial_noclobber(self): 

578 executor = SeparablePipelineExecutor( 

579 self.butler, 

580 skip_existing_in=[self.butler.run], 

581 clobber_output=False, 

582 ) 

583 pipeline = Pipeline.fromFile(self.pipeline_file) 

584 

585 self.butler.put({"zero": 0}, "input") 

586 self.butler.put({"zero": 0}, "intermediate") 

587 

588 with self.assertRaises(OutputExistsError): 

589 executor.build_quantum_graph(pipeline) 

590 

591 def test_build_quantum_graph_nowhere_noskip_clobber(self): 

592 executor = SeparablePipelineExecutor(self.butler, skip_existing_in=None, clobber_output=True) 

593 pipeline = Pipeline.fromFile(self.pipeline_file) 

594 

595 self.butler.put({"zero": 0}, "input") 

596 

597 graph = executor.make_quantum_graph(pipeline) 

598 self.assertTrue(graph.isConnected) 

599 self.assertEqual(len(graph), 2) 

600 self.assertEqual({q.taskDef.label for q in graph.inputQuanta}, {"a"}) 

601 self.assertEqual({q.taskDef.label for q in graph.outputQuanta}, {"b"}) 

602 

603 def test_make_quantum_graph_nowhere_noskip_clobber(self): 

604 executor = SeparablePipelineExecutor(self.butler, skip_existing_in=None, clobber_output=True) 

605 pipeline = Pipeline.fromFile(self.pipeline_file) 

606 self.butler.put({"zero": 0}, "input") 

607 graph = executor.build_quantum_graph(pipeline) 

608 self.assertEqual(len(graph), 2) 

609 self.assertEqual(graph.quanta_by_task.keys(), {"a", "b"}) 

610 

611 def test_build_quantum_graph_nowhere_noskip_clobber_conflict(self): 

612 executor = SeparablePipelineExecutor(self.butler, skip_existing_in=None, clobber_output=True) 

613 pipeline = Pipeline.fromFile(self.pipeline_file) 

614 

615 self.butler.put({"zero": 0}, "input") 

616 self.butler.put({"zero": 0}, "intermediate") 

617 self.butler.put(lsst.daf.butler.ButlerLogRecords.from_records([]), "a_log")