Coverage for tests / test_metrics.py: 28%

1# This file is part of ap_association. 

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 unittest 

23import unittest.mock 

24 

25import astropy.units as u 

26 

27import lsst.utils.tests 

28from lsst.pex.config import Config 

29from lsst.daf.base import PropertySet 

30from lsst.dax.apdb import Apdb 

31import lsst.pipe.base.testUtils 

32from lsst.verify import Name 

33from lsst.verify.tasks import MetricComputationError 

34from lsst.verify.tasks.testUtils import MetricTaskTestCase, MetadataMetricTestCase, ApdbMetricTestCase 

35 

36from lsst.ap.association.metrics import \ 

37 NumberNewDiaObjectsMetricTask, \ 

38 NumberUnassociatedDiaObjectsMetricTask, \ 

39 FractionUpdatedDiaObjectsMetricTask, \ 

40 NumberSolarSystemObjectsMetricTask, \ 

41 NumberAssociatedSolarSystemObjectsMetricTask, \ 

42 TotalUnassociatedDiaObjectsMetricTask 

43 

44 

45def _makeAssociationMetadata(numUpdated=27, numNew=4, numUnassociated=15, numSso=5, numAssocSso=1): 

46 metadata = PropertySet() 

47 metadata.add("association.numUpdatedDiaObjects", numUpdated) 

48 metadata.add("association.numNewDiaObjects", numNew) 

49 metadata.add("association.numTotalSolarSystemObjects", numSso) 

50 metadata.add("association.numAssociatedSsObjects", numAssocSso) 

51 metadata.add("association.numUnassociatedDiaObjects", numUnassociated) 

52 return metadata 

53 

54 

55class TestNewDiaObjects(MetadataMetricTestCase): 

56 

57 @classmethod 

58 def makeTask(cls): 

59 return NumberNewDiaObjectsMetricTask() 

60 

61 def testValid(self): 

62 metadata = _makeAssociationMetadata() 

63 result = self.task.run(metadata) 

64 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

65 meas = result.measurement 

66 

67 self.assertEqual(meas.metric_name, Name(metric="ap_association.numNewDiaObjects")) 

68 self.assertEqual(meas.quantity, metadata.getAsDouble("association.numNewDiaObjects") * u.count) 

69 

70 def testNoNew(self): 

71 metadata = _makeAssociationMetadata(numNew=0) 

72 result = self.task.run(metadata) 

73 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

74 meas = result.measurement 

75 

76 self.assertEqual(meas.metric_name, Name(metric="ap_association.numNewDiaObjects")) 

77 self.assertEqual(meas.quantity, 0.0 * u.count) 

78 

79 def testAssociationFailed(self): 

80 try: 

81 result = self.task.run(PropertySet()) 

82 except lsst.pipe.base.NoWorkFound: 

83 # Correct behavior 

84 pass 

85 else: 

86 # Alternative correct behavior 

87 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

88 meas = result.measurement 

89 self.assertIsNone(meas) 

90 

91 def testBadlyTypedKeys(self): 

92 metadata = _makeAssociationMetadata() 

93 metadata.set("association.numNewDiaObjects", "Ultimate Answer") 

94 

95 with self.assertRaises(MetricComputationError): 

96 self.task.run(metadata) 

97 

98 

99class TestUnassociatedDiaObjects(MetadataMetricTestCase): 

100 

101 @classmethod 

102 def makeTask(cls): 

103 return NumberUnassociatedDiaObjectsMetricTask() 

104 

105 def testValid(self): 

106 metadata = _makeAssociationMetadata() 

107 result = self.task.run(metadata) 

108 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

109 meas = result.measurement 

110 

111 self.assertEqual(meas.metric_name, Name(metric="ap_association.numUnassociatedDiaObjects")) 

112 self.assertEqual(meas.quantity, 

113 metadata.getAsDouble("association.numUnassociatedDiaObjects") * u.count) 

114 

115 def testAllUpdated(self): 

116 metadata = _makeAssociationMetadata(numUnassociated=0) 

117 result = self.task.run(metadata) 

118 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

119 meas = result.measurement 

120 

121 self.assertEqual(meas.metric_name, Name(metric="ap_association.numUnassociatedDiaObjects")) 

122 self.assertEqual(meas.quantity, 0.0 * u.count) 

123 

124 def testAssociationFailed(self): 

125 try: 

126 result = self.task.run(PropertySet()) 

127 except lsst.pipe.base.NoWorkFound: 

128 # Correct behavior 

129 pass 

130 else: 

131 # Alternative correct behavior 

132 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

133 meas = result.measurement 

134 self.assertIsNone(meas) 

135 

136 def testBadlyTypedKeys(self): 

137 metadata = _makeAssociationMetadata() 

138 metadata.set("association.numUnassociatedDiaObjects", "Ultimate Answer") 

139 

140 with self.assertRaises(MetricComputationError): 

141 self.task.run(metadata) 

142 

143 

144class TestFracUpdatedDiaObjects(MetadataMetricTestCase): 

145 

146 @classmethod 

147 def makeTask(cls): 

148 return FractionUpdatedDiaObjectsMetricTask() 

149 

150 def testValid(self): 

151 metadata = _makeAssociationMetadata() 

152 result = self.task.run(metadata) 

153 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

154 meas = result.measurement 

155 

156 self.assertEqual(meas.metric_name, Name(metric="ap_association.fracUpdatedDiaObjects")) 

157 nUpdated = metadata.getAsDouble("association.numUpdatedDiaObjects") 

158 nTotal = metadata.getAsDouble("association.numUnassociatedDiaObjects") + nUpdated 

159 self.assertEqual(meas.quantity, nUpdated / nTotal * u.dimensionless_unscaled) 

160 

161 def testNoUpdated(self): 

162 metadata = _makeAssociationMetadata(numUpdated=0) 

163 result = self.task.run(metadata) 

164 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

165 meas = result.measurement 

166 

167 self.assertEqual(meas.metric_name, Name(metric="ap_association.fracUpdatedDiaObjects")) 

168 self.assertEqual(meas.quantity, 0.0 * u.dimensionless_unscaled) 

169 

170 def testAllUpdated(self): 

171 metadata = _makeAssociationMetadata(numUnassociated=0) 

172 result = self.task.run(metadata) 

173 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

174 meas = result.measurement 

175 

176 self.assertEqual(meas.metric_name, Name(metric="ap_association.fracUpdatedDiaObjects")) 

177 self.assertEqual(meas.quantity, 1.0 * u.dimensionless_unscaled) 

178 

179 def testNoObjects(self): 

180 metadata = _makeAssociationMetadata(numUpdated=0, numUnassociated=0) 

181 try: 

182 result = self.task.run(metadata) 

183 except lsst.pipe.base.NoWorkFound: 

184 # Correct behavior 

185 pass 

186 else: 

187 # Alternative correct behavior 

188 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

189 meas = result.measurement 

190 self.assertIsNone(meas) 

191 

192 def testAssociationFailed(self): 

193 try: 

194 result = self.task.run(PropertySet()) 

195 except lsst.pipe.base.NoWorkFound: 

196 # Correct behavior 

197 pass 

198 else: 

199 # Alternative correct behavior 

200 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

201 meas = result.measurement 

202 self.assertIsNone(meas) 

203 

204 def testBadlyTypedKeys(self): 

205 metadata = _makeAssociationMetadata() 

206 metadata.set("association.numUnassociatedDiaObjects", "Ultimate Answer") 

207 

208 with self.assertRaises(MetricComputationError): 

209 self.task.run(metadata) 

210 

211 

212class TestNumberSolarSystemObjects(MetadataMetricTestCase): 

213 

214 @classmethod 

215 def makeTask(cls): 

216 return NumberSolarSystemObjectsMetricTask() 

217 

218 def testValid(self): 

219 metadata = _makeAssociationMetadata() 

220 result = self.task.run(metadata) 

221 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

222 meas = result.measurement 

223 

224 self.assertEqual(meas.metric_name, Name(metric="ap_association.numTotalSolarSystemObjects")) 

225 self.assertEqual(meas.quantity, 

226 metadata.getAsDouble("association.numTotalSolarSystemObjects") * u.count) 

227 

228 def testAllUpdated(self): 

229 metadata = _makeAssociationMetadata(numSso=0) 

230 result = self.task.run(metadata) 

231 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

232 meas = result.measurement 

233 

234 self.assertEqual(meas.metric_name, Name(metric="ap_association.numTotalSolarSystemObjects")) 

235 self.assertEqual(meas.quantity, 0.0 * u.count) 

236 

237 def testAssociationFailed(self): 

238 try: 

239 result = self.task.run(PropertySet()) 

240 except lsst.pipe.base.NoWorkFound: 

241 # Correct behavior 

242 pass 

243 else: 

244 # Alternative correct behavior 

245 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

246 meas = result.measurement 

247 self.assertIsNone(meas) 

248 

249 def testBadlyTypedKeys(self): 

250 metadata = _makeAssociationMetadata() 

251 metadata.set("association.numTotalSolarSystemObjects", "Ultimate Answer") 

252 

253 with self.assertRaises(MetricComputationError): 

254 self.task.run(metadata) 

255 

256 

257class TestNumberAssociatedSolarSystemObjects(MetadataMetricTestCase): 

258 

259 @classmethod 

260 def makeTask(cls): 

261 return NumberAssociatedSolarSystemObjectsMetricTask() 

262 

263 def testValid(self): 

264 metadata = _makeAssociationMetadata() 

265 result = self.task.run(metadata) 

266 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

267 meas = result.measurement 

268 

269 self.assertEqual(meas.metric_name, Name(metric="ap_association.numAssociatedSsObjects")) 

270 self.assertEqual(meas.quantity, 

271 metadata.getAsDouble("association.numAssociatedSsObjects") * u.count) 

272 

273 def testAllUpdated(self): 

274 metadata = _makeAssociationMetadata(numAssocSso=0) 

275 result = self.task.run(metadata) 

276 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

277 meas = result.measurement 

278 

279 self.assertEqual(meas.metric_name, Name(metric="ap_association.numAssociatedSsObjects")) 

280 self.assertEqual(meas.quantity, 0.0 * u.count) 

281 

282 def testAssociationFailed(self): 

283 try: 

284 result = self.task.run(PropertySet()) 

285 except lsst.pipe.base.NoWorkFound: 

286 # Correct behavior 

287 pass 

288 else: 

289 # Alternative correct behavior 

290 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

291 meas = result.measurement 

292 self.assertIsNone(meas) 

293 

294 def testBadlyTypedKeys(self): 

295 metadata = _makeAssociationMetadata() 

296 metadata.set("association.numAssociatedSsObjects", "Ultimate Answer") 

297 

298 with self.assertRaises(MetricComputationError): 

299 self.task.run(metadata) 

300 

301 

302class TestTotalUnassociatedObjects(ApdbMetricTestCase): 

303 

304 @classmethod 

305 def makeTask(cls): 

306 config = TotalUnassociatedDiaObjectsMetricTask.ConfigClass() 

307 config.apdb_config_url = "dummy/path.yaml" 

308 return TotalUnassociatedDiaObjectsMetricTask(config=config) 

309 

310 def setUp(self): 

311 super().setUp() 

312 

313 # Default patch that applies to ApdbMetricTestCase's tests 

314 loadPatcher = unittest.mock.patch( 

315 "lsst.dax.apdb.Apdb.from_uri", 

316 return_value=unittest.mock.Mock(Apdb, **{"countUnassociatedObjects.return_value": 42}) 

317 ) 

318 loadPatcher.start() 

319 self.addCleanup(loadPatcher.stop) 

320 

321 def testValid(self): 

322 result = self.task.run([Config()]) 

323 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

324 meas = result.measurement 

325 

326 self.assertEqual(meas.metric_name, Name(metric="ap_association.totalUnassociatedDiaObjects")) 

327 self.assertEqual(meas.quantity, 42 * u.count) 

328 

329 def testAllAssociated(self): 

330 with unittest.mock.patch( 

331 "lsst.dax.apdb.Apdb.from_uri", 

332 return_value=unittest.mock.Mock(Apdb, **{"countUnassociatedObjects.return_value": 0}) 

333 ): 

334 result = self.task.run([Config()]) 

335 lsst.pipe.base.testUtils.assertValidOutput(self.task, result) 

336 meas = result.measurement 

337 

338 self.assertEqual(meas.metric_name, Name(metric="ap_association.totalUnassociatedDiaObjects")) 

339 self.assertEqual(meas.quantity, 0.0 * u.count) 

340 

341 def testFineGrainedMetric(self): 

342 with self.assertRaises(ValueError): 

343 self.task.run([Config()], outputDataId={"visit": 42}) 

344 

345 

346# Hack around unittest's hacky test setup system 

347del MetricTaskTestCase 

348del MetadataMetricTestCase 

349del ApdbMetricTestCase 

350 

351 

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

353 pass 

354 

355 

356def setup_module(module): 

357 lsst.utils.tests.init() 

358 

359 

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

361 lsst.utils.tests.init() 

362 unittest.main()