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 Instrument, Pipeline, PipelineGraph, QuantumGraph, TaskMetadata 

39from lsst.pipe.base.all_dimensions_quantum_graph_builder import AllDimensionsQuantumGraphBuilder 

40from lsst.pipe.base.automatic_connection_constants import PACKAGES_INIT_OUTPUT_NAME 

41from lsst.pipe.base.quantum_graph_builder import OutputExistsError 

42from lsst.pipe.base.separable_pipeline_executor import SeparablePipelineExecutor 

43from lsst.resources import ResourcePath 

44 

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

46 

47 

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

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

50 

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

52 

53 def setUp(self): 

54 repodir = tempfile.TemporaryDirectory() 

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

56 # getting garbage-collected. 

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

58 

59 # standalone parameter forces the returned config to also include 

60 # the information from the search paths. 

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

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

63 ) 

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

65 self.enterContext(butler) 

66 output = "fake" 

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

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

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

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

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

72 self.butler = butler 

73 

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

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

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

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

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

79 

80 def build_empty_quantum_graph(self) -> None: 

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

82 pipeline_graph.resolve(self.butler.registry) 

83 builder = AllDimensionsQuantumGraphBuilder(pipeline_graph, self.butler) 

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

85 

86 def test_pre_execute_qgraph_old(self): 

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

88 executor = SeparablePipelineExecutor(self.butler) 

89 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

91 graph = executor.make_quantum_graph(pipeline) 

92 

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

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

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

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

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

98 

99 executor.pre_execute_qgraph( 

100 graph, 

101 register_dataset_types=False, 

102 save_init_outputs=False, 

103 save_versions=False, 

104 ) 

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

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

107 

108 def test_pre_execute_qgraph(self): 

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

110 executor = SeparablePipelineExecutor(self.butler) 

111 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

113 graph = executor.build_quantum_graph(pipeline) 

114 

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

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

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

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

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

120 

121 executor.pre_execute_qgraph( 

122 graph, 

123 register_dataset_types=False, 

124 save_init_outputs=False, 

125 save_versions=False, 

126 ) 

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

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

129 

130 def test_pre_execute_qgraph_unconnected_old(self): 

131 # Unconnected graph; see 

132 # test_make_quantum_graph_nowhere_skippartial_clobber. 

133 executor = SeparablePipelineExecutor( 

134 self.butler, 

135 skip_existing_in=[self.butler.run], 

136 clobber_output=True, 

137 ) 

138 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

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

141 graph = executor.make_quantum_graph(pipeline) 

142 

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

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

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

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

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

148 

149 executor.pre_execute_qgraph( 

150 graph, 

151 register_dataset_types=False, 

152 save_init_outputs=False, 

153 save_versions=False, 

154 ) 

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

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

157 

158 def test_pre_execute_qgraph_unconnected(self): 

159 # Unconnected graph; see 

160 # test_make_quantum_graph_nowhere_skippartial_clobber. 

161 executor = SeparablePipelineExecutor( 

162 self.butler, 

163 skip_existing_in=[self.butler.run], 

164 clobber_output=True, 

165 ) 

166 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

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

169 graph = executor.build_quantum_graph(pipeline) 

170 

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

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

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

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

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

176 

177 executor.pre_execute_qgraph( 

178 graph, 

179 register_dataset_types=False, 

180 save_init_outputs=False, 

181 save_versions=False, 

182 ) 

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

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

185 

186 def test_pre_execute_qgraph_empty_old(self): 

187 executor = SeparablePipelineExecutor(self.butler) 

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

189 

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

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

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

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

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

195 

196 executor.pre_execute_qgraph( 

197 graph, 

198 register_dataset_types=False, 

199 save_init_outputs=False, 

200 save_versions=False, 

201 ) 

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

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

204 

205 def test_pre_execute_qgraph_empty(self): 

206 executor = SeparablePipelineExecutor(self.butler) 

207 graph = self.build_empty_quantum_graph() 

208 

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

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

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

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

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

214 

215 executor.pre_execute_qgraph( 

216 graph, 

217 register_dataset_types=False, 

218 save_init_outputs=False, 

219 save_versions=False, 

220 ) 

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

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

223 

224 def test_pre_execute_qgraph_register_old(self): 

225 executor = SeparablePipelineExecutor(self.butler) 

226 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

228 graph = executor.make_quantum_graph(pipeline) 

229 

230 executor.pre_execute_qgraph( 

231 graph, 

232 register_dataset_types=True, 

233 save_init_outputs=False, 

234 save_versions=False, 

235 ) 

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

237 self.assertEqual( 

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

239 {"b_config", "b_log", "b_metadata"}, 

240 ) 

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

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

243 

244 def test_pre_execute_qgraph_register(self): 

245 executor = SeparablePipelineExecutor(self.butler) 

246 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

248 graph = executor.build_quantum_graph(pipeline) 

249 

250 executor.pre_execute_qgraph( 

251 graph, 

252 register_dataset_types=True, 

253 save_init_outputs=False, 

254 save_versions=False, 

255 ) 

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

257 self.assertEqual( 

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

259 {"b_config", "b_log", "b_metadata"}, 

260 ) 

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

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

263 

264 def test_pre_execute_qgraph_init_outputs_old(self): 

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

266 executor = SeparablePipelineExecutor(self.butler) 

267 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

269 graph = executor.make_quantum_graph(pipeline) 

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=True, 

281 save_versions=False, 

282 ) 

283 self.assertTrue(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_init_outputs(self): 

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

288 executor = SeparablePipelineExecutor(self.butler) 

289 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

291 graph = executor.build_quantum_graph(pipeline) 

292 

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

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

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

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

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

298 

299 executor.pre_execute_qgraph( 

300 graph, 

301 register_dataset_types=False, 

302 save_init_outputs=True, 

303 save_versions=False, 

304 ) 

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

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

307 

308 def test_pre_execute_qgraph_versions_old(self): 

309 executor = SeparablePipelineExecutor(self.butler) 

310 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

312 graph = executor.make_quantum_graph(pipeline) 

313 

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

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

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

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

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

319 

320 executor.pre_execute_qgraph( 

321 graph, 

322 register_dataset_types=False, 

323 save_init_outputs=True, 

324 save_versions=True, 

325 ) 

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

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

328 

329 def test_pre_execute_qgraph_versions(self): 

330 executor = SeparablePipelineExecutor(self.butler) 

331 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

333 graph = executor.build_quantum_graph(pipeline) 

334 

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

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

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

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

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

340 

341 executor.pre_execute_qgraph( 

342 graph, 

343 register_dataset_types=False, 

344 save_init_outputs=True, 

345 save_versions=True, 

346 ) 

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

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

349 

350 def test_init_badinput(self): 

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

352 with self.assertRaises(ValueError): 

353 SeparablePipelineExecutor(butler) 

354 

355 def test_init_badoutput(self): 

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

357 with self.assertRaises(ValueError): 

358 SeparablePipelineExecutor(butler) 

359 

360 def test_make_pipeline_full(self): 

361 executor = SeparablePipelineExecutor(self.butler) 

362 for uri in [ 

363 self.pipeline_file, 

364 ResourcePath(self.pipeline_file), 

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

366 ]: 

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

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

369 

370 def test_make_pipeline_subset(self): 

371 executor = SeparablePipelineExecutor(self.butler) 

372 path = self.pipeline_file + "#a" 

373 for uri in [ 

374 path, 

375 ResourcePath(path), 

376 ResourcePath(path).geturl(), 

377 ]: 

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

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

380 

381 def test_build_quantum_graph_nowhere_noskip_noclobber(self): 

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

383 pipeline = Pipeline.fromFile(self.pipeline_file) 

384 

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

386 

387 graph = executor.make_quantum_graph(pipeline) 

388 self.assertTrue(graph.isConnected) 

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

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

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

392 

393 def test_make_quantum_graph_nowhere_noskip_noclobber(self): 

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

395 pipeline = Pipeline.fromFile(self.pipeline_file) 

396 

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

398 

399 graph = executor.build_quantum_graph(pipeline) 

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

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

402 

403 def test_make_quantum_graph_nowhere_noskip_noclobber_conflict(self): 

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

405 pipeline = Pipeline.fromFile(self.pipeline_file) 

406 

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

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

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

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

411 

412 with self.assertRaises(OutputExistsError): 

413 executor.build_quantum_graph(pipeline) 

414 

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

416 # make_quantum_graph(where=...) 

417 

418 def test_build_quantum_graph_nowhere_skipnone_noclobber(self): 

419 executor = SeparablePipelineExecutor( 

420 self.butler, 

421 skip_existing_in=[self.butler.run], 

422 clobber_output=False, 

423 ) 

424 pipeline = Pipeline.fromFile(self.pipeline_file) 

425 

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

427 

428 graph = executor.make_quantum_graph(pipeline) 

429 self.assertTrue(graph.isConnected) 

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

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

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

433 

434 def test_make_quantum_graph_nowhere_skipnone_noclobber(self): 

435 executor = SeparablePipelineExecutor( 

436 self.butler, 

437 skip_existing_in=[self.butler.run], 

438 clobber_output=False, 

439 ) 

440 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

442 graph = executor.build_quantum_graph(pipeline) 

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

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

445 

446 def test_build_quantum_graph_nowhere_skiptotal_noclobber(self): 

447 executor = SeparablePipelineExecutor( 

448 self.butler, 

449 skip_existing_in=[self.butler.run], 

450 clobber_output=False, 

451 ) 

452 pipeline = Pipeline.fromFile(self.pipeline_file) 

453 

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

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

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

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

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

459 

460 graph = executor.make_quantum_graph(pipeline) 

461 self.assertTrue(graph.isConnected) 

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

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

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

465 

466 def test_make_quantum_graph_nowhere_skiptotal_noclobber(self): 

467 executor = SeparablePipelineExecutor( 

468 self.butler, 

469 skip_existing_in=[self.butler.run], 

470 clobber_output=False, 

471 ) 

472 pipeline = Pipeline.fromFile(self.pipeline_file) 

473 

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

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

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

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

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

479 

480 graph = executor.build_quantum_graph(pipeline) 

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

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

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

484 

485 def test_make_quantum_graph_nowhere_skippartial_noclobber(self): 

486 executor = SeparablePipelineExecutor( 

487 self.butler, 

488 skip_existing_in=[self.butler.run], 

489 clobber_output=False, 

490 ) 

491 pipeline = Pipeline.fromFile(self.pipeline_file) 

492 

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

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

495 

496 with self.assertRaises(OutputExistsError): 

497 executor.build_quantum_graph(pipeline) 

498 

499 def test_build_quantum_graph_nowhere_noskip_clobber(self): 

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

501 pipeline = Pipeline.fromFile(self.pipeline_file) 

502 

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

504 

505 graph = executor.make_quantum_graph(pipeline) 

506 self.assertTrue(graph.isConnected) 

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

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

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

510 

511 def test_make_quantum_graph_nowhere_noskip_clobber(self): 

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

513 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

515 graph = executor.build_quantum_graph(pipeline) 

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

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

518 

519 def test_build_quantum_graph_nowhere_noskip_clobber_conflict(self): 

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

521 pipeline = Pipeline.fromFile(self.pipeline_file) 

522 

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

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

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

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

527 

528 graph = executor.make_quantum_graph(pipeline) 

529 self.assertTrue(graph.isConnected) 

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

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

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

533 

534 def test_make_quantum_graph_nowhere_noskip_clobber_conflict(self): 

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

536 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

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

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

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

541 graph = executor.build_quantum_graph(pipeline) 

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

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

544 

545 def test_build_quantum_graph_nowhere_skipnone_clobber(self): 

546 executor = SeparablePipelineExecutor( 

547 self.butler, 

548 skip_existing_in=[self.butler.run], 

549 clobber_output=True, 

550 ) 

551 pipeline = Pipeline.fromFile(self.pipeline_file) 

552 

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

554 

555 graph = executor.make_quantum_graph(pipeline) 

556 self.assertTrue(graph.isConnected) 

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

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

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

560 

561 def test_make_quantum_graph_nowhere_skipnone_clobber(self): 

562 executor = SeparablePipelineExecutor( 

563 self.butler, 

564 skip_existing_in=[self.butler.run], 

565 clobber_output=True, 

566 ) 

567 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

569 graph = executor.build_quantum_graph(pipeline) 

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

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

572 

573 def test_make_quantum_graph_nowhere_skiptotal_clobber(self): 

574 executor = SeparablePipelineExecutor( 

575 self.butler, 

576 skip_existing_in=[self.butler.run], 

577 clobber_output=True, 

578 ) 

579 pipeline = Pipeline.fromFile(self.pipeline_file) 

580 

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

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

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

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

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

586 

587 graph = executor.make_quantum_graph(pipeline) 

588 self.assertTrue(graph.isConnected) 

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

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

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

592 

593 def test_build_quantum_graph_nowhere_skiptotal_clobber(self): 

594 executor = SeparablePipelineExecutor( 

595 self.butler, 

596 skip_existing_in=[self.butler.run], 

597 clobber_output=True, 

598 ) 

599 pipeline = Pipeline.fromFile(self.pipeline_file) 

600 

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

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

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

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

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

606 

607 graph = executor.make_quantum_graph(pipeline) 

608 self.assertTrue(graph.isConnected) 

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

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