Coverage for tests / test_separable_pipeline_executor.py: 12%

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 TaskMetadata, 

45) 

46from lsst.pipe.base.all_dimensions_quantum_graph_builder import AllDimensionsQuantumGraphBuilder 

47from lsst.pipe.base.automatic_connection_constants import PACKAGES_INIT_OUTPUT_NAME 

48from lsst.pipe.base.quantum_graph_builder import OutputExistsError 

49from lsst.pipe.base.separable_pipeline_executor import SeparablePipelineExecutor 

50from lsst.resources import ResourcePath 

51from lsst.utils.packages import Packages 

52 

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

54 

55 

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

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

58 

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

60 

61 def setUp(self): 

62 repodir = tempfile.TemporaryDirectory() 

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

64 # getting garbage-collected. 

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

66 

67 # standalone parameter forces the returned config to also include 

68 # the information from the search paths. 

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

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

71 ) 

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

73 self.enterContext(butler) 

74 output = "fake" 

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

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

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

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

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

80 self.butler = butler 

81 

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

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

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

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

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

87 provenance_dataset_type = butlerTests.addDatasetType( 

88 self.butler, "run_provenance", set(), "ProvenanceQuantumGraph" 

89 ) 

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

91 provenance_dataset_type, 

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

93 run=butler.run, 

94 ) 

95 

96 def build_empty_quantum_graph(self) -> None: 

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

98 pipeline_graph.resolve(self.butler.registry) 

99 builder = AllDimensionsQuantumGraphBuilder(pipeline_graph, self.butler) 

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

101 

102 def check_provenance_fullgraph(self): 

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

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

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

106 self.assertCountEqual( 

107 provenance_qg.datasets_by_type.keys(), 

108 { 

109 "input", 

110 "intermediate", 

111 "output", 

112 "a_config", 

113 "b_config", 

114 "a_metadata", 

115 "b_metadata", 

116 "a_log", 

117 "b_log", 

118 }, 

119 ) 

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

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

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

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

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

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

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

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

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

129 self.assertEqual( 

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

131 ) 

132 self.assertEqual( 

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

134 ) 

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

136 self.assertEqual( 

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

138 ) 

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

140 self.assertEqual( 

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

142 ) 

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

144 for dataset_id in datasets_by_data_id.values(): 

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

146 logs_pqg = self.butler.get( 

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

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

149 ) 

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

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

152 metadata_pqg = self.butler.get( 

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

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

155 ) 

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

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

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

159 

160 def check_provenance_emptygraph(self): 

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

162 self.assertFalse(provenance_qg.bipartite_xgraph) 

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

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

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

166 

167 def test_pre_execute_qgraph_old(self): 

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

169 executor = SeparablePipelineExecutor(self.butler) 

170 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

172 graph = executor.make_quantum_graph(pipeline) 

173 

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

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

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

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

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

179 

180 executor.pre_execute_qgraph( 

181 graph, 

182 register_dataset_types=False, 

183 save_init_outputs=False, 

184 save_versions=False, 

185 ) 

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

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

188 

189 def test_pre_execute_qgraph(self): 

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

191 executor = SeparablePipelineExecutor(self.butler) 

192 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

194 graph = executor.build_quantum_graph(pipeline) 

195 

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

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

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

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

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

201 

202 executor.pre_execute_qgraph( 

203 graph, 

204 register_dataset_types=False, 

205 save_init_outputs=False, 

206 save_versions=False, 

207 ) 

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

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

210 

211 def test_pre_execute_qgraph_unconnected_old(self): 

212 # Unconnected graph; see 

213 # test_make_quantum_graph_nowhere_skippartial_clobber. 

214 executor = SeparablePipelineExecutor( 

215 self.butler, 

216 skip_existing_in=[self.butler.run], 

217 clobber_output=True, 

218 ) 

219 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

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

222 graph = executor.make_quantum_graph(pipeline) 

223 

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

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

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

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

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

229 

230 executor.pre_execute_qgraph( 

231 graph, 

232 register_dataset_types=False, 

233 save_init_outputs=False, 

234 save_versions=False, 

235 ) 

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

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

238 

239 def test_pre_execute_qgraph_unconnected(self): 

240 # Unconnected graph; see 

241 # test_make_quantum_graph_nowhere_skippartial_clobber. 

242 executor = SeparablePipelineExecutor( 

243 self.butler, 

244 skip_existing_in=[self.butler.run], 

245 clobber_output=True, 

246 ) 

247 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

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

250 graph = executor.build_quantum_graph(pipeline) 

251 

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

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

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

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

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

257 

258 executor.pre_execute_qgraph( 

259 graph, 

260 register_dataset_types=False, 

261 save_init_outputs=False, 

262 save_versions=False, 

263 ) 

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

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

266 

267 def test_pre_execute_qgraph_empty_old(self): 

268 executor = SeparablePipelineExecutor(self.butler) 

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

270 

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

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

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

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

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

276 

277 executor.pre_execute_qgraph( 

278 graph, 

279 register_dataset_types=False, 

280 save_init_outputs=False, 

281 save_versions=False, 

282 ) 

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

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

285 

286 def test_pre_execute_qgraph_empty(self): 

287 executor = SeparablePipelineExecutor(self.butler) 

288 graph = self.build_empty_quantum_graph() 

289 

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

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

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

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

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

295 

296 executor.pre_execute_qgraph( 

297 graph, 

298 register_dataset_types=False, 

299 save_init_outputs=False, 

300 save_versions=False, 

301 ) 

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

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

304 

305 def test_pre_execute_qgraph_register_old(self): 

306 executor = SeparablePipelineExecutor(self.butler) 

307 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

309 graph = executor.make_quantum_graph(pipeline) 

310 

311 executor.pre_execute_qgraph( 

312 graph, 

313 register_dataset_types=True, 

314 save_init_outputs=False, 

315 save_versions=False, 

316 ) 

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

318 self.assertEqual( 

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

320 {"b_config", "b_log", "b_metadata"}, 

321 ) 

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

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

324 

325 def test_pre_execute_qgraph_register(self): 

326 executor = SeparablePipelineExecutor(self.butler) 

327 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

329 graph = executor.build_quantum_graph(pipeline) 

330 

331 executor.pre_execute_qgraph( 

332 graph, 

333 register_dataset_types=True, 

334 save_init_outputs=False, 

335 save_versions=False, 

336 ) 

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

338 self.assertEqual( 

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

340 {"b_config", "b_log", "b_metadata"}, 

341 ) 

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

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

344 

345 def test_pre_execute_qgraph_init_outputs_old(self): 

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

347 executor = SeparablePipelineExecutor(self.butler) 

348 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

350 graph = executor.make_quantum_graph(pipeline) 

351 

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

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

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

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

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

357 

358 executor.pre_execute_qgraph( 

359 graph, 

360 register_dataset_types=False, 

361 save_init_outputs=True, 

362 save_versions=False, 

363 ) 

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

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

366 

367 def test_pre_execute_qgraph_init_outputs(self): 

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

369 executor = SeparablePipelineExecutor(self.butler) 

370 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

372 graph = executor.build_quantum_graph(pipeline) 

373 

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

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

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

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

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

379 

380 executor.pre_execute_qgraph( 

381 graph, 

382 register_dataset_types=False, 

383 save_init_outputs=True, 

384 save_versions=False, 

385 ) 

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

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

388 

389 def test_pre_execute_qgraph_versions_old(self): 

390 executor = SeparablePipelineExecutor(self.butler) 

391 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

393 graph = executor.make_quantum_graph(pipeline) 

394 

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

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

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

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

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

400 

401 executor.pre_execute_qgraph( 

402 graph, 

403 register_dataset_types=False, 

404 save_init_outputs=True, 

405 save_versions=True, 

406 ) 

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

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

409 

410 def test_pre_execute_qgraph_versions(self): 

411 executor = SeparablePipelineExecutor(self.butler) 

412 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

414 graph = executor.build_quantum_graph(pipeline) 

415 

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

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

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

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

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

421 

422 executor.pre_execute_qgraph( 

423 graph, 

424 register_dataset_types=False, 

425 save_init_outputs=True, 

426 save_versions=True, 

427 ) 

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

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

430 

431 def test_init_badinput(self): 

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

433 with self.assertRaises(ValueError): 

434 SeparablePipelineExecutor(butler) 

435 

436 def test_init_badoutput(self): 

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

438 with self.assertRaises(ValueError): 

439 SeparablePipelineExecutor(butler) 

440 

441 def test_make_pipeline_full(self): 

442 executor = SeparablePipelineExecutor(self.butler) 

443 for uri in [ 

444 self.pipeline_file, 

445 ResourcePath(self.pipeline_file), 

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

447 ]: 

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

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

450 

451 def test_make_pipeline_subset(self): 

452 executor = SeparablePipelineExecutor(self.butler) 

453 path = self.pipeline_file + "#a" 

454 for uri in [ 

455 path, 

456 ResourcePath(path), 

457 ResourcePath(path).geturl(), 

458 ]: 

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

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

461 

462 def test_build_quantum_graph_nowhere_noskip_noclobber(self): 

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

464 pipeline = Pipeline.fromFile(self.pipeline_file) 

465 

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

467 

468 graph = executor.make_quantum_graph(pipeline) 

469 self.assertTrue(graph.isConnected) 

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

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

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

473 

474 def test_make_quantum_graph_nowhere_noskip_noclobber(self): 

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

476 pipeline = Pipeline.fromFile(self.pipeline_file) 

477 

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

479 

480 graph = executor.build_quantum_graph(pipeline) 

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

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

483 

484 def test_make_quantum_graph_nowhere_noskip_noclobber_conflict(self): 

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

486 pipeline = Pipeline.fromFile(self.pipeline_file) 

487 

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

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

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

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

492 

493 with self.assertRaises(OutputExistsError): 

494 executor.build_quantum_graph(pipeline) 

495 

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

497 # make_quantum_graph(where=...) 

498 

499 def test_build_quantum_graph_nowhere_skipnone_noclobber(self): 

500 executor = SeparablePipelineExecutor( 

501 self.butler, 

502 skip_existing_in=[self.butler.run], 

503 clobber_output=False, 

504 ) 

505 pipeline = Pipeline.fromFile(self.pipeline_file) 

506 

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

508 

509 graph = executor.make_quantum_graph(pipeline) 

510 self.assertTrue(graph.isConnected) 

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

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

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

514 

515 def test_make_quantum_graph_nowhere_skipnone_noclobber(self): 

516 executor = SeparablePipelineExecutor( 

517 self.butler, 

518 skip_existing_in=[self.butler.run], 

519 clobber_output=False, 

520 ) 

521 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

523 graph = executor.build_quantum_graph(pipeline) 

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

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

526 

527 def test_build_quantum_graph_nowhere_skiptotal_noclobber(self): 

528 executor = SeparablePipelineExecutor( 

529 self.butler, 

530 skip_existing_in=[self.butler.run], 

531 clobber_output=False, 

532 ) 

533 pipeline = Pipeline.fromFile(self.pipeline_file) 

534 

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

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

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

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

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

540 

541 graph = executor.make_quantum_graph(pipeline) 

542 self.assertTrue(graph.isConnected) 

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

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

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

546 

547 def test_make_quantum_graph_nowhere_skiptotal_noclobber(self): 

548 executor = SeparablePipelineExecutor( 

549 self.butler, 

550 skip_existing_in=[self.butler.run], 

551 clobber_output=False, 

552 ) 

553 pipeline = Pipeline.fromFile(self.pipeline_file) 

554 

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

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

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

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

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

560 

561 graph = executor.build_quantum_graph(pipeline) 

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

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

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

565 

566 def test_make_quantum_graph_nowhere_skippartial_noclobber(self): 

567 executor = SeparablePipelineExecutor( 

568 self.butler, 

569 skip_existing_in=[self.butler.run], 

570 clobber_output=False, 

571 ) 

572 pipeline = Pipeline.fromFile(self.pipeline_file) 

573 

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

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

576 

577 with self.assertRaises(OutputExistsError): 

578 executor.build_quantum_graph(pipeline) 

579 

580 def test_build_quantum_graph_nowhere_noskip_clobber(self): 

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

582 pipeline = Pipeline.fromFile(self.pipeline_file) 

583 

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

585 

586 graph = executor.make_quantum_graph(pipeline) 

587 self.assertTrue(graph.isConnected) 

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

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

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

591 

592 def test_make_quantum_graph_nowhere_noskip_clobber(self): 

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

594 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

596 graph = executor.build_quantum_graph(pipeline) 

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

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

599 

600 def test_build_quantum_graph_nowhere_noskip_clobber_conflict(self): 

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

602 pipeline = Pipeline.fromFile(self.pipeline_file) 

603 

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

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

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

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

608 

609 graph = executor.make_quantum_graph(pipeline) 

610 self.assertTrue(graph.isConnected) 

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

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

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