lsst.dax.apdb.tests._apdb.ApdbTest Class Reference

Inheritance diagram for lsst.dax.apdb.tests._apdb.ApdbTest:

Public Member Functions
ApdbConfig	make_instance (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_getChunks (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
	schema_path
	enable_replica

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	enable_replica = False
bool	allow_visit_query = True
str	schema_path
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 115 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 157 of file _apdb.py.

    def assert_catalog(self, catalog: Any, rows: int, table: ApdbTables) -> None:
        """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.
        """
        self.assertIsInstance(catalog, pandas.DataFrame)
        self.assertEqual(catalog.shape[0], rows)
        self.assertEqual(catalog.shape[1], self.table_column_count[table])
 

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 173 of file _apdb.py.

    def assert_table_data(self, catalog: Any, rows: int, table: ApdbTables) -> None:
        """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``.
        """
        self.assertIsInstance(catalog, ApdbTableData)
        n_rows = sum(1 for row in catalog.rows())
        self.assertEqual(n_rows, rows)
        # One extra column for replica chunk id
        self.assertEqual(len(catalog.column_names()), self.table_column_count[table] + 1)
 

getDiaObjects_table()

ApdbTables lsst.dax.apdb.tests._apdb.ApdbTest.getDiaObjects_table

(

self

)

Return type of table returned from getDiaObjects method.

Definition at line 153 of file _apdb.py.

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

make_instance()

ApdbConfig lsst.dax.apdb.tests._apdb.ApdbTest.make_instance	(		self,
		**Any	kwargs )

Make database instance and return configuration for it.

Definition at line 148 of file _apdb.py.

    def make_instance(self, **kwargs: Any) -> ApdbConfig:
        """Make database instance and return configuration for it."""
        raise NotImplementedError()
 

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 705 of file _apdb.py.

    def test_config_freeze(self) -> None:
        """Test that some config fields are correctly frozen in database."""
        config = self.make_instance()
 
        # `use_insert_id` is the only parameter that is frozen in all
        # implementations.
        config.use_insert_id = not self.enable_replica
        apdb = Apdb.from_config(config)
        frozen_config = apdb.config  # type: ignore[attr-defined]
        self.assertEqual(frozen_config.use_insert_id, self.enable_replica)
 
 

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 215 of file _apdb.py.

    def test_empty_gets(self) -> None:
        """Test for getting data from empty database.
 
        All get() methods should return empty results, only useful for
        checking that code is not broken.
        """
        # use non-zero months for Forced/Source fetching
        config = self.make_instance()
        apdb = Apdb.from_config(config)
 
        region = _make_region()
        visit_time = self.visit_time
 
        res: pandas.DataFrame | None
 
        # get objects by region
        res = apdb.getDiaObjects(region)
        self.assert_catalog(res, 0, self.getDiaObjects_table())
 
        # get sources by region
        res = apdb.getDiaSources(region, None, visit_time)
        self.assert_catalog(res, 0, ApdbTables.DiaSource)
 
        res = apdb.getDiaSources(region, [], visit_time)
        self.assert_catalog(res, 0, ApdbTables.DiaSource)
 
        # get sources by object ID, non-empty object list
        res = apdb.getDiaSources(region, [1, 2, 3], visit_time)
        self.assert_catalog(res, 0, ApdbTables.DiaSource)
 
        # get forced sources by object ID, empty object list
        res = apdb.getDiaForcedSources(region, [], visit_time)
        self.assert_catalog(res, 0, ApdbTables.DiaForcedSource)
 
        # get sources by object ID, non-empty object list
        res = apdb.getDiaForcedSources(region, [1, 2, 3], visit_time)
        self.assert_catalog(res, 0, ApdbTables.DiaForcedSource)
 
        # test if a visit has objects/sources
        if self.allow_visit_query:
            res = apdb.containsVisitDetector(visit=0, detector=0)
            self.assertFalse(res)
        else:
            with self.assertRaises(NotImplementedError):
                apdb.containsVisitDetector(visit=0, detector=0)
 
        # alternative method not part of the Apdb API
        if isinstance(apdb, ApdbSql):
            res = apdb.containsCcdVisit(1)
            self.assertFalse(res)
 
        # get sources by region
        if self.fsrc_requires_id_list:
            with self.assertRaises(NotImplementedError):
                apdb.getDiaForcedSources(region, None, visit_time)
        else:
            apdb.getDiaForcedSources(region, None, visit_time)
            self.assert_catalog(res, 0, ApdbTables.DiaForcedSource)
 

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 274 of file _apdb.py.

    def test_empty_gets_0months(self) -> None:
        """Test for getting data from empty database.
 
        All get() methods should return empty DataFrame or None.
        """
        # set read_sources_months to 0 so that Forced/Sources are None
        config = self.make_instance(read_sources_months=0, read_forced_sources_months=0)
        apdb = Apdb.from_config(config)
 
        region = _make_region()
        visit_time = self.visit_time
 
        res: pandas.DataFrame | None
 
        # get objects by region
        res = apdb.getDiaObjects(region)
        self.assert_catalog(res, 0, self.getDiaObjects_table())
 
        # get sources by region
        res = apdb.getDiaSources(region, None, visit_time)
        self.assertIs(res, None)
 
        # get sources by object ID, empty object list
        res = apdb.getDiaSources(region, [], visit_time)
        self.assertIs(res, None)
 
        # get forced sources by object ID, empty object list
        res = apdb.getDiaForcedSources(region, [], visit_time)
        self.assertIs(res, None)
 
        # test if a visit has objects/sources
        if self.allow_visit_query:
            res = apdb.containsVisitDetector(visit=0, detector=0)
            self.assertFalse(res)
        else:
            with self.assertRaises(NotImplementedError):
                apdb.containsVisitDetector(visit=0, detector=0)
 
        # alternative method not part of the Apdb API
        if isinstance(apdb, ApdbSql):
            res = apdb.containsCcdVisit(1)
            self.assertFalse(res)
 

test_getChunks()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_getChunks

(

self

)

Store and retrieve replica chunks.

Definition at line 427 of file _apdb.py.

    def test_getChunks(self) -> None:
        """Store and retrieve replica chunks."""
        # don't care about sources.
        config = self.make_instance()
        apdb = Apdb.from_config(config)
        apdb_replica = ApdbReplica.from_config(config)
        visit_time = self.visit_time
 
        region1 = _make_region((1.0, 1.0, -1.0))
        region2 = _make_region((-1.0, -1.0, -1.0))
        nobj = 100
        objects1 = makeObjectCatalog(region1, nobj, visit_time)
        objects2 = makeObjectCatalog(region2, nobj, visit_time, start_id=nobj * 2)
 
        # With the default 10 minutes replica chunk window we should have 4
        # records.
        visits = [
            (astropy.time.Time("2021-01-01T00:01:00", format="isot", scale="tai"), objects1),
            (astropy.time.Time("2021-01-01T00:02:00", format="isot", scale="tai"), objects2),
            (astropy.time.Time("2021-01-01T00:11:00", format="isot", scale="tai"), objects1),
            (astropy.time.Time("2021-01-01T00:12:00", format="isot", scale="tai"), objects2),
            (astropy.time.Time("2021-01-01T00:45:00", format="isot", scale="tai"), objects1),
            (astropy.time.Time("2021-01-01T00:46:00", format="isot", scale="tai"), objects2),
            (astropy.time.Time("2021-03-01T00:01:00", format="isot", scale="tai"), objects1),
            (astropy.time.Time("2021-03-01T00:02:00", format="isot", scale="tai"), objects2),
        ]
 
        start_id = 0
        for visit_time, objects in visits:
            sources = makeSourceCatalog(objects, visit_time, start_id=start_id)
            fsources = makeForcedSourceCatalog(objects, visit_time, ccdVisitId=start_id)
            apdb.store(visit_time, objects, sources, fsources)
            start_id += nobj
 
        replica_chunks = apdb_replica.getReplicaChunks()
        if not self.enable_replica:
            self.assertIsNone(replica_chunks)
 
            with self.assertRaisesRegex(ValueError, "APDB is not configured for replication"):
                apdb_replica.getDiaObjectsChunks([])
 
        else:
            assert replica_chunks is not None
            self.assertEqual(len(replica_chunks), 4)
 
            def _check_chunks(replica_chunks: list[ReplicaChunk], n_records: int | None = None) -> None:
                if n_records is None:
                    n_records = len(replica_chunks) * nobj
                res = apdb_replica.getDiaObjectsChunks(chunk.id for chunk in replica_chunks)
                self.assert_table_data(res, n_records, ApdbTables.DiaObject)
                res = apdb_replica.getDiaSourcesChunks(chunk.id for chunk in replica_chunks)
                self.assert_table_data(res, n_records, ApdbTables.DiaSource)
                res = apdb_replica.getDiaForcedSourcesChunks(chunk.id for chunk in replica_chunks)
                self.assert_table_data(res, n_records, ApdbTables.DiaForcedSource)
 
            # read it back and check sizes
            _check_chunks(replica_chunks, 800)
            _check_chunks(replica_chunks[1:], 600)
            _check_chunks(replica_chunks[1:-1], 400)
            _check_chunks(replica_chunks[2:3], 200)
            _check_chunks([])
 
            # try to remove some of those
            deleted_chunks = replica_chunks[:1]
            apdb_replica.deleteReplicaChunks(chunk.id for chunk in deleted_chunks)
 
            # All queries on deleted ids should return empty set.
            _check_chunks(deleted_chunks, 0)
 
            replica_chunks = apdb_replica.getReplicaChunks()
            assert replica_chunks is not None
            self.assertEqual(len(replica_chunks), 3)
 
            _check_chunks(replica_chunks, 600)
 

test_makeSchema()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_makeSchema

(

self

)

Test for making APDB schema.

Definition at line 193 of file _apdb.py.

    def test_makeSchema(self) -> None:
        """Test for making APDB schema."""
        config = self.make_instance()
        apdb = Apdb.from_config(config)
 
        self.assertIsNotNone(apdb.tableDef(ApdbTables.DiaObject))
        self.assertIsNotNone(apdb.tableDef(ApdbTables.DiaObjectLast))
        self.assertIsNotNone(apdb.tableDef(ApdbTables.DiaSource))
        self.assertIsNotNone(apdb.tableDef(ApdbTables.DiaForcedSource))
        self.assertIsNotNone(apdb.tableDef(ApdbTables.metadata))
 
        # Test from_uri factory method with the same config.
        with tempfile.NamedTemporaryFile() as tmpfile:
            config.save(tmpfile.name)
            apdb = Apdb.from_uri(tmpfile.name)
 
        self.assertIsNotNone(apdb.tableDef(ApdbTables.DiaObject))
        self.assertIsNotNone(apdb.tableDef(ApdbTables.DiaObjectLast))
        self.assertIsNotNone(apdb.tableDef(ApdbTables.DiaSource))
        self.assertIsNotNone(apdb.tableDef(ApdbTables.DiaForcedSource))
        self.assertIsNotNone(apdb.tableDef(ApdbTables.metadata))
 

test_metadata()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_metadata

(

self

)

Simple test for writing/reading metadata table

Definition at line 632 of file _apdb.py.

    def test_metadata(self) -> None:
        """Simple test for writing/reading metadata table"""
        config = self.make_instance()
        apdb = Apdb.from_config(config)
        metadata = apdb.metadata
 
        # APDB should write two or three metadata items with version numbers
        # and a frozen JSON config.
        self.assertFalse(metadata.empty())
        expected_rows = 4 if self.enable_replica else 3
        self.assertEqual(len(list(metadata.items())), expected_rows)
 
        metadata.set("meta", "data")
        metadata.set("data", "meta")
 
        self.assertFalse(metadata.empty())
        self.assertTrue(set(metadata.items()) >= {("meta", "data"), ("data", "meta")})
 
        with self.assertRaisesRegex(KeyError, "Metadata key 'meta' already exists"):
            metadata.set("meta", "data1")
 
        metadata.set("meta", "data2", force=True)
        self.assertTrue(set(metadata.items()) >= {("meta", "data2"), ("data", "meta")})
 
        self.assertTrue(metadata.delete("meta"))
        self.assertIsNone(metadata.get("meta"))
        self.assertFalse(metadata.delete("meta"))
 
        self.assertEqual(metadata.get("data"), "meta")
        self.assertEqual(metadata.get("meta", "meta"), "meta")
 

test_midpointMjdTai_fsrc()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_midpointMjdTai_fsrc

(

self

)

Test for time filtering of DiaForcedSources.

Definition at line 596 of file _apdb.py.

    def test_midpointMjdTai_fsrc(self) -> None:
        """Test for time filtering of DiaForcedSources."""
        config = self.make_instance()
        apdb = Apdb.from_config(config)
 
        region = _make_region()
        src_time1 = astropy.time.Time("2021-01-01T00:00:00", format="isot", scale="tai")
        src_time2 = astropy.time.Time("2021-01-01T00:00:02", format="isot", scale="tai")
        visit_time0 = astropy.time.Time("2021-12-26T23:59:59", format="isot", scale="tai")
        visit_time1 = astropy.time.Time("2021-12-27T00:00:01", format="isot", scale="tai")
        visit_time2 = astropy.time.Time("2021-12-27T00:00:03", format="isot", scale="tai")
 
        objects = makeObjectCatalog(region, 100, visit_time0)
        oids = list(objects["diaObjectId"])
        sources = makeForcedSourceCatalog(objects, src_time1, 1)
        apdb.store(src_time1, objects, forced_sources=sources)
 
        sources = makeForcedSourceCatalog(objects, src_time2, 2)
        apdb.store(src_time2, objects, forced_sources=sources)
 
        # reading at time of last save should read all
        res = apdb.getDiaForcedSources(region, oids, src_time2)
        self.assert_catalog(res, 200, ApdbTables.DiaForcedSource)
 
        # one second before 12 months
        res = apdb.getDiaForcedSources(region, oids, visit_time0)
        self.assert_catalog(res, 200, ApdbTables.DiaForcedSource)
 
        # reading at later time of last save should only read a subset
        res = apdb.getDiaForcedSources(region, oids, visit_time1)
        self.assert_catalog(res, 100, ApdbTables.DiaForcedSource)
 
        # reading at later time of last save should only read a subset
        res = apdb.getDiaForcedSources(region, oids, visit_time2)
        self.assert_catalog(res, 0, ApdbTables.DiaForcedSource)
 

test_midpointMjdTai_src()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_midpointMjdTai_src

(

self

)

Test for time filtering of DiaSources.

Definition at line 559 of file _apdb.py.

    def test_midpointMjdTai_src(self) -> None:
        """Test for time filtering of DiaSources."""
        config = self.make_instance()
        apdb = Apdb.from_config(config)
 
        region = _make_region()
        # 2021-01-01 plus 360 days is 2021-12-27
        src_time1 = astropy.time.Time("2021-01-01T00:00:00", format="isot", scale="tai")
        src_time2 = astropy.time.Time("2021-01-01T00:00:02", format="isot", scale="tai")
        visit_time0 = astropy.time.Time("2021-12-26T23:59:59", format="isot", scale="tai")
        visit_time1 = astropy.time.Time("2021-12-27T00:00:01", format="isot", scale="tai")
        visit_time2 = astropy.time.Time("2021-12-27T00:00:03", format="isot", scale="tai")
 
        objects = makeObjectCatalog(region, 100, visit_time0)
        oids = list(objects["diaObjectId"])
        sources = makeSourceCatalog(objects, src_time1, 0)
        apdb.store(src_time1, objects, sources)
 
        sources = makeSourceCatalog(objects, src_time2, 100)
        apdb.store(src_time2, objects, sources)
 
        # reading at time of last save should read all
        res = apdb.getDiaSources(region, oids, src_time2)
        self.assert_catalog(res, 200, ApdbTables.DiaSource)
 
        # one second before 12 months
        res = apdb.getDiaSources(region, oids, visit_time0)
        self.assert_catalog(res, 200, ApdbTables.DiaSource)
 
        # reading at later time of last save should only read a subset
        res = apdb.getDiaSources(region, oids, visit_time1)
        self.assert_catalog(res, 100, ApdbTables.DiaSource)
 
        # reading at later time of last save should only read a subset
        res = apdb.getDiaSources(region, oids, visit_time2)
        self.assert_catalog(res, 0, ApdbTables.DiaSource)
 

test_nometadata()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_nometadata

(

self

)

Test case for when metadata table is missing

Definition at line 663 of file _apdb.py.

    def test_nometadata(self) -> None:
        """Test case for when metadata table is missing"""
        # We expect that schema includes metadata table, drop it.
        with update_schema_yaml(self.schema_path, drop_metadata=True) as schema_file:
            config = self.make_instance(schema_file=schema_file)
            apdb = Apdb.from_config(config)
            metadata = apdb.metadata
 
            self.assertTrue(metadata.empty())
            self.assertEqual(list(metadata.items()), [])
            with self.assertRaisesRegex(RuntimeError, "Metadata table does not exist"):
                metadata.set("meta", "data")
 
            self.assertTrue(metadata.empty())
            self.assertIsNone(metadata.get("meta"))
 
        # Also check what happens when configured schema has metadata, but
        # database is missing it. Database was initialized inside above context
        # without metadata table, here we use schema config which includes
        # metadata table.
        config.schema_file = self.schema_path
        apdb = Apdb.from_config(config)
        metadata = apdb.metadata
        self.assertTrue(metadata.empty())
 

test_reassignObjects()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_reassignObjects

(

self

)

Reassign DiaObjects.

Definition at line 526 of file _apdb.py.

    def test_reassignObjects(self) -> None:
        """Reassign DiaObjects."""
        # don't care about sources.
        config = self.make_instance()
        apdb = Apdb.from_config(config)
 
        region = _make_region()
        visit_time = self.visit_time
        objects = makeObjectCatalog(region, 100, visit_time)
        oids = list(objects["diaObjectId"])
        sources = makeSourceCatalog(objects, visit_time)
        apdb.store(visit_time, objects, sources)
 
        catalog = makeSSObjectCatalog(100)
        apdb.storeSSObjects(catalog)
 
        # read it back and filter by ID
        res = apdb.getDiaSources(region, oids, visit_time)
        self.assert_catalog(res, len(sources), ApdbTables.DiaSource)
 
        apdb.reassignDiaSources({1: 1, 2: 2, 5: 5})
        res = apdb.getDiaSources(region, oids, visit_time)
        self.assert_catalog(res, len(sources) - 3, ApdbTables.DiaSource)
 
        with self.assertRaisesRegex(ValueError, r"do not exist.*\D1000"):
            apdb.reassignDiaSources(
                {
                    1000: 1,
                    7: 3,
                }
            )
        self.assert_catalog(res, len(sources) - 3, ApdbTables.DiaSource)
 

test_schemaVersionFromYaml()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_schemaVersionFromYaml

(

self

)

Check version number handling for reading schema from YAML.

Definition at line 688 of file _apdb.py.

    def test_schemaVersionFromYaml(self) -> None:
        """Check version number handling for reading schema from YAML."""
        config = self.make_instance()
        default_schema = config.schema_file
        apdb = Apdb.from_config(config)
        self.assertEqual(apdb.apdbSchemaVersion(), VersionTuple(0, 1, 1))
 
        with update_schema_yaml(default_schema, version="") as schema_file:
            config = self.make_instance(schema_file=schema_file)
            apdb = Apdb.from_config(config)
            self.assertEqual(apdb.apdbSchemaVersion(), VersionTuple(0, 1, 0))
 
        with update_schema_yaml(default_schema, version="99.0.0") as schema_file:
            config = self.make_instance(schema_file=schema_file)
            apdb = Apdb.from_config(config)
            self.assertEqual(apdb.apdbSchemaVersion(), VersionTuple(99, 0, 0))
 

test_storeForcedSources()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_storeForcedSources

(

self

)

Store and retrieve DiaForcedSources.

Definition at line 395 of file _apdb.py.

    def test_storeForcedSources(self) -> None:
        """Store and retrieve DiaForcedSources."""
        config = self.make_instance()
        apdb = Apdb.from_config(config)
 
        region = _make_region()
        visit_time = self.visit_time
 
        # have to store Objects first
        objects = makeObjectCatalog(region, 100, visit_time)
        oids = list(objects["diaObjectId"])
        catalog = makeForcedSourceCatalog(objects, visit_time)
 
        apdb.store(visit_time, objects, forced_sources=catalog)
 
        # read it back and check sizes
        res = apdb.getDiaForcedSources(region, oids, visit_time)
        self.assert_catalog(res, len(catalog), ApdbTables.DiaForcedSource)
 
        # read it back to get schema
        res = apdb.getDiaForcedSources(region, [], visit_time)
        self.assert_catalog(res, 0, ApdbTables.DiaForcedSource)
 
        # TODO: test apdb.contains with generic implementation from DM-41671
 
        # alternative method not part of the Apdb API
        if isinstance(apdb, ApdbSql):
            res = apdb.containsCcdVisit(1)
            self.assertTrue(res)
            res = apdb.containsCcdVisit(42)
            self.assertFalse(res)
 

test_storeObjects()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_storeObjects

(

self

)

Store and retrieve DiaObjects.

Definition at line 317 of file _apdb.py.

    def test_storeObjects(self) -> None:
        """Store and retrieve DiaObjects."""
        # don't care about sources.
        config = self.make_instance()
        apdb = Apdb.from_config(config)
 
        region = _make_region()
        visit_time = self.visit_time
 
        # make catalog with Objects
        catalog = makeObjectCatalog(region, 100, visit_time)
 
        # store catalog
        apdb.store(visit_time, catalog)
 
        # read it back and check sizes
        res = apdb.getDiaObjects(region)
        self.assert_catalog(res, len(catalog), self.getDiaObjects_table())
 
        # TODO: test apdb.contains with generic implementation from DM-41671
 

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 338 of file _apdb.py.

    def test_storeObjects_empty(self) -> None:
        """Test calling storeObject when there are no objects: see DM-43270."""
        config = self.make_instance()
        apdb = Apdb.from_config(config)
        region = _make_region()
        visit_time = self.visit_time
        # make catalog with no Objects
        catalog = makeObjectCatalog(region, 0, visit_time)
 
        with self.assertLogs("lsst.dax.apdb", level="DEBUG") as cm:
            apdb.store(visit_time, catalog)
        self.assertIn("No objects", "\n".join(cm.output))
 

test_storeSources()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_storeSources

(

self

)

Store and retrieve DiaSources.

Definition at line 351 of file _apdb.py.

    def test_storeSources(self) -> None:
        """Store and retrieve DiaSources."""
        config = self.make_instance()
        apdb = Apdb.from_config(config)
 
        region = _make_region()
        visit_time = self.visit_time
 
        # have to store Objects first
        objects = makeObjectCatalog(region, 100, visit_time)
        oids = list(objects["diaObjectId"])
        sources = makeSourceCatalog(objects, visit_time)
 
        # save the objects and sources
        apdb.store(visit_time, objects, sources)
 
        # read it back, no ID filtering
        res = apdb.getDiaSources(region, None, visit_time)
        self.assert_catalog(res, len(sources), ApdbTables.DiaSource)
 
        # read it back and filter by ID
        res = apdb.getDiaSources(region, oids, visit_time)
        self.assert_catalog(res, len(sources), ApdbTables.DiaSource)
 
        # read it back to get schema
        res = apdb.getDiaSources(region, [], visit_time)
        self.assert_catalog(res, 0, ApdbTables.DiaSource)
 
        # test if a visit is present
        # data_factory's ccdVisitId generation corresponds to (0, 0)
        if self.allow_visit_query:
            res = apdb.containsVisitDetector(visit=0, detector=0)
            self.assertTrue(res)
        else:
            with self.assertRaises(NotImplementedError):
                apdb.containsVisitDetector(visit=0, detector=0)
 
        # alternative method not part of the Apdb API
        if isinstance(apdb, ApdbSql):
            res = apdb.containsCcdVisit(1)
            self.assertTrue(res)
            res = apdb.containsCcdVisit(42)
            self.assertFalse(res)
 

test_storeSSObjects()

None lsst.dax.apdb.tests._apdb.ApdbTest.test_storeSSObjects

(

self

)

Store and retrieve SSObjects.

Definition at line 502 of file _apdb.py.

    def test_storeSSObjects(self) -> None:
        """Store and retrieve SSObjects."""
        # don't care about sources.
        config = self.make_instance()
        apdb = Apdb.from_config(config)
 
        # make catalog with SSObjects
        catalog = makeSSObjectCatalog(100, flags=1)
 
        # store catalog
        apdb.storeSSObjects(catalog)
 
        # read it back and check sizes
        res = apdb.getSSObjects()
        self.assert_catalog(res, len(catalog), ApdbTables.SSObject)
 
        # check that override works, make catalog with SSObjects, ID = 51-150
        catalog = makeSSObjectCatalog(100, 51, flags=2)
        apdb.storeSSObjects(catalog)
        res = apdb.getSSObjects()
        self.assert_catalog(res, 150, ApdbTables.SSObject)
        self.assertEqual(len(res[res["flags"] == 1]), 50)
        self.assertEqual(len(res[res["flags"] == 2]), 100)
 

Member Data Documentation

allow_visit_query

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

static

Definition at line 132 of file _apdb.py.

enable_replica [1/2]

bool lsst.dax.apdb.tests._apdb.ApdbTest.enable_replica = False

static

Definition at line 129 of file _apdb.py.

enable_replica [2/2]

lsst.dax.apdb.tests._apdb.ApdbTest.enable_replica

Definition at line 714 of file _apdb.py.

fsrc_requires_id_list

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

static

Definition at line 126 of file _apdb.py.

schema_path [1/2]

str lsst.dax.apdb.tests._apdb.ApdbTest.schema_path

static

Definition at line 135 of file _apdb.py.

schema_path [2/2]

lsst.dax.apdb.tests._apdb.ApdbTest.schema_path

Definition at line 666 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 139 of file _apdb.py.

time_partition_tables

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

static

Definition at line 123 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 124 of file _apdb.py.

---

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

• /j/snowflake/release/lsstsw/stack/lsst-scipipe-8.0.0/Linux64/dax_apdb/g295015adf3+81dd352a9d/python/lsst/dax/apdb/tests/_apdb.py