Coverage for tests / test_separable_pipeline_executor.py: 11%

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.quantum_graph_builder import OutputExistsError 

54from lsst.pipe.base.separable_pipeline_executor import SeparablePipelineExecutor 

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

56 DirectButlerRepo, 

57 DynamicTestPipelineTaskConfig, 

58) 

59from lsst.resources import ResourcePath 

60from lsst.utils.packages import Packages 

61 

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

63 

64 

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

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

67 

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

69 

70 def setUp(self): 

71 repodir = tempfile.TemporaryDirectory() 

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

73 # getting garbage-collected. 

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

75 

76 # standalone parameter forces the returned config to also include 

77 # the information from the search paths. 

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

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

80 ) 

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

82 self.enterContext(butler) 

83 output = "fake" 

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

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

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

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

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

89 self.butler = butler 

90 

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

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

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

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

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

96 provenance_dataset_type = butlerTests.addDatasetType( 

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

98 ) 

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

100 provenance_dataset_type, 

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

102 run=butler.run, 

103 ) 

104 

105 def build_empty_quantum_graph(self) -> None: 

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

107 pipeline_graph.resolve(self.butler.registry) 

108 builder = AllDimensionsQuantumGraphBuilder(pipeline_graph, self.butler) 

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

110 

111 def check_provenance_fullgraph(self): 

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

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

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

115 self.assertCountEqual( 

116 provenance_qg.datasets_by_type.keys(), 

117 { 

118 "input", 

119 "intermediate", 

120 "output", 

121 "a_config", 

122 "b_config", 

123 "a_metadata", 

124 "b_metadata", 

125 "a_log", 

126 "b_log", 

127 }, 

128 ) 

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

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

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

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

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

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

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

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

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

138 self.assertEqual( 

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

140 ) 

141 self.assertEqual( 

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

143 ) 

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

145 self.assertEqual( 

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

147 ) 

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

149 self.assertEqual( 

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

151 ) 

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

153 for dataset_id in datasets_by_data_id.values(): 

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

155 logs_pqg = self.butler.get( 

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

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

158 ) 

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

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

161 metadata_pqg = self.butler.get( 

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

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

164 ) 

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

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

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

168 

169 def check_provenance_emptygraph(self): 

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

171 self.assertFalse(provenance_qg.bipartite_xgraph) 

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

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

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

175 

176 def test_pre_execute_qgraph_old(self): 

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

178 executor = SeparablePipelineExecutor(self.butler) 

179 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

181 graph = executor.make_quantum_graph(pipeline) 

182 

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

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

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

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

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

188 

189 executor.pre_execute_qgraph( 

190 graph, 

191 register_dataset_types=False, 

192 save_init_outputs=False, 

193 save_versions=False, 

194 ) 

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

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

197 

198 def test_pre_execute_qgraph(self): 

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

200 executor = SeparablePipelineExecutor(self.butler) 

201 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

203 graph = executor.build_quantum_graph(pipeline) 

204 

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

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

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

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

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

210 

211 executor.pre_execute_qgraph( 

212 graph, 

213 register_dataset_types=False, 

214 save_init_outputs=False, 

215 save_versions=False, 

216 ) 

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

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

219 

220 def test_pre_execute_qgraph_unconnected_old(self): 

221 # Unconnected graph; see 

222 # test_make_quantum_graph_nowhere_skippartial_clobber. 

223 executor = SeparablePipelineExecutor( 

224 self.butler, 

225 skip_existing_in=[self.butler.run], 

226 clobber_output=True, 

227 ) 

228 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

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

231 graph = executor.make_quantum_graph(pipeline) 

232 

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

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

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

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

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

238 

239 executor.pre_execute_qgraph( 

240 graph, 

241 register_dataset_types=False, 

242 save_init_outputs=False, 

243 save_versions=False, 

244 ) 

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

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

247 

248 def test_pre_execute_qgraph_unconnected(self): 

249 # Unconnected graph; see 

250 # test_make_quantum_graph_nowhere_skippartial_clobber. 

251 executor = SeparablePipelineExecutor( 

252 self.butler, 

253 skip_existing_in=[self.butler.run], 

254 clobber_output=True, 

255 ) 

256 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

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

259 graph = executor.build_quantum_graph(pipeline) 

260 

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

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

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

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

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

266 

267 executor.pre_execute_qgraph( 

268 graph, 

269 register_dataset_types=False, 

270 save_init_outputs=False, 

271 save_versions=False, 

272 ) 

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

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

275 

276 def test_pre_execute_qgraph_empty_old(self): 

277 executor = SeparablePipelineExecutor(self.butler) 

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

279 

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

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

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

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

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

285 

286 executor.pre_execute_qgraph( 

287 graph, 

288 register_dataset_types=False, 

289 save_init_outputs=False, 

290 save_versions=False, 

291 ) 

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

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

294 

295 def test_pre_execute_qgraph_empty(self): 

296 executor = SeparablePipelineExecutor(self.butler) 

297 graph = self.build_empty_quantum_graph() 

298 

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

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

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

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

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

304 

305 executor.pre_execute_qgraph( 

306 graph, 

307 register_dataset_types=False, 

308 save_init_outputs=False, 

309 save_versions=False, 

310 ) 

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

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

313 

314 def test_pre_execute_qgraph_register_old(self): 

315 executor = SeparablePipelineExecutor(self.butler) 

316 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

318 graph = executor.make_quantum_graph(pipeline) 

319 

320 executor.pre_execute_qgraph( 

321 graph, 

322 register_dataset_types=True, 

323 save_init_outputs=False, 

324 save_versions=False, 

325 ) 

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

327 self.assertEqual( 

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

329 {"b_config", "b_log", "b_metadata"}, 

330 ) 

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

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

333 

334 def test_pre_execute_qgraph_register(self): 

335 executor = SeparablePipelineExecutor(self.butler) 

336 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

338 graph = executor.build_quantum_graph(pipeline) 

339 

340 executor.pre_execute_qgraph( 

341 graph, 

342 register_dataset_types=True, 

343 save_init_outputs=False, 

344 save_versions=False, 

345 ) 

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

347 self.assertEqual( 

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

349 {"b_config", "b_log", "b_metadata"}, 

350 ) 

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

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

353 

354 def test_pre_execute_qgraph_init_outputs_old(self): 

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

356 executor = SeparablePipelineExecutor(self.butler) 

357 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

359 graph = executor.make_quantum_graph(pipeline) 

360 

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

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

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

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

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

366 

367 executor.pre_execute_qgraph( 

368 graph, 

369 register_dataset_types=False, 

370 save_init_outputs=True, 

371 save_versions=False, 

372 ) 

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

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

375 

376 def test_pre_execute_qgraph_init_outputs(self): 

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

378 executor = SeparablePipelineExecutor(self.butler) 

379 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

381 graph = executor.build_quantum_graph(pipeline) 

382 

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

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

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

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

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

388 

389 executor.pre_execute_qgraph( 

390 graph, 

391 register_dataset_types=False, 

392 save_init_outputs=True, 

393 save_versions=False, 

394 ) 

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

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

397 

398 def test_pre_execute_qgraph_versions_old(self): 

399 executor = SeparablePipelineExecutor(self.butler) 

400 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

402 graph = executor.make_quantum_graph(pipeline) 

403 

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

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

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

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

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

409 

410 executor.pre_execute_qgraph( 

411 graph, 

412 register_dataset_types=False, 

413 save_init_outputs=True, 

414 save_versions=True, 

415 ) 

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

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

418 

419 def test_pre_execute_qgraph_versions(self): 

420 executor = SeparablePipelineExecutor(self.butler) 

421 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

423 graph = executor.build_quantum_graph(pipeline) 

424 

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

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

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

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

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

430 

431 executor.pre_execute_qgraph( 

432 graph, 

433 register_dataset_types=False, 

434 save_init_outputs=True, 

435 save_versions=True, 

436 ) 

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

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

439 

440 def test_init_badinput(self): 

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

442 with self.assertRaises(ValueError): 

443 SeparablePipelineExecutor(butler) 

444 

445 def test_init_badoutput(self): 

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

447 with self.assertRaises(ValueError): 

448 SeparablePipelineExecutor(butler) 

449 

450 def test_make_pipeline_full(self): 

451 executor = SeparablePipelineExecutor(self.butler) 

452 for uri in [ 

453 self.pipeline_file, 

454 ResourcePath(self.pipeline_file), 

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

456 ]: 

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

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

459 

460 def test_make_pipeline_subset(self): 

461 executor = SeparablePipelineExecutor(self.butler) 

462 path = self.pipeline_file + "#a" 

463 for uri in [ 

464 path, 

465 ResourcePath(path), 

466 ResourcePath(path).geturl(), 

467 ]: 

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

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

470 

471 def test_build_quantum_graph_nowhere_noskip_noclobber(self): 

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

473 pipeline = Pipeline.fromFile(self.pipeline_file) 

474 

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

476 

477 graph = executor.make_quantum_graph(pipeline) 

478 self.assertTrue(graph.isConnected) 

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

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

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

482 

483 def test_make_quantum_graph_nowhere_noskip_noclobber(self): 

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

485 pipeline = Pipeline.fromFile(self.pipeline_file) 

486 

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

488 

489 graph = executor.build_quantum_graph(pipeline) 

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

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

492 

493 def test_make_quantum_graph_nowhere_noskip_noclobber_conflict(self): 

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

495 pipeline = Pipeline.fromFile(self.pipeline_file) 

496 

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

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

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

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

501 

502 with self.assertRaises(OutputExistsError): 

503 executor.build_quantum_graph(pipeline) 

504 

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

506 # make_quantum_graph(where=...) 

507 

508 def test_build_quantum_graph_nowhere_skipnone_noclobber(self): 

509 executor = SeparablePipelineExecutor( 

510 self.butler, 

511 skip_existing_in=[self.butler.run], 

512 clobber_output=False, 

513 ) 

514 pipeline = Pipeline.fromFile(self.pipeline_file) 

515 

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

517 

518 graph = executor.make_quantum_graph(pipeline) 

519 self.assertTrue(graph.isConnected) 

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

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

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

523 

524 def test_make_quantum_graph_nowhere_skipnone_noclobber(self): 

525 executor = SeparablePipelineExecutor( 

526 self.butler, 

527 skip_existing_in=[self.butler.run], 

528 clobber_output=False, 

529 ) 

530 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

532 graph = executor.build_quantum_graph(pipeline) 

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

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

535 

536 def test_build_quantum_graph_nowhere_skiptotal_noclobber(self): 

537 executor = SeparablePipelineExecutor( 

538 self.butler, 

539 skip_existing_in=[self.butler.run], 

540 clobber_output=False, 

541 ) 

542 pipeline = Pipeline.fromFile(self.pipeline_file) 

543 

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

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

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

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

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

549 

550 graph = executor.make_quantum_graph(pipeline) 

551 self.assertTrue(graph.isConnected) 

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

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

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

555 

556 def test_make_quantum_graph_nowhere_skiptotal_noclobber(self): 

557 executor = SeparablePipelineExecutor( 

558 self.butler, 

559 skip_existing_in=[self.butler.run], 

560 clobber_output=False, 

561 ) 

562 pipeline = Pipeline.fromFile(self.pipeline_file) 

563 

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

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

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

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

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

569 

570 graph = executor.build_quantum_graph(pipeline) 

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

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

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

574 

575 def test_make_quantum_graph_nowhere_skippartial_noclobber(self): 

576 executor = SeparablePipelineExecutor( 

577 self.butler, 

578 skip_existing_in=[self.butler.run], 

579 clobber_output=False, 

580 ) 

581 pipeline = Pipeline.fromFile(self.pipeline_file) 

582 

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

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

585 

586 with self.assertRaises(OutputExistsError): 

587 executor.build_quantum_graph(pipeline) 

588 

589 def test_build_quantum_graph_nowhere_noskip_clobber(self): 

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

591 pipeline = Pipeline.fromFile(self.pipeline_file) 

592 

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

594 

595 graph = executor.make_quantum_graph(pipeline) 

596 self.assertTrue(graph.isConnected) 

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

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

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

600 

601 def test_make_quantum_graph_nowhere_noskip_clobber(self): 

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

603 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

605 graph = executor.build_quantum_graph(pipeline) 

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

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

608 

609 def test_build_quantum_graph_nowhere_noskip_clobber_conflict(self): 

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

611 pipeline = Pipeline.fromFile(self.pipeline_file) 

612 

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

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

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

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

617 

618 graph = executor.make_quantum_graph(pipeline)