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

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

Public Member Functions
ApdbConfig	make_instance (self, **Any kwargs)
None	test_schema_add_replica (self)
None	test_schemaVersionCheck (self)

Static Public Attributes
	visit_time = astropy.time.Time("2021-01-01T00:00:00", format="isot", scale="tai")

Detailed Description

Base class for unit tests that verify how schema changes work.

Definition at line 717 of file _apdb.py.

Member Function Documentation

make_instance()

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

Make config class instance used in all tests.

This method should return configuration that point to the identical
database instance on each call (i.e. ``db_url`` must be the same,
which also means for sqlite it has to use on-disk storage).

Definition at line 723 of file _apdb.py.

    def make_instance(self, **kwargs: Any) -> ApdbConfig:
        """Make config class instance used in all tests.
 
        This method should return configuration that point to the identical
        database instance on each call (i.e. ``db_url`` must be the same,
        which also means for sqlite it has to use on-disk storage).
        """
        raise NotImplementedError()
 

test_schema_add_replica()

None lsst.dax.apdb.tests._apdb.ApdbSchemaUpdateTest.test_schema_add_replica

(

self

)

Check that new code can work with old schema without replica
tables.

Definition at line 732 of file _apdb.py.

    def test_schema_add_replica(self) -> None:
        """Check that new code can work with old schema without replica
        tables.
        """
        # Make schema without replica tables.
        config = self.make_instance(use_insert_id=False)
        apdb = Apdb.from_config(config)
        apdb_replica = ApdbReplica.from_config(config)
 
        # Make APDB instance configured for replication.
        config.use_insert_id = True
        apdb = Apdb.from_config(config)
 
        # Try to insert something, should work OK.
        region = _make_region()
        visit_time = self.visit_time
 
        # have to store Objects first
        objects = makeObjectCatalog(region, 100, visit_time)
        sources = makeSourceCatalog(objects, visit_time)
        fsources = makeForcedSourceCatalog(objects, visit_time)
        apdb.store(visit_time, objects, sources, fsources)
 
        # There should be no replica chunks.
        replica_chunks = apdb_replica.getReplicaChunks()
        self.assertIsNone(replica_chunks)
 

test_schemaVersionCheck()

None lsst.dax.apdb.tests._apdb.ApdbSchemaUpdateTest.test_schemaVersionCheck

(

self

)

Check version number compatibility.

Definition at line 759 of file _apdb.py.

    def test_schemaVersionCheck(self) -> None:
        """Check version number compatibility."""
        config = self.make_instance()
        apdb = Apdb.from_config(config)
 
        self.assertEqual(apdb.apdbSchemaVersion(), VersionTuple(0, 1, 1))
 
        # Claim that schema version is now 99.0.0, must raise an exception.
        with update_schema_yaml(config.schema_file, version="99.0.0") as schema_file:
            config.schema_file = schema_file
            with self.assertRaises(IncompatibleVersionError):
                apdb = Apdb.from_config(config)

Member Data Documentation

visit_time

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

static

Definition at line 720 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+7d3e92f0ec/python/lsst/dax/apdb/tests/_apdb.py