Coverage for tests / test_updateRecord.py: 19%

1# This file is part of dax_apdb. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://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 <http://www.gnu.org/licenses/>. 

21 

22import json 

23import unittest 

24 

25from lsst.dax.apdb.apdbUpdateRecord import ( 

26 ApdbCloseDiaObjectValidityRecord, 

27 ApdbReassignDiaSourceToDiaObjectRecord, 

28 ApdbReassignDiaSourceToSSObjectRecord, 

29 ApdbUpdateNDiaSourcesRecord, 

30 ApdbUpdateRecord, 

31 ApdbWithdrawDiaForcedSourceRecord, 

32 ApdbWithdrawDiaSourceRecord, 

33) 

34 

35 

36class ApdbUpdateRecordTestCase(unittest.TestCase): 

37 """A test case for ApdbUpdateRecord sub-classes.""" 

38 

39 update_time_ns1 = 2_000_000_000_000_000_000 

40 update_time_ns2 = 2_000_000_001_000_000_000 

41 

42 def test_reassign_diasource_to_ssobject(self) -> None: 

43 """Test round-tripping ApdbReassignDiaSourceToSSObjectRecord class.""" 

44 record = ApdbReassignDiaSourceToSSObjectRecord( 

45 update_time_ns=self.update_time_ns1, 

46 update_order=0, 

47 diaSourceId=123456, 

48 ssObjectId=1, 

49 ssObjectReassocTimeMjdTai=60000.0, 

50 ra=45.0, 

51 dec=-45.0, 

52 midpointMjdTai=59999.0, 

53 ) 

54 record_json = record.to_json() 

55 record_dict = json.loads(record_json) 

56 self.assertEqual( 

57 record_dict, 

58 { 

59 "diaSourceId": 123456, 

60 "ssObjectId": 1, 

61 "ssObjectReassocTimeMjdTai": 60000.0, 

62 "ra": 45.0, 

63 "dec": -45.0, 

64 "midpointMjdTai": 59999.0, 

65 "update_type": "reassign_diasource_to_ssobject", 

66 }, 

67 ) 

68 

69 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json) 

70 self.assertIsInstance(record2, ApdbReassignDiaSourceToSSObjectRecord) 

71 self.assertEqual(record2, record) 

72 

73 def test_reassign_diasource_to_diaobject(self) -> None: 

74 """Test round-tripping ApdbReassignDiaSourceToDiaObjectRecord class.""" 

75 record = ApdbReassignDiaSourceToDiaObjectRecord( 

76 update_time_ns=self.update_time_ns1, 

77 update_order=0, 

78 diaSourceId=123456, 

79 diaObjectId=321, 

80 ra=45.0, 

81 dec=-45.0, 

82 midpointMjdTai=59999.0, 

83 ) 

84 record_json = record.to_json() 

85 record_dict = json.loads(record_json) 

86 self.assertEqual( 

87 record_dict, 

88 { 

89 "diaSourceId": 123456, 

90 "diaObjectId": 321, 

91 "ra": 45.0, 

92 "dec": -45.0, 

93 "midpointMjdTai": 59999.0, 

94 "update_type": "reassign_diasource_to_diaobject", 

95 }, 

96 ) 

97 

98 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json) 

99 self.assertIsInstance(record2, ApdbReassignDiaSourceToDiaObjectRecord) 

100 self.assertEqual(record2, record) 

101 

102 def test_close_diaobject_validity(self) -> None: 

103 """Test round-tripping ApdbCloseDiaObjectValidityRecord class.""" 

104 record = ApdbCloseDiaObjectValidityRecord( 

105 update_time_ns=self.update_time_ns1, 

106 update_order=0, 

107 diaObjectId=321, 

108 validityEndMjdTai=60000.0, 

109 nDiaSources=None, 

110 ra=45.0, 

111 dec=-45.0, 

112 ) 

113 record_json = record.to_json() 

114 record_dict = json.loads(record_json) 

115 self.assertEqual( 

116 record_dict, 

117 { 

118 "diaObjectId": 321, 

119 "validityEndMjdTai": 60000.0, 

120 "nDiaSources": None, 

121 "ra": 45.0, 

122 "dec": -45.0, 

123 "update_type": "close_diaobject_validity", 

124 }, 

125 ) 

126 

127 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json) 

128 self.assertIsInstance(record2, ApdbCloseDiaObjectValidityRecord) 

129 self.assertEqual(record2, record) 

130 

131 def test_update_n_dia_sources(self) -> None: 

132 """Test round-tripping ApdbUpdateNDiaSourcesRecord class.""" 

133 record = ApdbUpdateNDiaSourcesRecord( 

134 update_time_ns=self.update_time_ns1, 

135 update_order=0, 

136 diaObjectId=321, 

137 nDiaSources=6, 

138 ra=45.0, 

139 dec=-45.0, 

140 ) 

141 record_json = record.to_json() 

142 record_dict = json.loads(record_json) 

143 self.assertEqual( 

144 record_dict, 

145 { 

146 "diaObjectId": 321, 

147 "nDiaSources": 6, 

148 "ra": 45.0, 

149 "dec": -45.0, 

150 "update_type": "update_n_dia_sources", 

151 }, 

152 ) 

153 

154 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json) 

155 self.assertIsInstance(record2, ApdbUpdateNDiaSourcesRecord) 

156 self.assertEqual(record2, record) 

157 

158 def test_withdraw_diasource(self) -> None: 

159 """Test round-tripping ApdbWithdrawDiaSourceRecord class.""" 

160 record = ApdbWithdrawDiaSourceRecord( 

161 update_time_ns=self.update_time_ns1, 

162 update_order=0, 

163 diaSourceId=123456, 

164 timeWithdrawnMjdTai=61000.0, 

165 ra=45.0, 

166 dec=-45.0, 

167 midpointMjdTai=60000.0, 

168 ) 

169 record_json = record.to_json() 

170 record_dict = json.loads(record_json) 

171 self.assertEqual( 

172 record_dict, 

173 { 

174 "diaSourceId": 123456, 

175 "timeWithdrawnMjdTai": 61000.0, 

176 "ra": 45.0, 

177 "dec": -45.0, 

178 "midpointMjdTai": 60000.0, 

179 "update_type": "withdraw_diasource", 

180 }, 

181 ) 

182 

183 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json) 

184 self.assertIsInstance(record2, ApdbWithdrawDiaSourceRecord) 

185 self.assertEqual(record2, record) 

186 

187 def test_withdraw_diaforcedsource(self) -> None: 

188 """Test round-tripping ApdbWithdrawDiaForcedSourceRecord class.""" 

189 record = ApdbWithdrawDiaForcedSourceRecord( 

190 update_time_ns=self.update_time_ns1, 

191 update_order=0, 

192 diaObjectId=1234, 

193 visit=555, 

194 detector=123, 

195 timeWithdrawnMjdTai=61000.0, 

196 ra=45.0, 

197 dec=-45.0, 

198 midpointMjdTai=60000.0, 

199 ) 

200 record_json = record.to_json() 

201 record_dict = json.loads(record_json) 

202 self.assertEqual( 

203 record_dict, 

204 { 

205 "diaObjectId": 1234, 

206 "visit": 555, 

207 "detector": 123, 

208 "timeWithdrawnMjdTai": 61000.0, 

209 "ra": 45.0, 

210 "dec": -45.0, 

211 "midpointMjdTai": 60000.0, 

212 "update_type": "withdraw_diaforcedsource", 

213 }, 

214 ) 

215 

216 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json) 

217 self.assertIsInstance(record2, ApdbWithdrawDiaForcedSourceRecord) 

218 self.assertEqual(record2, record) 

219 

220 def test_ordering(self) -> None: 

221 """Test ordering of records.""" 

222 record1 = ApdbReassignDiaSourceToSSObjectRecord( 

223 update_time_ns=self.update_time_ns1, 

224 update_order=0, 

225 diaSourceId=1, 

226 ssObjectId=1, 

227 ssObjectReassocTimeMjdTai=60000.0, 

228 ra=45.0, 

229 dec=-45.0, 

230 midpointMjdTai=60000.0, 

231 ) 

232 record2 = ApdbWithdrawDiaSourceRecord( 

233 update_time_ns=self.update_time_ns1, 

234 update_order=1, 

235 diaSourceId=123456, 

236 timeWithdrawnMjdTai=61000.0, 

237 ra=45.0, 

238 dec=-45.0, 

239 midpointMjdTai=60000.0, 

240 ) 

241 record3 = ApdbReassignDiaSourceToSSObjectRecord( 

242 update_time_ns=self.update_time_ns1, 

243 update_order=3, 

244 diaSourceId=2, 

245 ssObjectId=3, 

246 ssObjectReassocTimeMjdTai=60000.0, 

247 ra=45.0, 

248 dec=-45.0, 

249 midpointMjdTai=60000.0, 

250 ) 

251 record4 = ApdbWithdrawDiaSourceRecord( 

252 update_time_ns=self.update_time_ns2, 

253 update_order=0, 

254 diaSourceId=123456, 

255 timeWithdrawnMjdTai=61000.0, 

256 ra=45.0, 

257 dec=-45.0, 

258 midpointMjdTai=60000.0, 

259 ) 

260 record5 = ApdbWithdrawDiaForcedSourceRecord( 

261 update_time_ns=self.update_time_ns2, 

262 update_order=1, 

263 diaObjectId=1234, 

264 visit=555, 

265 detector=123, 

266 timeWithdrawnMjdTai=61000.0, 

267 ra=45.0, 

268 dec=-45.0, 

269 midpointMjdTai=60000.0, 

270 ) 

271 

272 unordered = [record5, record3, record1, record4, record2] 

273 ordered = sorted(unordered) 

274 self.assertTrue(ordered[0] < ordered[1] < ordered[2] < ordered[3] < ordered[4]) 

275 self.assertIs(ordered[0], record1) 

276 self.assertIs(ordered[1], record2) 

277 self.assertIs(ordered[2], record3) 

278 self.assertIs(ordered[3], record4) 

279 self.assertIs(ordered[4], record5) 

280 

281 with self.assertRaisesRegex(TypeError, "ordering is not supported between"): 

282 record1 < (1, 2, 3) 

283 

284 

285if __name__ == "__main__": 

286 unittest.main()