Coverage for tests / test_apdbMetricTask.py: 34%

1# This file is part of verify. 

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 program is free software: you can redistribute it and/or modify 

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

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

12# (at your option) any later version. 

13# 

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

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

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

17# GNU General Public License for more details. 

18# 

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

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

21 

22import shutil 

23import tempfile 

24import unittest.mock 

25 

26import astropy.units as u 

27 

28import lsst.utils.tests 

29from lsst.pex.config import Config 

30import lsst.dax.apdb as daxApdb 

31import lsst.daf.butler.tests as butlerTests 

32from lsst.pipe.base import Struct, testUtils 

33 

34from lsst.verify import Measurement 

35from lsst.verify.tasks import ApdbMetricTask 

36from lsst.verify.tasks.testUtils import ApdbMetricTestCase 

37 

38 

39class DummyTask(ApdbMetricTask): 

40 _DefaultName = "NotARealTask" 

41 

42 def makeMeasurement(self, _dbHandle, outputDataId): 

43 if outputDataId: 

44 nChars = len(outputDataId["instrument"]) 

45 return Measurement(self.config.metricName, 

46 nChars * u.dimensionless_unscaled) 

47 else: 

48 return Measurement(self.config.metricName, 

49 0 * u.dimensionless_unscaled) 

50 

51 

52class Gen3ApdbTestSuite(ApdbMetricTestCase): 

53 @classmethod 

54 def makeTask(cls): 

55 config = DummyTask.ConfigClass() 

56 config.apdb_config_url = cls.config_file.name 

57 config.connections.package = "verify" 

58 config.connections.metric = "DummyApdb" 

59 return DummyTask(config=config) 

60 

61 @classmethod 

62 def setUpClass(cls): 

63 super().setUpClass() 

64 

65 sqlite_file = tempfile.NamedTemporaryFile() 

66 cls.addClassCleanup(sqlite_file.close) 

67 cls.config_file = tempfile.NamedTemporaryFile() 

68 cls.addClassCleanup(cls.config_file.close) 

69 apdb_config = daxApdb.ApdbSql.init_database(db_url=f"sqlite:///{sqlite_file.name}") 

70 apdb_config.save(cls.config_file.name) 

71 

72 cls.CAMERA_ID = "NotACam" 

73 cls.VISIT_ID = 42 

74 cls.CHIP_ID = 5 

75 

76 # makeTestRepo called in setUpClass because it's *very* slow 

77 cls.root = tempfile.mkdtemp() 

78 cls.addClassCleanup(shutil.rmtree, cls.root, ignore_errors=True) 

79 cls.repo = butlerTests.makeTestRepo(cls.root, { 

80 "instrument": [cls.CAMERA_ID], 

81 "visit": [cls.VISIT_ID], 

82 "detector": [cls.CHIP_ID], 

83 }) 

84 

85 # self.task not visible at class level 

86 task = cls.makeTask() 

87 connections = task.config.ConnectionsClass(config=task.config) 

88 

89 butlerTests.addDatasetType( 

90 cls.repo, 

91 connections.measurement.name, 

92 connections.measurement.dimensions, 

93 connections.measurement.storageClass) 

94 butlerTests.addDatasetType( 

95 cls.repo, 

96 connections.dbInfo.name, 

97 connections.dbInfo.dimensions, 

98 connections.dbInfo.storageClass) 

99 

100 def setUp(self): 

101 super().setUp() 

102 

103 self.connections = self.task.config.ConnectionsClass( 

104 config=self.task.config) 

105 

106 def _prepareQuantum(self, task): 

107 globalId = { 

108 "instrument": self.CAMERA_ID, 

109 } 

110 detectorId = { 

111 "instrument": self.CAMERA_ID, 

112 "visit": self.VISIT_ID, 

113 "detector": self.CHIP_ID, 

114 } 

115 

116 butler = butlerTests.makeTestCollection(self.repo, uniqueId=self.id()) 

117 info = Config() 

118 butler.put(info, "apdb_marker", detectorId) 

119 

120 quantum = testUtils.makeQuantum( 

121 task, butler, globalId, 

122 {"dbInfo": [detectorId], "measurement": globalId}) 

123 

124 return (butler, quantum, info) 

125 

126 def testRunQuantum(self): 

127 butler, quantum, input = self._prepareQuantum(self.task) 

128 

129 run = testUtils.runTestQuantum(self.task, butler, quantum) 

130 

131 # Did output data ID get passed to DummyTask.run? 

132 expectedId = lsst.daf.butler.DataCoordinate.standardize( 

133 {"instrument": self.CAMERA_ID}, 

134 universe=butler.dimensions) 

135 run.assert_called_once_with( 

136 dbInfo=[input], 

137 outputDataId=expectedId) 

138 

139 def testRunQuantumNone(self): 

140 class NoneTask(DummyTask): 

141 def run(self, *args, **kwargs): 

142 return Struct(measurement=None) 

143 

144 config = NoneTask.ConfigClass() 

145 config.apdb_config_url = self.config_file.name 

146 config.connections.package = "verify" 

147 config.connections.metric = "DummyApdb" 

148 task = NoneTask(config=config) 

149 butler, quantum, input = self._prepareQuantum(task) 

150 

151 with unittest.mock.patch.object( 

152 lsst.pipe.base.QuantumContext, "put") as put: 

153 testUtils.runTestQuantum(task, butler, quantum, mockRun=False) 

154 # Should not attempt to write nonexistent data 

155 put.assert_not_called() 

156 

157 

158# Hack around unittest's hacky test setup system 

159del ApdbMetricTestCase 

160 

161 

162class MemoryTester(lsst.utils.tests.MemoryTestCase): 

163 pass 

164 

165 

166def setup_module(module): 

167 lsst.utils.tests.init() 

168 

169 

170if __name__ == "__main__": 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true

171 lsst.utils.tests.init() 

172 unittest.main()