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 ApdbReassignDiaSourceRecord, 

28 ApdbUpdateNDiaSourcesRecord, 

29 ApdbUpdateRecord, 

30 ApdbWithdrawDiaForcedSourceRecord, 

31 ApdbWithdrawDiaSourceRecord, 

32) 

33 

34 

35class ApdbUpdateRecordTestCase(unittest.TestCase): 

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

37 

38 update_time_ns1 = 2_000_000_000_000_000_000 

39 update_time_ns2 = 2_000_000_001_000_000_000 

40 

41 def test_reassign_diasource(self) -> None: 

42 """Test round-tripping ApdbReassignDiaSourceRecord class.""" 

43 record = ApdbReassignDiaSourceRecord( 

44 update_time_ns=self.update_time_ns1, 

45 update_order=0, 

46 diaSourceId=123456, 

47 diaObjectId=321, 

48 ssObjectId=1, 

49 ssObjectReassocTimeMjdTai=60000.0, 

50 ra=45.0, 

51 dec=-45.0, 

52 ) 

53 record_json = record.to_json() 

54 record_dict = json.loads(record_json) 

55 self.assertEqual( 

56 record_dict, 

57 { 

58 "diaSourceId": 123456, 

59 "diaObjectId": 321, 

60 "ssObjectId": 1, 

61 "ssObjectReassocTimeMjdTai": 60000.0, 

62 "ra": 45.0, 

63 "dec": -45.0, 

64 "update_type": "reassign_diasource", 

65 }, 

66 ) 

67 

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

69 self.assertIsInstance(record2, ApdbReassignDiaSourceRecord) 

70 self.assertEqual(record2, record) 

71 

72 def test_close_diaobject_validity(self) -> None: 

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

74 record = ApdbCloseDiaObjectValidityRecord( 

75 update_time_ns=self.update_time_ns1, 

76 update_order=0, 

77 diaObjectId=321, 

78 validityEndMjdTai=60000.0, 

79 nDiaSources=None, 

80 ra=45.0, 

81 dec=-45.0, 

82 ) 

83 record_json = record.to_json() 

84 record_dict = json.loads(record_json) 

85 self.assertEqual( 

86 record_dict, 

87 { 

88 "diaObjectId": 321, 

89 "validityEndMjdTai": 60000.0, 

90 "nDiaSources": None, 

91 "ra": 45.0, 

92 "dec": -45.0, 

93 "update_type": "close_diaobject_validity", 

94 }, 

95 ) 

96 

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

98 self.assertIsInstance(record2, ApdbCloseDiaObjectValidityRecord) 

99 self.assertEqual(record2, record) 

100 

101 def test_update_n_dia_sources(self) -> None: 

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

103 record = ApdbUpdateNDiaSourcesRecord( 

104 update_time_ns=self.update_time_ns1, 

105 update_order=0, 

106 diaObjectId=321, 

107 nDiaSources=6, 

108 ra=45.0, 

109 dec=-45.0, 

110 ) 

111 record_json = record.to_json() 

112 record_dict = json.loads(record_json) 

113 self.assertEqual( 

114 record_dict, 

115 { 

116 "diaObjectId": 321, 

117 "nDiaSources": 6, 

118 "ra": 45.0, 

119 "dec": -45.0, 

120 "update_type": "update_n_dia_sources", 

121 }, 

122 ) 

123 

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

125 self.assertIsInstance(record2, ApdbUpdateNDiaSourcesRecord) 

126 self.assertEqual(record2, record) 

127 

128 def test_withdraw_diasource(self) -> None: 

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

130 record = ApdbWithdrawDiaSourceRecord( 

131 update_time_ns=self.update_time_ns1, 

132 update_order=0, 

133 diaSourceId=123456, 

134 diaObjectId=321, 

135 timeWithdrawnMjdTai=61000.0, 

136 ra=45.0, 

137 dec=-45.0, 

138 ) 

139 record_json = record.to_json() 

140 record_dict = json.loads(record_json) 

141 self.assertEqual( 

142 record_dict, 

143 { 

144 "diaSourceId": 123456, 

145 "diaObjectId": 321, 

146 "timeWithdrawnMjdTai": 61000.0, 

147 "ra": 45.0, 

148 "dec": -45.0, 

149 "update_type": "withdraw_diasource", 

150 }, 

151 ) 

152 

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

154 self.assertIsInstance(record2, ApdbWithdrawDiaSourceRecord) 

155 self.assertEqual(record2, record) 

156 

157 def test_withdraw_diaforcedsource(self) -> None: 

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

159 record = ApdbWithdrawDiaForcedSourceRecord( 

160 update_time_ns=self.update_time_ns1, 

161 update_order=0, 

162 diaObjectId=1234, 

163 visit=555, 

164 detector=123, 

165 timeWithdrawnMjdTai=61000.0, 

166 ra=45.0, 

167 dec=-45.0, 

168 ) 

169 record_json = record.to_json() 

170 record_dict = json.loads(record_json) 

171 self.assertEqual( 

172 record_dict, 

173 { 

174 "diaObjectId": 1234, 

175 "visit": 555, 

176 "detector": 123, 

177 "timeWithdrawnMjdTai": 61000.0, 

178 "ra": 45.0, 

179 "dec": -45.0, 

180 "update_type": "withdraw_diaforcedsource", 

181 }, 

182 ) 

183 

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

185 self.assertIsInstance(record2, ApdbWithdrawDiaForcedSourceRecord) 

186 self.assertEqual(record2, record) 

187 

188 def test_ordering(self) -> None: 

189 """Test ordering of records.""" 

190 record1 = ApdbReassignDiaSourceRecord( 

191 update_time_ns=self.update_time_ns1, 

192 update_order=0, 

193 diaSourceId=1, 

194 diaObjectId=321, 

195 ssObjectId=1, 

196 ssObjectReassocTimeMjdTai=60000.0, 

197 ra=45.0, 

198 dec=-45.0, 

199 ) 

200 record2 = ApdbWithdrawDiaSourceRecord( 

201 update_time_ns=self.update_time_ns1, 

202 update_order=1, 

203 diaSourceId=123456, 

204 diaObjectId=321, 

205 timeWithdrawnMjdTai=61000.0, 

206 ra=45.0, 

207 dec=-45.0, 

208 ) 

209 record3 = ApdbReassignDiaSourceRecord( 

210 update_time_ns=self.update_time_ns1, 

211 update_order=3, 

212 diaSourceId=2, 

213 diaObjectId=3, 

214 ssObjectId=3, 

215 ssObjectReassocTimeMjdTai=60000.0, 

216 ra=45.0, 

217 dec=-45.0, 

218 ) 

219 record4 = ApdbWithdrawDiaSourceRecord( 

220 update_time_ns=self.update_time_ns2, 

221 update_order=0, 

222 diaSourceId=123456, 

223 diaObjectId=321, 

224 timeWithdrawnMjdTai=61000.0, 

225 ra=45.0, 

226 dec=-45.0, 

227 ) 

228 record5 = ApdbWithdrawDiaForcedSourceRecord( 

229 update_time_ns=self.update_time_ns2, 

230 update_order=1, 

231 diaObjectId=1234, 

232 visit=555, 

233 detector=123, 

234 timeWithdrawnMjdTai=61000.0, 

235 ra=45.0, 

236 dec=-45.0, 

237 ) 

238 

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

240 ordered = sorted(unordered) 

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

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

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

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

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

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

247 

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

249 record1 < (1, 2, 3) 

250 

251 

252if __name__ == "__main__": 

253 unittest.main()