Coverage for tests / test_updateRecord.py: 16%
87 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-01 08:19 +0000
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/>.
22import json
23import unittest
25from lsst.dax.apdb.apdbUpdateRecord import (
26 ApdbCloseDiaObjectValidityRecord,
27 ApdbReassignDiaSourceToDiaObjectRecord,
28 ApdbReassignDiaSourceToSSObjectRecord,
29 ApdbUpdateNDiaSourcesRecord,
30 ApdbUpdateRecord,
31 ApdbWithdrawDiaForcedSourceRecord,
32 ApdbWithdrawDiaSourceRecord,
33)
36class ApdbUpdateRecordTestCase(unittest.TestCase):
37 """A test case for ApdbUpdateRecord sub-classes."""
39 update_time_ns1 = 2_000_000_000_000_000_000
40 update_time_ns2 = 2_000_000_001_000_000_000
42 def test_all_types(self) -> None:
43 """Test that we know full set of update types."""
44 self.assertEqual(
45 set(ApdbUpdateRecord._update_types),
46 {
47 "reassign_diasource_to_diaobject",
48 "reassign_diasource_to_ssobject",
49 "withdraw_diasource",
50 "withdraw_diaforcedsource",
51 "close_diaobject_validity",
52 "update_n_dia_sources",
53 },
54 )
56 def test_reassign_diasource_to_ssobject(self) -> None:
57 """Test round-tripping ApdbReassignDiaSourceToSSObjectRecord class."""
58 record = ApdbReassignDiaSourceToSSObjectRecord(
59 update_time_ns=self.update_time_ns1,
60 update_order=0,
61 diaSourceId=123456,
62 ssObjectId=1,
63 ssObjectReassocTimeMjdTai=60000.0,
64 ra=45.0,
65 dec=-45.0,
66 midpointMjdTai=59999.0,
67 )
68 self.assertEqual(record.record_id(), (("diaSourceId", 123456),))
69 self.assertEqual(record.record_payload(), (("ssObjectId", 1), ("ssObjectReassocTimeMjdTai", 60000.0)))
71 record_json = record.to_json()
72 record_dict = json.loads(record_json)
73 self.assertEqual(
74 record_dict,
75 {
76 "diaSourceId": 123456,
77 "ssObjectId": 1,
78 "ssObjectReassocTimeMjdTai": 60000.0,
79 "ra": 45.0,
80 "dec": -45.0,
81 "midpointMjdTai": 59999.0,
82 "update_type": "reassign_diasource_to_ssobject",
83 },
84 )
86 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json)
87 self.assertIsInstance(record2, ApdbReassignDiaSourceToSSObjectRecord)
88 self.assertEqual(record2, record)
90 def test_reassign_diasource_to_diaobject(self) -> None:
91 """Test round-tripping ApdbReassignDiaSourceToDiaObjectRecord class."""
92 record = ApdbReassignDiaSourceToDiaObjectRecord(
93 update_time_ns=self.update_time_ns1,
94 update_order=0,
95 diaSourceId=123456,
96 diaObjectId=321,
97 ra=45.0,
98 dec=-45.0,
99 midpointMjdTai=59999.0,
100 )
101 self.assertEqual(record.record_id(), (("diaSourceId", 123456),))
102 self.assertEqual(record.record_payload(), (("diaObjectId", 321),))
104 record_json = record.to_json()
105 record_dict = json.loads(record_json)
106 self.assertEqual(
107 record_dict,
108 {
109 "diaSourceId": 123456,
110 "diaObjectId": 321,
111 "ra": 45.0,
112 "dec": -45.0,
113 "midpointMjdTai": 59999.0,
114 "update_type": "reassign_diasource_to_diaobject",
115 },
116 )
118 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json)
119 self.assertIsInstance(record2, ApdbReassignDiaSourceToDiaObjectRecord)
120 self.assertEqual(record2, record)
122 def test_close_diaobject_validity(self) -> None:
123 """Test round-tripping ApdbCloseDiaObjectValidityRecord class."""
124 record = ApdbCloseDiaObjectValidityRecord(
125 update_time_ns=self.update_time_ns1,
126 update_order=0,
127 diaObjectId=321,
128 validityEndMjdTai=60000.0,
129 nDiaSources=None,
130 ra=45.0,
131 dec=-45.0,
132 )
133 self.assertEqual(record.record_id(), (("diaObjectId", 321),))
134 self.assertEqual(record.record_payload(), (("validityEndMjdTai", 60000.0),))
136 record_json = record.to_json()
137 record_dict = json.loads(record_json)
138 self.assertEqual(
139 record_dict,
140 {
141 "diaObjectId": 321,
142 "validityEndMjdTai": 60000.0,
143 "nDiaSources": None,
144 "ra": 45.0,
145 "dec": -45.0,
146 "update_type": "close_diaobject_validity",
147 },
148 )
150 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json)
151 self.assertIsInstance(record2, ApdbCloseDiaObjectValidityRecord)
152 self.assertEqual(record2, record)
154 # Also check record_payload with nDiaSources != None
155 record = ApdbCloseDiaObjectValidityRecord(
156 update_time_ns=self.update_time_ns1,
157 update_order=0,
158 diaObjectId=321,
159 validityEndMjdTai=60000.0,
160 nDiaSources=3,
161 ra=45.0,
162 dec=-45.0,
163 )
164 self.assertEqual(record.record_id(), (("diaObjectId", 321),))
165 self.assertEqual(record.record_payload(), (("validityEndMjdTai", 60000.0), ("nDiaSources", 3)))
167 def test_update_n_dia_sources(self) -> None:
168 """Test round-tripping ApdbUpdateNDiaSourcesRecord class."""
169 record = ApdbUpdateNDiaSourcesRecord(
170 update_time_ns=self.update_time_ns1,
171 update_order=0,
172 diaObjectId=321,
173 nDiaSources=6,
174 ra=45.0,
175 dec=-45.0,
176 )
177 self.assertEqual(record.record_id(), (("diaObjectId", 321),))
178 self.assertEqual(record.record_payload(), (("nDiaSources", 6),))
180 record_json = record.to_json()
181 record_dict = json.loads(record_json)
182 self.assertEqual(
183 record_dict,
184 {
185 "diaObjectId": 321,
186 "nDiaSources": 6,
187 "ra": 45.0,
188 "dec": -45.0,
189 "update_type": "update_n_dia_sources",
190 },
191 )
193 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json)
194 self.assertIsInstance(record2, ApdbUpdateNDiaSourcesRecord)
195 self.assertEqual(record2, record)
197 def test_withdraw_diasource(self) -> None:
198 """Test round-tripping ApdbWithdrawDiaSourceRecord class."""
199 record = ApdbWithdrawDiaSourceRecord(
200 update_time_ns=self.update_time_ns1,
201 update_order=0,
202 diaSourceId=123456,
203 timeWithdrawnMjdTai=61000.0,
204 ra=45.0,
205 dec=-45.0,
206 midpointMjdTai=60000.0,
207 )
208 self.assertEqual(record.record_id(), (("diaSourceId", 123456),))
209 self.assertEqual(record.record_payload(), (("timeWithdrawnMjdTai", 61000.0),))
211 record_json = record.to_json()
212 record_dict = json.loads(record_json)
213 self.assertEqual(
214 record_dict,
215 {
216 "diaSourceId": 123456,
217 "timeWithdrawnMjdTai": 61000.0,
218 "ra": 45.0,
219 "dec": -45.0,
220 "midpointMjdTai": 60000.0,
221 "update_type": "withdraw_diasource",
222 },
223 )
225 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json)
226 self.assertIsInstance(record2, ApdbWithdrawDiaSourceRecord)
227 self.assertEqual(record2, record)
229 def test_withdraw_diaforcedsource(self) -> None:
230 """Test round-tripping ApdbWithdrawDiaForcedSourceRecord class."""
231 record = ApdbWithdrawDiaForcedSourceRecord(
232 update_time_ns=self.update_time_ns1,
233 update_order=0,
234 diaObjectId=1234,
235 visit=555,
236 detector=123,
237 timeWithdrawnMjdTai=61000.0,
238 ra=45.0,
239 dec=-45.0,
240 midpointMjdTai=60000.0,
241 )
242 self.assertEqual(record.record_id(), (("diaObjectId", 1234), ("visit", 555), ("detector", 123)))
243 self.assertEqual(record.record_payload(), (("timeWithdrawnMjdTai", 61000.0),))
245 record_json = record.to_json()
246 record_dict = json.loads(record_json)
247 self.assertEqual(
248 record_dict,
249 {
250 "diaObjectId": 1234,
251 "visit": 555,
252 "detector": 123,
253 "timeWithdrawnMjdTai": 61000.0,
254 "ra": 45.0,
255 "dec": -45.0,
256 "midpointMjdTai": 60000.0,
257 "update_type": "withdraw_diaforcedsource",
258 },
259 )
261 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json)
262 self.assertIsInstance(record2, ApdbWithdrawDiaForcedSourceRecord)
263 self.assertEqual(record2, record)
265 def test_ordering(self) -> None:
266 """Test ordering of records."""
267 record1 = ApdbReassignDiaSourceToSSObjectRecord(
268 update_time_ns=self.update_time_ns1,
269 update_order=0,
270 diaSourceId=1,
271 ssObjectId=1,
272 ssObjectReassocTimeMjdTai=60000.0,
273 ra=45.0,
274 dec=-45.0,
275 midpointMjdTai=60000.0,
276 )
277 record2 = ApdbWithdrawDiaSourceRecord(
278 update_time_ns=self.update_time_ns1,
279 update_order=1,
280 diaSourceId=123456,
281 timeWithdrawnMjdTai=61000.0,
282 ra=45.0,
283 dec=-45.0,
284 midpointMjdTai=60000.0,
285 )
286 record3 = ApdbReassignDiaSourceToSSObjectRecord(
287 update_time_ns=self.update_time_ns1,
288 update_order=3,
289 diaSourceId=2,
290 ssObjectId=3,
291 ssObjectReassocTimeMjdTai=60000.0,
292 ra=45.0,
293 dec=-45.0,
294 midpointMjdTai=60000.0,
295 )
296 record4 = ApdbWithdrawDiaSourceRecord(
297 update_time_ns=self.update_time_ns2,
298 update_order=0,
299 diaSourceId=123456,
300 timeWithdrawnMjdTai=61000.0,
301 ra=45.0,
302 dec=-45.0,
303 midpointMjdTai=60000.0,
304 )
305 record5 = ApdbWithdrawDiaForcedSourceRecord(
306 update_time_ns=self.update_time_ns2,
307 update_order=1,
308 diaObjectId=1234,
309 visit=555,
310 detector=123,
311 timeWithdrawnMjdTai=61000.0,
312 ra=45.0,
313 dec=-45.0,
314 midpointMjdTai=60000.0,
315 )
317 unordered = [record5, record3, record1, record4, record2]
318 ordered = sorted(unordered)
319 self.assertTrue(ordered[0] < ordered[1] < ordered[2] < ordered[3] < ordered[4])
320 self.assertIs(ordered[0], record1)
321 self.assertIs(ordered[1], record2)
322 self.assertIs(ordered[2], record3)
323 self.assertIs(ordered[3], record4)
324 self.assertIs(ordered[4], record5)
326 with self.assertRaisesRegex(TypeError, "ordering is not supported between"):
327 record1 < (1, 2, 3)
330if __name__ == "__main__":
331 unittest.main()