Coverage for tests/test_graphSkeleton.py: 100%

1# This file is part of pipe_base. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This software is dual licensed under the GNU General Public License and also 

10# under a 3-clause BSD license. Recipients may choose which of these licenses 

11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, 

12# respectively. If you choose the GPL option then the following text applies 

13# (but note that there is still no warranty even if you opt for BSD instead): 

14# 

15# This program is free software: you can redistribute it and/or modify 

16# it under the terms of the GNU General Public License as published by 

17# the Free Software Foundation, either version 3 of the License, or 

18# (at your option) any later version. 

19# 

20# This program is distributed in the hope that it will be useful, 

21# but WITHOUT ANY WARRANTY; without even the implied warranty of 

22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

23# GNU General Public License for more details. 

24# 

25# You should have received a copy of the GNU General Public License 

26# along with this program. If not, see <https://www.gnu.org/licenses/>. 

27 

28"""Unit tests for QuantumGraphSkeleton.""" 

29 

30import unittest 

31 

32from lsst.daf.butler import DataCoordinate, DimensionUniverse 

33from lsst.pipe.base.quantum_graph_skeleton import DatasetKey, QuantumGraphSkeleton 

34 

35 

36class RemoveUnanchoredQuantaTestCase(unittest.TestCase): 

37 """Tests for ``QuantumGraphSkeleton.remove_unanchored_quanta``. 

38 

39 Graph: 

40 source1 -> d1 -> anchor1 (band 1: anchored) 

41 source2 -> d2 -> anchor2 (band 2: anchored) 

42 source3 -> d3 (band 3: unanchored) 

43 source4 -> d4 (band 4: unanchored) 

44 """ 

45 

46 def setUp(self): 

47 universe = DimensionUniverse() 

48 self.skeleton = QuantumGraphSkeleton(["source", "anchor"]) 

49 for i in range(1, 5): 

50 data_id = DataCoordinate.standardize({"band": i}, universe=universe) 

51 source_quantum = self.skeleton.add_quantum_node("source", data_id) 

52 dataset = self.skeleton.add_dataset_node(f"d{i}", data_id) 

53 self.skeleton.add_output_edge(source_quantum, dataset) 

54 if i <= 2: 

55 anchor_quantum = self.skeleton.add_quantum_node("anchor", data_id) 

56 self.skeleton.add_input_edge(anchor_quantum, dataset) 

57 

58 def test_remove_unanchored(self): 

59 """Unanchored source quanta and their descendants are removed.""" 

60 removed = self.skeleton.remove_unanchored_quanta("source", "anchor") 

61 self.assertEqual(removed, {"source": 2}) 

62 self.assertEqual(len(self.skeleton.get_quanta("source")), 2) 

63 self.assertEqual(len(self.skeleton.get_quanta("anchor")), 2) 

64 self.assertNotIn(DatasetKey("d3", (3,)), self.skeleton) 

65 self.assertNotIn(DatasetKey("d4", (4,)), self.skeleton) 

66 

67 # Second call on an already-pruned skeleton is a no-op. 

68 removed = self.skeleton.remove_unanchored_quanta("source", "anchor") 

69 self.assertEqual(removed, {}) 

70 

71 def test_task_dropped_when_all_unanchored(self): 

72 """Task is dropped when all its quanta are removed.""" 

73 removed = self.skeleton.remove_unanchored_quanta("source", "nonexistent") 

74 self.assertEqual(removed["source"], 4) 

75 self.assertFalse(self.skeleton.has_task("source")) 

76 

77 

78if __name__ == "__main__": 

79 unittest.main()