LSST Applications g0265f82a02+c6dfa2ddaf,g1162b98a3f+ffe7eabc7e,g2079a07aa2+1b2e822518,g2bbee38e9b+c6dfa2ddaf,g337abbeb29+c6dfa2ddaf,g36da64cc00+ea84795170,g3ddfee87b4+955a963fd8,g50ff169b8f+2eb0e556e8,g52b1c1532d+90ebb246c7,g555ede804d+955a963fd8,g591dd9f2cf+bac198a2cb,g5ec818987f+420292cfeb,g858d7b2824+d6c9a0a3b8,g876c692160+aabc49a3c3,g8a8a8dda67+90ebb246c7,g8cdfe0ae6a+4fd9e222a8,g99cad8db69+e6cd765486,g9ddcbc5298+a1346535a5,ga1e77700b3+df8f93165b,ga8c6da7877+acd47f83f4,gae46bcf261+c6dfa2ddaf,gb0e22166c9+8634eb87fb,gb3f2274832+12c8382528,gba4ed39666+1ac82b564f,gbb8dafda3b+0574160a1f,gbeb006f7da+dea2fbb49f,gc28159a63d+c6dfa2ddaf,gc86a011abf+d6c9a0a3b8,gcf0d15dbbd+955a963fd8,gdaeeff99f8+1cafcb7cd4,gdc0c513512+d6c9a0a3b8,ge79ae78c31+c6dfa2ddaf,geb67518f79+ba1859f325,gee10cc3b42+90ebb246c7,gf1cff7945b+d6c9a0a3b8,w.2024.13
LSST Data Management Base Package
Loading...
Searching...
No Matches
Public Member Functions | Public Attributes | Static Public Attributes | List of all members
lsst.dax.apdb.tests._apdb.ApdbTest Class Reference
Inheritance diagram for lsst.dax.apdb.tests._apdb.ApdbTest:
lsst.dax.apdb.tests._apdb.TestCaseMixin

Public Member Functions

ApdbConfig make_config (self, **Any kwargs)
 
ApdbTables getDiaObjects_table (self)
 
None assert_catalog (self, Any catalog, int rows, ApdbTables table)
 
None assert_table_data (self, Any catalog, int rows, ApdbTables table)
 
None test_makeSchema (self)
 
None test_empty_gets (self)
 
None test_empty_gets_0months (self)
 
None test_storeObjects (self)
 
None test_storeObjects_empty (self)
 
None test_storeSources (self)
 
None test_storeForcedSources (self)
 
None test_getHistory (self)
 
None test_storeSSObjects (self)
 
None test_reassignObjects (self)
 
None test_midpointMjdTai_src (self)
 
None test_midpointMjdTai_fsrc (self)
 
None test_metadata (self)
 
None test_nometadata (self)
 
None test_schemaVersionFromYaml (self)
 
None test_config_freeze (self)
 

Public Attributes

 use_insert_id
 

Static Public Attributes

bool time_partition_tables = False
 
 visit_time = astropy.time.Time("2021-01-01T00:00:00", format="isot", scale="tai")
 
bool fsrc_requires_id_list = False
 
bool use_insert_id = False
 
bool allow_visit_query = True
 
dict table_column_count
 

Detailed Description

Base class for Apdb tests that can be specialized for concrete
implementation.

This can only be used as a mixin class for a unittest.TestCase and it
calls various assert methods.

Definition at line 114 of file _apdb.py.

Member Function Documentation

◆ assert_catalog()

None lsst.dax.apdb.tests._apdb.ApdbTest.assert_catalog ( self,
Any catalog,
int rows,
ApdbTables table )
Validate catalog type and size

Parameters
----------
catalog : `object`
    Expected type of this is ``pandas.DataFrame``.
rows : `int`
    Expected number of rows in a catalog.
table : `ApdbTables`
    APDB table type.

Definition at line 153 of file _apdb.py.

153 def assert_catalog(self, catalog: Any, rows: int, table: ApdbTables) -> None:
154 """Validate catalog type and size
155
156 Parameters
157 ----------
158 catalog : `object`
159 Expected type of this is ``pandas.DataFrame``.
160 rows : `int`
161 Expected number of rows in a catalog.
162 table : `ApdbTables`
163 APDB table type.
164 """
165 self.assertIsInstance(catalog, pandas.DataFrame)
166 self.assertEqual(catalog.shape[0], rows)
167 self.assertEqual(catalog.shape[1], self.table_column_count[table])
168

◆ assert_table_data()

None lsst.dax.apdb.tests._apdb.ApdbTest.assert_table_data ( self,
Any catalog,
int rows,
ApdbTables table )
Validate catalog type and size

Parameters
----------
catalog : `object`
    Expected type of this is `ApdbTableData`.
rows : `int`
    Expected number of rows in a catalog.
table : `ApdbTables`
    APDB table type.
extra_columns : `int`
    Count of additional columns expected in ``catalog``.

Definition at line 169 of file _apdb.py.

169 def assert_table_data(self, catalog: Any, rows: int, table: ApdbTables) -> None:
170 """Validate catalog type and size
171
172 Parameters
173 ----------
174 catalog : `object`
175 Expected type of this is `ApdbTableData`.
176 rows : `int`
177 Expected number of rows in a catalog.
178 table : `ApdbTables`
179 APDB table type.
180 extra_columns : `int`
181 Count of additional columns expected in ``catalog``.
182 """
183 self.assertIsInstance(catalog, ApdbTableData)
184 n_rows = sum(1 for row in catalog.rows())
185 self.assertEqual(n_rows, rows)
186 # One extra column for insert_id
187 self.assertEqual(len(catalog.column_names()), self.table_column_count[table] + 1)
188

◆ getDiaObjects_table()

ApdbTables lsst.dax.apdb.tests._apdb.ApdbTest.getDiaObjects_table ( self)
Return type of table returned from getDiaObjects method.

Definition at line 149 of file _apdb.py.

149 def getDiaObjects_table(self) -> ApdbTables:
150 """Return type of table returned from getDiaObjects method."""
151 raise NotImplementedError()
152

◆ make_config()

ApdbConfig lsst.dax.apdb.tests._apdb.ApdbTest.make_config ( self,
**Any kwargs )
Make config class instance used in all tests.

Definition at line 144 of file _apdb.py.

144 def make_config(self, **kwargs: Any) -> ApdbConfig:
145 """Make config class instance used in all tests."""
146 raise NotImplementedError()
147

◆ test_config_freeze()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_config_freeze ( self)
Test that some config fields are correctly frozen in database.

Definition at line 700 of file _apdb.py.

700 def test_config_freeze(self) -> None:
701 """Test that some config fields are correctly frozen in database."""
702 config = self.make_config()
703 Apdb.makeSchema(config)
704
705 # `use_insert_id` is the only parameter that is frozen in all
706 # implementations.
707 config.use_insert_id = not self.use_insert_id
708 apdb = make_apdb(config)
709 frozen_config = apdb.config # type: ignore[attr-defined]
710 self.assertEqual(frozen_config.use_insert_id, self.use_insert_id)
711
712

◆ test_empty_gets()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_empty_gets ( self)
Test for getting data from empty database.

All get() methods should return empty results, only useful for
checking that code is not broken.

Definition at line 201 of file _apdb.py.

201 def test_empty_gets(self) -> None:
202 """Test for getting data from empty database.
203
204 All get() methods should return empty results, only useful for
205 checking that code is not broken.
206 """
207 # use non-zero months for Forced/Source fetching
208 config = self.make_config()
209 Apdb.makeSchema(config)
210 apdb = make_apdb(config)
211
212 region = _make_region()
213 visit_time = self.visit_time
214
215 res: pandas.DataFrame | None
216
217 # get objects by region
218 res = apdb.getDiaObjects(region)
219 self.assert_catalog(res, 0, self.getDiaObjects_table())
220
221 # get sources by region
222 res = apdb.getDiaSources(region, None, visit_time)
223 self.assert_catalog(res, 0, ApdbTables.DiaSource)
224
225 res = apdb.getDiaSources(region, [], visit_time)
226 self.assert_catalog(res, 0, ApdbTables.DiaSource)
227
228 # get sources by object ID, non-empty object list
229 res = apdb.getDiaSources(region, [1, 2, 3], visit_time)
230 self.assert_catalog(res, 0, ApdbTables.DiaSource)
231
232 # get forced sources by object ID, empty object list
233 res = apdb.getDiaForcedSources(region, [], visit_time)
234 self.assert_catalog(res, 0, ApdbTables.DiaForcedSource)
235
236 # get sources by object ID, non-empty object list
237 res = apdb.getDiaForcedSources(region, [1, 2, 3], visit_time)
238 self.assert_catalog(res, 0, ApdbTables.DiaForcedSource)
239
240 # test if a visit has objects/sources
241 if self.allow_visit_query:
242 res = apdb.containsVisitDetector(visit=0, detector=0)
243 self.assertFalse(res)
244 else:
245 with self.assertRaises(NotImplementedError):
246 apdb.containsVisitDetector(visit=0, detector=0)
247
248 # alternative method not part of the Apdb API
249 if isinstance(apdb, ApdbSql):
250 res = apdb.containsCcdVisit(1)
251 self.assertFalse(res)
252
253 # get sources by region
254 if self.fsrc_requires_id_list:
255 with self.assertRaises(NotImplementedError):
256 apdb.getDiaForcedSources(region, None, visit_time)
257 else:
258 apdb.getDiaForcedSources(region, None, visit_time)
259 self.assert_catalog(res, 0, ApdbTables.DiaForcedSource)
260

◆ test_empty_gets_0months()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_empty_gets_0months ( self)
Test for getting data from empty database.

All get() methods should return empty DataFrame or None.

Definition at line 261 of file _apdb.py.

261 def test_empty_gets_0months(self) -> None:
262 """Test for getting data from empty database.
263
264 All get() methods should return empty DataFrame or None.
265 """
266 # set read_sources_months to 0 so that Forced/Sources are None
267 config = self.make_config(read_sources_months=0, read_forced_sources_months=0)
268 Apdb.makeSchema(config)
269 apdb = make_apdb(config)
270
271 region = _make_region()
272 visit_time = self.visit_time
273
274 res: pandas.DataFrame | None
275
276 # get objects by region
277 res = apdb.getDiaObjects(region)
278 self.assert_catalog(res, 0, self.getDiaObjects_table())
279
280 # get sources by region
281 res = apdb.getDiaSources(region, None, visit_time)
282 self.assertIs(res, None)
283
284 # get sources by object ID, empty object list
285 res = apdb.getDiaSources(region, [], visit_time)
286 self.assertIs(res, None)
287
288 # get forced sources by object ID, empty object list
289 res = apdb.getDiaForcedSources(region, [], visit_time)
290 self.assertIs(res, None)
291
292 # test if a visit has objects/sources
293 if self.allow_visit_query:
294 res = apdb.containsVisitDetector(visit=0, detector=0)
295 self.assertFalse(res)
296 else:
297 with self.assertRaises(NotImplementedError):
298 apdb.containsVisitDetector(visit=0, detector=0)
299
300 # alternative method not part of the Apdb API
301 if isinstance(apdb, ApdbSql):
302 res = apdb.containsCcdVisit(1)
303 self.assertFalse(res)
304

◆ test_getHistory()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_getHistory ( self)
Store and retrieve catalog history.

Definition at line 419 of file _apdb.py.

419 def test_getHistory(self) -> None:
420 """Store and retrieve catalog history."""
421 # don't care about sources.
422 config = self.make_config()
423 Apdb.makeSchema(config)
424 apdb = make_apdb(config)
425 visit_time = self.visit_time
426
427 region1 = _make_region((1.0, 1.0, -1.0))
428 region2 = _make_region((-1.0, -1.0, -1.0))
429 nobj = 100
430 objects1 = makeObjectCatalog(region1, nobj, visit_time)
431 objects2 = makeObjectCatalog(region2, nobj, visit_time, start_id=nobj * 2)
432
433 visits = [
434 (astropy.time.Time("2021-01-01T00:01:00", format="isot", scale="tai"), objects1),
435 (astropy.time.Time("2021-01-01T00:02:00", format="isot", scale="tai"), objects2),
436 (astropy.time.Time("2021-01-01T00:03:00", format="isot", scale="tai"), objects1),
437 (astropy.time.Time("2021-01-01T00:04:00", format="isot", scale="tai"), objects2),
438 (astropy.time.Time("2021-01-01T00:05:00", format="isot", scale="tai"), objects1),
439 (astropy.time.Time("2021-01-01T00:06:00", format="isot", scale="tai"), objects2),
440 (astropy.time.Time("2021-03-01T00:01:00", format="isot", scale="tai"), objects1),
441 (astropy.time.Time("2021-03-01T00:02:00", format="isot", scale="tai"), objects2),
442 ]
443
444 start_id = 0
445 for visit_time, objects in visits:
446 sources = makeSourceCatalog(objects, visit_time, start_id=start_id)
447 fsources = makeForcedSourceCatalog(objects, visit_time, ccdVisitId=start_id)
448 apdb.store(visit_time, objects, sources, fsources)
449 start_id += nobj
450
451 insert_ids = apdb.getInsertIds()
452 if not self.use_insert_id:
453 self.assertIsNone(insert_ids)
454
455 with self.assertRaisesRegex(ValueError, "APDB is not configured for history retrieval"):
456 apdb.getDiaObjectsHistory([])
457
458 else:
459 assert insert_ids is not None
460 self.assertEqual(len(insert_ids), 8)
461
462 def _check_history(insert_ids: list[ApdbInsertId], n_records: int | None = None) -> None:
463 if n_records is None:
464 n_records = len(insert_ids) * nobj
465 res = apdb.getDiaObjectsHistory(insert_ids)
466 self.assert_table_data(res, n_records, ApdbTables.DiaObject)
467 res = apdb.getDiaSourcesHistory(insert_ids)
468 self.assert_table_data(res, n_records, ApdbTables.DiaSource)
469 res = apdb.getDiaForcedSourcesHistory(insert_ids)
470 self.assert_table_data(res, n_records, ApdbTables.DiaForcedSource)
471
472 # read it back and check sizes
473 _check_history(insert_ids)
474 _check_history(insert_ids[1:])
475 _check_history(insert_ids[1:-1])
476 _check_history(insert_ids[3:4])
477 _check_history([])
478
479 # try to remove some of those
480 deleted_ids = insert_ids[:2]
481 apdb.deleteInsertIds(deleted_ids)
482
483 # All queries on deleted ids should return empty set.
484 _check_history(deleted_ids, 0)
485
486 insert_ids = apdb.getInsertIds()
487 assert insert_ids is not None
488 self.assertEqual(len(insert_ids), 6)
489
490 _check_history(insert_ids)
491

◆ test_makeSchema()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_makeSchema ( self)
Test for making APDB schema.

Definition at line 189 of file _apdb.py.

189 def test_makeSchema(self) -> None:
190 """Test for making APDB schema."""
191 config = self.make_config()
192 Apdb.makeSchema(config)
193 apdb = make_apdb(config)
194
195 self.assertIsNotNone(apdb.tableDef(ApdbTables.DiaObject))
196 self.assertIsNotNone(apdb.tableDef(ApdbTables.DiaObjectLast))
197 self.assertIsNotNone(apdb.tableDef(ApdbTables.DiaSource))
198 self.assertIsNotNone(apdb.tableDef(ApdbTables.DiaForcedSource))
199 self.assertIsNotNone(apdb.tableDef(ApdbTables.metadata))
200

◆ test_metadata()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_metadata ( self)
Simple test for writing/reading metadata table

Definition at line 626 of file _apdb.py.

626 def test_metadata(self) -> None:
627 """Simple test for writing/reading metadata table"""
628 config = self.make_config()
629 Apdb.makeSchema(config)
630 apdb = make_apdb(config)
631 metadata = apdb.metadata
632
633 # APDB should write two metadata items with version numbers and a
634 # frozen JSON config.
635 self.assertFalse(metadata.empty())
636 self.assertEqual(len(list(metadata.items())), 3)
637
638 metadata.set("meta", "data")
639 metadata.set("data", "meta")
640
641 self.assertFalse(metadata.empty())
642 self.assertTrue(set(metadata.items()) >= {("meta", "data"), ("data", "meta")})
643
644 with self.assertRaisesRegex(KeyError, "Metadata key 'meta' already exists"):
645 metadata.set("meta", "data1")
646
647 metadata.set("meta", "data2", force=True)
648 self.assertTrue(set(metadata.items()) >= {("meta", "data2"), ("data", "meta")})
649
650 self.assertTrue(metadata.delete("meta"))
651 self.assertIsNone(metadata.get("meta"))
652 self.assertFalse(metadata.delete("meta"))
653
654 self.assertEqual(metadata.get("data"), "meta")
655 self.assertEqual(metadata.get("meta", "meta"), "meta")
656
daf::base::PropertySet * set
Definition fits.cc:931

◆ test_midpointMjdTai_fsrc()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_midpointMjdTai_fsrc ( self)
Test for time filtering of DiaForcedSources.

Definition at line 589 of file _apdb.py.

589 def test_midpointMjdTai_fsrc(self) -> None:
590 """Test for time filtering of DiaForcedSources."""
591 config = self.make_config()
592 Apdb.makeSchema(config)
593 apdb = make_apdb(config)
594
595 region = _make_region()
596 src_time1 = astropy.time.Time("2021-01-01T00:00:00", format="isot", scale="tai")
597 src_time2 = astropy.time.Time("2021-01-01T00:00:02", format="isot", scale="tai")
598 visit_time0 = astropy.time.Time("2021-12-26T23:59:59", format="isot", scale="tai")
599 visit_time1 = astropy.time.Time("2021-12-27T00:00:01", format="isot", scale="tai")
600 visit_time2 = astropy.time.Time("2021-12-27T00:00:03", format="isot", scale="tai")
601
602 objects = makeObjectCatalog(region, 100, visit_time0)
603 oids = list(objects["diaObjectId"])
604 sources = makeForcedSourceCatalog(objects, src_time1, 1)
605 apdb.store(src_time1, objects, forced_sources=sources)
606
607 sources = makeForcedSourceCatalog(objects, src_time2, 2)
608 apdb.store(src_time2, objects, forced_sources=sources)
609
610 # reading at time of last save should read all
611 res = apdb.getDiaForcedSources(region, oids, src_time2)
612 self.assert_catalog(res, 200, ApdbTables.DiaForcedSource)
613
614 # one second before 12 months
615 res = apdb.getDiaForcedSources(region, oids, visit_time0)
616 self.assert_catalog(res, 200, ApdbTables.DiaForcedSource)
617
618 # reading at later time of last save should only read a subset
619 res = apdb.getDiaForcedSources(region, oids, visit_time1)
620 self.assert_catalog(res, 100, ApdbTables.DiaForcedSource)
621
622 # reading at later time of last save should only read a subset
623 res = apdb.getDiaForcedSources(region, oids, visit_time2)
624 self.assert_catalog(res, 0, ApdbTables.DiaForcedSource)
625

◆ test_midpointMjdTai_src()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_midpointMjdTai_src ( self)
Test for time filtering of DiaSources.

Definition at line 551 of file _apdb.py.

551 def test_midpointMjdTai_src(self) -> None:
552 """Test for time filtering of DiaSources."""
553 config = self.make_config()
554 Apdb.makeSchema(config)
555 apdb = make_apdb(config)
556
557 region = _make_region()
558 # 2021-01-01 plus 360 days is 2021-12-27
559 src_time1 = astropy.time.Time("2021-01-01T00:00:00", format="isot", scale="tai")
560 src_time2 = astropy.time.Time("2021-01-01T00:00:02", format="isot", scale="tai")
561 visit_time0 = astropy.time.Time("2021-12-26T23:59:59", format="isot", scale="tai")
562 visit_time1 = astropy.time.Time("2021-12-27T00:00:01", format="isot", scale="tai")
563 visit_time2 = astropy.time.Time("2021-12-27T00:00:03", format="isot", scale="tai")
564
565 objects = makeObjectCatalog(region, 100, visit_time0)
566 oids = list(objects["diaObjectId"])
567 sources = makeSourceCatalog(objects, src_time1, 0)
568 apdb.store(src_time1, objects, sources)
569
570 sources = makeSourceCatalog(objects, src_time2, 100)
571 apdb.store(src_time2, objects, sources)
572
573 # reading at time of last save should read all
574 res = apdb.getDiaSources(region, oids, src_time2)
575 self.assert_catalog(res, 200, ApdbTables.DiaSource)
576
577 # one second before 12 months
578 res = apdb.getDiaSources(region, oids, visit_time0)
579 self.assert_catalog(res, 200, ApdbTables.DiaSource)
580
581 # reading at later time of last save should only read a subset
582 res = apdb.getDiaSources(region, oids, visit_time1)
583 self.assert_catalog(res, 100, ApdbTables.DiaSource)
584
585 # reading at later time of last save should only read a subset
586 res = apdb.getDiaSources(region, oids, visit_time2)
587 self.assert_catalog(res, 0, ApdbTables.DiaSource)
588

◆ test_nometadata()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_nometadata ( self)
Test case for when metadata table is missing

Definition at line 657 of file _apdb.py.

657 def test_nometadata(self) -> None:
658 """Test case for when metadata table is missing"""
659 config = self.make_config()
660 # We expect that schema includes metadata table, drop it.
661 with update_schema_yaml(config.schema_file, drop_metadata=True) as schema_file:
662 config_nometa = self.make_config(schema_file=schema_file)
663 Apdb.makeSchema(config_nometa)
664 apdb = make_apdb(config_nometa)
665 metadata = apdb.metadata
666
667 self.assertTrue(metadata.empty())
668 self.assertEqual(list(metadata.items()), [])
669 with self.assertRaisesRegex(RuntimeError, "Metadata table does not exist"):
670 metadata.set("meta", "data")
671
672 self.assertTrue(metadata.empty())
673 self.assertIsNone(metadata.get("meta"))
674
675 # Also check what happens when configured schema has metadata, but
676 # database is missing it. Database was initialized inside above context
677 # without metadata table, here we use schema config which includes
678 # metadata table.
679 apdb = make_apdb(config)
680 metadata = apdb.metadata
681 self.assertTrue(metadata.empty())
682

◆ test_reassignObjects()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_reassignObjects ( self)
Reassign DiaObjects.

Definition at line 517 of file _apdb.py.

517 def test_reassignObjects(self) -> None:
518 """Reassign DiaObjects."""
519 # don't care about sources.
520 config = self.make_config()
521 Apdb.makeSchema(config)
522 apdb = make_apdb(config)
523
524 region = _make_region()
525 visit_time = self.visit_time
526 objects = makeObjectCatalog(region, 100, visit_time)
527 oids = list(objects["diaObjectId"])
528 sources = makeSourceCatalog(objects, visit_time)
529 apdb.store(visit_time, objects, sources)
530
531 catalog = makeSSObjectCatalog(100)
532 apdb.storeSSObjects(catalog)
533
534 # read it back and filter by ID
535 res = apdb.getDiaSources(region, oids, visit_time)
536 self.assert_catalog(res, len(sources), ApdbTables.DiaSource)
537
538 apdb.reassignDiaSources({1: 1, 2: 2, 5: 5})
539 res = apdb.getDiaSources(region, oids, visit_time)
540 self.assert_catalog(res, len(sources) - 3, ApdbTables.DiaSource)
541
542 with self.assertRaisesRegex(ValueError, r"do not exist.*\D1000"):
543 apdb.reassignDiaSources(
544 {
545 1000: 1,
546 7: 3,
547 }
548 )
549 self.assert_catalog(res, len(sources) - 3, ApdbTables.DiaSource)
550

◆ test_schemaVersionFromYaml()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_schemaVersionFromYaml ( self)
Check version number handling for reading schema from YAML.

Definition at line 683 of file _apdb.py.

683 def test_schemaVersionFromYaml(self) -> None:
684 """Check version number handling for reading schema from YAML."""
685 config = self.make_config()
686 default_schema = config.schema_file
687 apdb = make_apdb(config)
688 self.assertEqual(apdb.apdbSchemaVersion(), VersionTuple(0, 1, 1))
689
690 with update_schema_yaml(default_schema, version="") as schema_file:
691 config = self.make_config(schema_file=schema_file)
692 apdb = make_apdb(config)
693 self.assertEqual(apdb.apdbSchemaVersion(), VersionTuple(0, 1, 0))
694
695 with update_schema_yaml(default_schema, version="99.0.0") as schema_file:
696 config = self.make_config(schema_file=schema_file)
697 apdb = make_apdb(config)
698 self.assertEqual(apdb.apdbSchemaVersion(), VersionTuple(99, 0, 0))
699

◆ test_storeForcedSources()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_storeForcedSources ( self)
Store and retrieve DiaForcedSources.

Definition at line 386 of file _apdb.py.

386 def test_storeForcedSources(self) -> None:
387 """Store and retrieve DiaForcedSources."""
388 config = self.make_config()
389 Apdb.makeSchema(config)
390 apdb = make_apdb(config)
391
392 region = _make_region()
393 visit_time = self.visit_time
394
395 # have to store Objects first
396 objects = makeObjectCatalog(region, 100, visit_time)
397 oids = list(objects["diaObjectId"])
398 catalog = makeForcedSourceCatalog(objects, visit_time)
399
400 apdb.store(visit_time, objects, forced_sources=catalog)
401
402 # read it back and check sizes
403 res = apdb.getDiaForcedSources(region, oids, visit_time)
404 self.assert_catalog(res, len(catalog), ApdbTables.DiaForcedSource)
405
406 # read it back to get schema
407 res = apdb.getDiaForcedSources(region, [], visit_time)
408 self.assert_catalog(res, 0, ApdbTables.DiaForcedSource)
409
410 # TODO: test apdb.contains with generic implementation from DM-41671
411
412 # alternative method not part of the Apdb API
413 if isinstance(apdb, ApdbSql):
414 res = apdb.containsCcdVisit(1)
415 self.assertTrue(res)
416 res = apdb.containsCcdVisit(42)
417 self.assertFalse(res)
418

◆ test_storeObjects()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_storeObjects ( self)
Store and retrieve DiaObjects.

Definition at line 305 of file _apdb.py.

305 def test_storeObjects(self) -> None:
306 """Store and retrieve DiaObjects."""
307 # don't care about sources.
308 config = self.make_config()
309 Apdb.makeSchema(config)
310 apdb = make_apdb(config)
311
312 region = _make_region()
313 visit_time = self.visit_time
314
315 # make catalog with Objects
316 catalog = makeObjectCatalog(region, 100, visit_time)
317
318 # store catalog
319 apdb.store(visit_time, catalog)
320
321 # read it back and check sizes
322 res = apdb.getDiaObjects(region)
323 self.assert_catalog(res, len(catalog), self.getDiaObjects_table())
324
325 # TODO: test apdb.contains with generic implementation from DM-41671
326

◆ test_storeObjects_empty()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_storeObjects_empty ( self)
Test calling storeObject when there are no objects: see DM-43270.

Definition at line 327 of file _apdb.py.

327 def test_storeObjects_empty(self) -> None:
328 """Test calling storeObject when there are no objects: see DM-43270."""
329 config = self.make_config()
330 Apdb.makeSchema(config)
331 apdb = make_apdb(config)
332 region = _make_region()
333 visit_time = self.visit_time
334 # make catalog with no Objects
335 catalog = makeObjectCatalog(region, 0, visit_time)
336
337 with self.assertLogs("lsst.dax.apdb.apdbSql", level="DEBUG") as cm:
338 apdb.store(visit_time, catalog)
339 self.assertIn("No objects", "\n".join(cm.output))
340

◆ test_storeSources()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_storeSources ( self)
Store and retrieve DiaSources.

Definition at line 341 of file _apdb.py.

341 def test_storeSources(self) -> None:
342 """Store and retrieve DiaSources."""
343 config = self.make_config()
344 Apdb.makeSchema(config)
345 apdb = make_apdb(config)
346
347 region = _make_region()
348 visit_time = self.visit_time
349
350 # have to store Objects first
351 objects = makeObjectCatalog(region, 100, visit_time)
352 oids = list(objects["diaObjectId"])
353 sources = makeSourceCatalog(objects, visit_time)
354
355 # save the objects and sources
356 apdb.store(visit_time, objects, sources)
357
358 # read it back, no ID filtering
359 res = apdb.getDiaSources(region, None, visit_time)
360 self.assert_catalog(res, len(sources), ApdbTables.DiaSource)
361
362 # read it back and filter by ID
363 res = apdb.getDiaSources(region, oids, visit_time)
364 self.assert_catalog(res, len(sources), ApdbTables.DiaSource)
365
366 # read it back to get schema
367 res = apdb.getDiaSources(region, [], visit_time)
368 self.assert_catalog(res, 0, ApdbTables.DiaSource)
369
370 # test if a visit is present
371 # data_factory's ccdVisitId generation corresponds to (0, 0)
372 if self.allow_visit_query:
373 res = apdb.containsVisitDetector(visit=0, detector=0)
374 self.assertTrue(res)
375 else:
376 with self.assertRaises(NotImplementedError):
377 apdb.containsVisitDetector(visit=0, detector=0)
378
379 # alternative method not part of the Apdb API
380 if isinstance(apdb, ApdbSql):
381 res = apdb.containsCcdVisit(1)
382 self.assertTrue(res)
383 res = apdb.containsCcdVisit(42)
384 self.assertFalse(res)
385

◆ test_storeSSObjects()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_storeSSObjects ( self)
Store and retrieve SSObjects.

Definition at line 492 of file _apdb.py.

492 def test_storeSSObjects(self) -> None:
493 """Store and retrieve SSObjects."""
494 # don't care about sources.
495 config = self.make_config()
496 Apdb.makeSchema(config)
497 apdb = make_apdb(config)
498
499 # make catalog with SSObjects
500 catalog = makeSSObjectCatalog(100, flags=1)
501
502 # store catalog
503 apdb.storeSSObjects(catalog)
504
505 # read it back and check sizes
506 res = apdb.getSSObjects()
507 self.assert_catalog(res, len(catalog), ApdbTables.SSObject)
508
509 # check that override works, make catalog with SSObjects, ID = 51-150
510 catalog = makeSSObjectCatalog(100, 51, flags=2)
511 apdb.storeSSObjects(catalog)
512 res = apdb.getSSObjects()
513 self.assert_catalog(res, 150, ApdbTables.SSObject)
514 self.assertEqual(len(res[res["flags"] == 1]), 50)
515 self.assertEqual(len(res[res["flags"] == 2]), 100)
516

Member Data Documentation

◆ allow_visit_query

bool lsst.dax.apdb.tests._apdb.ApdbTest.allow_visit_query = True
static

Definition at line 131 of file _apdb.py.

◆ fsrc_requires_id_list

bool lsst.dax.apdb.tests._apdb.ApdbTest.fsrc_requires_id_list = False
static

Definition at line 125 of file _apdb.py.

◆ table_column_count

dict lsst.dax.apdb.tests._apdb.ApdbTest.table_column_count
static
Initial value:
= {
ApdbTables.DiaObject: 8,
ApdbTables.DiaObjectLast: 5,
ApdbTables.DiaSource: 10,
ApdbTables.DiaForcedSource: 4,
ApdbTables.SSObject: 3,
}

Definition at line 135 of file _apdb.py.

◆ time_partition_tables

bool lsst.dax.apdb.tests._apdb.ApdbTest.time_partition_tables = False
static

Definition at line 122 of file _apdb.py.

◆ use_insert_id [1/2]

bool lsst.dax.apdb.tests._apdb.ApdbTest.use_insert_id = False
static

Definition at line 128 of file _apdb.py.

◆ use_insert_id [2/2]

lsst.dax.apdb.tests._apdb.ApdbTest.use_insert_id

Definition at line 710 of file _apdb.py.

◆ visit_time

lsst.dax.apdb.tests._apdb.ApdbTest.visit_time = astropy.time.Time("2021-01-01T00:00:00", format="isot", scale="tai")
static

Definition at line 123 of file _apdb.py.


The documentation for this class was generated from the following file: