Coverage for tests / test_updateRecord.py: 19%
70 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-27 01:29 -0800
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_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 )
69 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json)
70 self.assertIsInstance(record2, ApdbReassignDiaSourceToSSObjectRecord)
71 self.assertEqual(record2, record)
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 )
98 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json)
99 self.assertIsInstance(record2, ApdbReassignDiaSourceToDiaObjectRecord)
100 self.assertEqual(record2, record)
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 )
127 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json)
128 self.assertIsInstance(record2, ApdbCloseDiaObjectValidityRecord)
129 self.assertEqual(record2, record)
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 )
154 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json)
155 self.assertIsInstance(record2, ApdbUpdateNDiaSourcesRecord)
156 self.assertEqual(record2, record)
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 )
183 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json)
184 self.assertIsInstance(record2, ApdbWithdrawDiaSourceRecord)
185 self.assertEqual(record2, record)
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 )
216 record2 = ApdbUpdateRecord.from_json(self.update_time_ns1, 0, record_json)
217 self.assertIsInstance(record2, ApdbWithdrawDiaForcedSourceRecord)
218 self.assertEqual(record2, record)
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 )
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)
281 with self.assertRaisesRegex(TypeError, "ordering is not supported between"):
282 record1 < (1, 2, 3)
285if __name__ == "__main__":
286 unittest.main()