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.pipe.base import Instrument, Pipeline, PipelineGraph, QuantumGraph, TaskMetadata 

38from lsst.pipe.base.all_dimensions_quantum_graph_builder import AllDimensionsQuantumGraphBuilder 

39from lsst.pipe.base.automatic_connection_constants import PACKAGES_INIT_OUTPUT_NAME 

40from lsst.pipe.base.quantum_graph_builder import OutputExistsError 

41from lsst.pipe.base.separable_pipeline_executor import SeparablePipelineExecutor 

42from lsst.resources import ResourcePath 

43 

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

45 

46 

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

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

49 

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

51 

52 def setUp(self): 

53 repodir = tempfile.TemporaryDirectory() 

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

55 # getting garbage-collected. 

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

57 

58 # standalone parameter forces the returned config to also include 

59 # the information from the search paths. 

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

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

62 ) 

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

64 output = "fake" 

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

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

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

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

69 self.butler = lsst.daf.butler.Butler.from_config(butler=butler, collections=[output], run=output_run) 

70 

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

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

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

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

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

76 

77 def build_empty_quantum_graph(self) -> None: 

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

79 pipeline_graph.resolve(self.butler.registry) 

80 builder = AllDimensionsQuantumGraphBuilder(pipeline_graph, self.butler) 

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

82 

83 def test_pre_execute_qgraph_old(self): 

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

85 executor = SeparablePipelineExecutor(self.butler) 

86 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

88 graph = executor.make_quantum_graph(pipeline) 

89 

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

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

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

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

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

95 

96 executor.pre_execute_qgraph( 

97 graph, 

98 register_dataset_types=False, 

99 save_init_outputs=False, 

100 save_versions=False, 

101 ) 

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

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

104 

105 def test_pre_execute_qgraph(self): 

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

107 executor = SeparablePipelineExecutor(self.butler) 

108 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

110 graph = executor.build_quantum_graph(pipeline) 

111 

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

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

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

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

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

117 

118 executor.pre_execute_qgraph( 

119 graph, 

120 register_dataset_types=False, 

121 save_init_outputs=False, 

122 save_versions=False, 

123 ) 

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

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

126 

127 def test_pre_execute_qgraph_unconnected_old(self): 

128 # Unconnected graph; see 

129 # test_make_quantum_graph_nowhere_skippartial_clobber. 

130 executor = SeparablePipelineExecutor( 

131 self.butler, 

132 skip_existing_in=[self.butler.run], 

133 clobber_output=True, 

134 ) 

135 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

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

138 graph = executor.make_quantum_graph(pipeline) 

139 

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

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

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

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

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

145 

146 executor.pre_execute_qgraph( 

147 graph, 

148 register_dataset_types=False, 

149 save_init_outputs=False, 

150 save_versions=False, 

151 ) 

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

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

154 

155 def test_pre_execute_qgraph_unconnected(self): 

156 # Unconnected graph; see 

157 # test_make_quantum_graph_nowhere_skippartial_clobber. 

158 executor = SeparablePipelineExecutor( 

159 self.butler, 

160 skip_existing_in=[self.butler.run], 

161 clobber_output=True, 

162 ) 

163 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

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

166 graph = executor.build_quantum_graph(pipeline) 

167 

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

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

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

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

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

173 

174 executor.pre_execute_qgraph( 

175 graph, 

176 register_dataset_types=False, 

177 save_init_outputs=False, 

178 save_versions=False, 

179 ) 

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

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

182 

183 def test_pre_execute_qgraph_empty_old(self): 

184 executor = SeparablePipelineExecutor(self.butler) 

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

186 

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

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

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

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

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

192 

193 executor.pre_execute_qgraph( 

194 graph, 

195 register_dataset_types=False, 

196 save_init_outputs=False, 

197 save_versions=False, 

198 ) 

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

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

201 

202 def test_pre_execute_qgraph_empty(self): 

203 executor = SeparablePipelineExecutor(self.butler) 

204 graph = self.build_empty_quantum_graph() 

205 

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

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

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

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

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

211 

212 executor.pre_execute_qgraph( 

213 graph, 

214 register_dataset_types=False, 

215 save_init_outputs=False, 

216 save_versions=False, 

217 ) 

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

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

220 

221 def test_pre_execute_qgraph_register_old(self): 

222 executor = SeparablePipelineExecutor(self.butler) 

223 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

225 graph = executor.make_quantum_graph(pipeline) 

226 

227 executor.pre_execute_qgraph( 

228 graph, 

229 register_dataset_types=True, 

230 save_init_outputs=False, 

231 save_versions=False, 

232 ) 

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

234 self.assertEqual( 

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

236 {"b_config", "b_log", "b_metadata"}, 

237 ) 

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

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

240 

241 def test_pre_execute_qgraph_register(self): 

242 executor = SeparablePipelineExecutor(self.butler) 

243 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

245 graph = executor.build_quantum_graph(pipeline) 

246 

247 executor.pre_execute_qgraph( 

248 graph, 

249 register_dataset_types=True, 

250 save_init_outputs=False, 

251 save_versions=False, 

252 ) 

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

254 self.assertEqual( 

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

256 {"b_config", "b_log", "b_metadata"}, 

257 ) 

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

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

260 

261 def test_pre_execute_qgraph_init_outputs_old(self): 

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

263 executor = SeparablePipelineExecutor(self.butler) 

264 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

266 graph = executor.make_quantum_graph(pipeline) 

267 

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

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

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

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

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

273 

274 executor.pre_execute_qgraph( 

275 graph, 

276 register_dataset_types=False, 

277 save_init_outputs=True, 

278 save_versions=False, 

279 ) 

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

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

282 

283 def test_pre_execute_qgraph_init_outputs(self): 

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

285 executor = SeparablePipelineExecutor(self.butler) 

286 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

288 graph = executor.build_quantum_graph(pipeline) 

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

300 save_versions=False, 

301 ) 

302 self.assertTrue(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_versions_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 butlerTests.addDatasetType(self.butler, "output", set(), "StructuredDataDict") 

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

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

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

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

316 

317 executor.pre_execute_qgraph( 

318 graph, 

319 register_dataset_types=False, 

320 save_init_outputs=True, 

321 save_versions=True, 

322 ) 

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

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

325 

326 def test_pre_execute_qgraph_versions(self): 

327 executor = SeparablePipelineExecutor(self.butler) 

328 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

330 graph = executor.build_quantum_graph(pipeline) 

331 

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

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

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

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

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

337 

338 executor.pre_execute_qgraph( 

339 graph, 

340 register_dataset_types=False, 

341 save_init_outputs=True, 

342 save_versions=True, 

343 ) 

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

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

346 

347 def test_init_badinput(self): 

348 butler = lsst.daf.butler.Butler.from_config(butler=self.butler, collections=[], run="foo") 

349 

350 with self.assertRaises(ValueError): 

351 SeparablePipelineExecutor(butler) 

352 

353 def test_init_badoutput(self): 

354 butler = lsst.daf.butler.Butler.from_config(butler=self.butler, collections=["foo"]) 

355 

356 with self.assertRaises(ValueError): 

357 SeparablePipelineExecutor(butler) 

358 

359 def test_make_pipeline_full(self): 

360 executor = SeparablePipelineExecutor(self.butler) 

361 for uri in [ 

362 self.pipeline_file, 

363 ResourcePath(self.pipeline_file), 

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

365 ]: 

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

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

368 

369 def test_make_pipeline_subset(self): 

370 executor = SeparablePipelineExecutor(self.butler) 

371 path = self.pipeline_file + "#a" 

372 for uri in [ 

373 path, 

374 ResourcePath(path), 

375 ResourcePath(path).geturl(), 

376 ]: 

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

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

379 

380 def test_build_quantum_graph_nowhere_noskip_noclobber(self): 

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

382 pipeline = Pipeline.fromFile(self.pipeline_file) 

383 

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

385 

386 graph = executor.make_quantum_graph(pipeline) 

387 self.assertTrue(graph.isConnected) 

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

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

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

391 

392 def test_make_quantum_graph_nowhere_noskip_noclobber(self): 

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

394 pipeline = Pipeline.fromFile(self.pipeline_file) 

395 

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

397 

398 graph = executor.build_quantum_graph(pipeline) 

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

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

401 

402 def test_make_quantum_graph_nowhere_noskip_noclobber_conflict(self): 

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

404 pipeline = Pipeline.fromFile(self.pipeline_file) 

405 

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

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

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

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

410 

411 with self.assertRaises(OutputExistsError): 

412 executor.build_quantum_graph(pipeline) 

413 

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

415 # make_quantum_graph(where=...) 

416 

417 def test_build_quantum_graph_nowhere_skipnone_noclobber(self): 

418 executor = SeparablePipelineExecutor( 

419 self.butler, 

420 skip_existing_in=[self.butler.run], 

421 clobber_output=False, 

422 ) 

423 pipeline = Pipeline.fromFile(self.pipeline_file) 

424 

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

426 

427 graph = executor.make_quantum_graph(pipeline) 

428 self.assertTrue(graph.isConnected) 

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

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

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

432 

433 def test_make_quantum_graph_nowhere_skipnone_noclobber(self): 

434 executor = SeparablePipelineExecutor( 

435 self.butler, 

436 skip_existing_in=[self.butler.run], 

437 clobber_output=False, 

438 ) 

439 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

441 graph = executor.build_quantum_graph(pipeline) 

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

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

444 

445 def test_build_quantum_graph_nowhere_skiptotal_noclobber(self): 

446 executor = SeparablePipelineExecutor( 

447 self.butler, 

448 skip_existing_in=[self.butler.run], 

449 clobber_output=False, 

450 ) 

451 pipeline = Pipeline.fromFile(self.pipeline_file) 

452 

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

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

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

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

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

458 

459 graph = executor.make_quantum_graph(pipeline) 

460 self.assertTrue(graph.isConnected) 

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

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

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

464 

465 def test_make_quantum_graph_nowhere_skiptotal_noclobber(self): 

466 executor = SeparablePipelineExecutor( 

467 self.butler, 

468 skip_existing_in=[self.butler.run], 

469 clobber_output=False, 

470 ) 

471 pipeline = Pipeline.fromFile(self.pipeline_file) 

472 

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

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

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

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

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

478 

479 graph = executor.build_quantum_graph(pipeline) 

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

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

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

483 

484 def test_make_quantum_graph_nowhere_skippartial_noclobber(self): 

485 executor = SeparablePipelineExecutor( 

486 self.butler, 

487 skip_existing_in=[self.butler.run], 

488 clobber_output=False, 

489 ) 

490 pipeline = Pipeline.fromFile(self.pipeline_file) 

491 

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

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

494 

495 with self.assertRaises(OutputExistsError): 

496 executor.build_quantum_graph(pipeline) 

497 

498 def test_build_quantum_graph_nowhere_noskip_clobber(self): 

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

500 pipeline = Pipeline.fromFile(self.pipeline_file) 

501 

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

503 

504 graph = executor.make_quantum_graph(pipeline) 

505 self.assertTrue(graph.isConnected) 

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

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

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

509 

510 def test_make_quantum_graph_nowhere_noskip_clobber(self): 

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

512 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

514 graph = executor.build_quantum_graph(pipeline) 

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

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

517 

518 def test_build_quantum_graph_nowhere_noskip_clobber_conflict(self): 

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

520 pipeline = Pipeline.fromFile(self.pipeline_file) 

521 

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

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

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

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

526 

527 graph = executor.make_quantum_graph(pipeline) 

528 self.assertTrue(graph.isConnected) 

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

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

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

532 

533 def test_make_quantum_graph_nowhere_noskip_clobber_conflict(self): 

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

535 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

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

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

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

540 graph = executor.build_quantum_graph(pipeline) 

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

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

543 

544 def test_build_quantum_graph_nowhere_skipnone_clobber(self): 

545 executor = SeparablePipelineExecutor( 

546 self.butler, 

547 skip_existing_in=[self.butler.run], 

548 clobber_output=True, 

549 ) 

550 pipeline = Pipeline.fromFile(self.pipeline_file) 

551 

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

553 

554 graph = executor.make_quantum_graph(pipeline) 

555 self.assertTrue(graph.isConnected) 

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

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

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

559 

560 def test_make_quantum_graph_nowhere_skipnone_clobber(self): 

561 executor = SeparablePipelineExecutor( 

562 self.butler, 

563 skip_existing_in=[self.butler.run], 

564 clobber_output=True, 

565 ) 

566 pipeline = Pipeline.fromFile(self.pipeline_file) 

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

568 graph = executor.build_quantum_graph(pipeline) 

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

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

571 

572 def test_make_quantum_graph_nowhere_skiptotal_clobber(self): 

573 executor = SeparablePipelineExecutor( 

574 self.butler, 

575 skip_existing_in=[self.butler.run], 

576 clobber_output=True, 

577 ) 

578 pipeline = Pipeline.fromFile(self.pipeline_file) 

579 

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

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

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

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

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

585 

586 graph = executor.make_quantum_graph(pipeline) 

587 self.assertTrue(graph.isConnected) 

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

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

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

591 

592 def test_build_quantum_graph_nowhere_skiptotal_clobber(self): 

593 executor = SeparablePipelineExecutor( 

594 self.butler, 

595 skip_existing_in=[self.butler.run], 

596 clobber_output=True, 

597 ) 

598 pipeline = Pipeline.fromFile(self.pipeline_file) 

599 

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

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

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

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

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

605 

606 graph = executor.make_quantum_graph(pipeline) 

607 self.assertTrue(graph.isConnected) 

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

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