splits up testing

src/soakdb3_api/databases/database_definition.py

@@ -1,5 +1,7 @@
 import logging
 
+from soakdb3_api.databases.constants import Tablenames
+
 # Base class for all aiosqlite database objects.
 from soakdb3_api.databases.table_definitions import BodyTable, HeadTable, VisitTable
 
@@ -35,27 +37,8 @@ async def add_table_definitions(self, database):
     # ----------------------------------------------------------------------------------------
     async def apply_revision(self, database, revision):
 
-        # Let the base class add any common updates.
-        # Usually only does anything if upgrading to revision 1
-        # which means we are starting with a legacy database with no revision information.
-        await NormsqlAiosqlite.apply_revision(self, database, revision)
-
         # Updating to revision 1 presumably means
         # this is a legacy database with no revision table in it.
+        # TODO: Test revision 1 adding Visit table.
         if revision == 1:
             await database.create_table(Tablenames.VISIT)
-
-        # if revision == 2:
-        #     await self.execute(
-        #         f"ALTER TABLE {Tablenames.ROCKMAKER_IMAGES} ADD COLUMN {ImageFieldnames.NEWFIELD} TEXT",
-        #         why=f"revision 2: add {Tablenames.ROCKMAKER_IMAGES} {ImageFieldnames.NEWFIELD} column",
-        #     )
-        #     await self.execute(
-        #         "CREATE INDEX %s_%s ON %s(%s)"
-        #         % (
-        #             Tablenames.ROCKMAKER_IMAGES,
-        #             ImageFieldnames.NEWFIELD,
-        #             Tablenames.ROCKMAKER_IMAGES,
-        #             ImageFieldnames.NEWFIELD,
-        #         )
-        #     )

tests/base_tester.py

@@ -33,3 +33,33 @@ def main(self, constants, output_directory):
 
         if failure_message is not None:
             pytest.fail(failure_message)
+
+
+# ----------------------------------------------------------------------------------------
+class BaseTester2:
+    """
+    Provide asyncio loop and error checking over *Tester classes.
+    """
+
+    def main(self, constants, specification, output_directory):
+        """
+        This is the main program which calls the test using asyncio.
+        """
+
+        multiprocessing.current_process().name = "main"
+
+        failure_message = None
+        try:
+            # Run main test in asyncio event loop.
+            asyncio.run(
+                self._main_coroutine(constants, specification, output_directory)
+            )
+
+        except Exception as exception:
+            logger.exception(
+                "unexpected exception in the test method", exc_info=exception
+            )
+            failure_message = str(exception)
+
+        if failure_message is not None:
+            pytest.fail(failure_message)

tests/test_database.py

@@ -1,13 +1,11 @@
-import asyncio
 import logging
-import multiprocessing
 
-import pytest
 from dls_normsql.constants import ClassTypes
 from dls_normsql.databases import Databases
 
 from soakdb3_api.databases.constants import BodyFieldnames, HeadFieldnames, Tablenames
 from soakdb3_api.databases.database_definition import DatabaseDefinition
+from tests.base_tester import BaseTester2
 
 logger = logging.getLogger(__name__)
 
@@ -53,57 +51,7 @@ def test(self, constants, logging_setup, output_directory):
 
 
 # ----------------------------------------------------------------------------------------
-class TestDatabaseSqliteBackupRestore:
-    def test(self, constants, logging_setup, output_directory):
-        """
-        Tests the sqlite implementation of XchemBeDatabase.
-        """
-
-        database_specification = {
-            "type": "dls_normsql.aiosqlite",
-            "filename": "%s/soakdb3.sqlite" % (output_directory),
-        }
-
-        # Test direct SQL access to the database.
-        DatabaseTesterBackupRestore().main(
-            constants,
-            database_specification,
-            output_directory,
-        )
-
-
-# ----------------------------------------------------------------------------------------
-class _BaseTester:
-    """
-    Provide asyncio loop and error checking over *Tester classes.
-    """
-
-    def main(self, constants, specification, output_directory):
-        """
-        This is the main program which calls the test using asyncio.
-        """
-
-        multiprocessing.current_process().name = "main"
-
-        failure_message = None
-        try:
-            # Run main test in asyncio event loop.
-            asyncio.run(
-                self._main_coroutine(constants, specification, output_directory)
-            )
-
-        except Exception as exception:
-            logger.exception(
-                "unexpected exception in the test method", exc_info=exception
-            )
-            failure_message = str(exception)
-
-        if failure_message is not None:
-            pytest.fail(failure_message)
-
-
-# ----------------------------------------------------------------------------------------
-class DatabaseTesterHead(_BaseTester):
+class DatabaseTesterHead(BaseTester2):
     """
     Test direct SQL access to the database.
     """
@@ -138,7 +86,7 @@ async def _main_coroutine(
 
 
 # ----------------------------------------------------------------------------------------
-class DatabaseTesterBody(_BaseTester):
+class DatabaseTesterBody(BaseTester2):
     """
     Test direct SQL access to the database.
     """
@@ -202,66 +150,3 @@ async def _main_coroutine(
         finally:
             # Connect from the database... necessary to allow asyncio loop to exit.
             await database.disconnect()
-
-
-# ----------------------------------------------------------------------------------------
-class DatabaseTesterBackupRestore(_BaseTester):
-    """
-    Test direct SQL backup and restore.
-    """
-
-    async def _main_coroutine(
-        self, constants, database_specification, output_directory
-    ):
-        """ """
-
-        databases = Databases()
-        database = databases.build_object(
-            database_specification,
-            DatabaseDefinition(),
-        )
-
-        # Connect to database.
-        await database.connect(should_drop_database=True)
-
-        try:
-            uuid1 = 1000
-            uuid2 = 2000
-
-            # Write one record.
-            await database.insert(
-                Tablenames.BODY,
-                [{BodyFieldnames.LabVisit: "x", BodyFieldnames.ID: uuid1}],
-            )
-
-            # Backup.
-            await database.backup()
-
-            # Write another record.
-            await database.insert(
-                Tablenames.BODY,
-                [{BodyFieldnames.LabVisit: "y", BodyFieldnames.ID: uuid2}],
-            )
-
-            # Backup again (with two records)
-            await database.backup()
-
-            # Restore one in the past (when it had a single record).
-            await database.restore(1)
-
-            all_sql = (
-                f"SELECT * FROM {Tablenames.BODY} ORDER BY ID ASC /* first query */"
-            )
-            records = await database.query(all_sql)
-            assert len(records) == 1, "first %s count expected 1" % (all_sql)
-
-            # Restore most recent (two records).
-            await database.restore(0)
-
-            all_sql = f"SELECT * FROM {Tablenames.BODY} ORDER BY ID ASC"
-            records = await database.query(all_sql)
-            assert len(records) == 2, "second %s count expected 2" % (all_sql)
-
-        finally:
-            # Connect from the database... necessary to allow asyncio loop to exit.
-            await database.disconnect()

tests/test_database_backup.py

@@ -0,0 +1,92 @@
+import logging
+
+from dls_normsql.databases import Databases
+
+from soakdb3_api.databases.constants import BodyFieldnames, Tablenames
+from soakdb3_api.databases.database_definition import DatabaseDefinition
+from tests.base_tester import BaseTester2
+
+logger = logging.getLogger(__name__)
+
+
+# ----------------------------------------------------------------------------------------
+class TestDatabaseSqliteBackupRestore:
+    def test(self, constants, logging_setup, output_directory):
+        """
+        Tests the sqlite implementation of XchemBeDatabase.
+        """
+
+        database_specification = {
+            "type": "dls_normsql.aiosqlite",
+            "filename": "%s/soakdb3.sqlite" % (output_directory),
+        }
+
+        # Test direct SQL access to the database.
+        DatabaseTesterBackupRestore().main(
+            constants,
+            database_specification,
+            output_directory,
+        )
+
+
+# ----------------------------------------------------------------------------------------
+class DatabaseTesterBackupRestore(BaseTester2):
+    """
+    Test direct SQL backup and restore.
+    """
+
+    async def _main_coroutine(
+        self, constants, database_specification, output_directory
+    ):
+        """ """
+
+        databases = Databases()
+        database = databases.build_object(
+            database_specification,
+            DatabaseDefinition(),
+        )
+
+        # Connect to database.
+        await database.connect(should_drop_database=True)
+
+        try:
+            uuid1 = 1000
+            uuid2 = 2000
+
+            # Write one record.
+            await database.insert(
+                Tablenames.BODY,
+                [{BodyFieldnames.LabVisit: "x", BodyFieldnames.ID: uuid1}],
+            )
+
+            # Backup.
+            await database.backup()
+
+            # Write another record.
+            await database.insert(
+                Tablenames.BODY,
+                [{BodyFieldnames.LabVisit: "y", BodyFieldnames.ID: uuid2}],
+            )
+
+            # Backup again (with two records)
+            await database.backup()
+
+            # Restore one in the past (when it had a single record).
+            await database.restore(1)
+
+            all_sql = (
+                f"SELECT * FROM {Tablenames.BODY} ORDER BY ID ASC /* first query */"
+            )
+            records = await database.query(all_sql)
+            assert len(records) == 1, "first %s count expected 1" % (all_sql)
+
+            # Restore most recent (two records).
+            await database.restore(0)
+
+            all_sql = f"SELECT * FROM {Tablenames.BODY} ORDER BY ID ASC"
+            records = await database.query(all_sql)
+            assert len(records) == 2, "second %s count expected 2" % (all_sql)
+
+        finally:
+            # Connect from the database... necessary to allow asyncio loop to exit.
+            await database.disconnect()

tests/test_database_revision.py

@@ -0,0 +1,85 @@
+import logging
+
+from dls_normsql.constants import Tablenames as DlsNormsqlTablenames
+from dls_normsql.databases import Databases
+
+from soakdb3_api.databases.constants import Tablenames
+from soakdb3_api.databases.database_definition import DatabaseDefinition
+from tests.base_tester import BaseTester2
+
+logger = logging.getLogger(__name__)
+
+
+# ----------------------------------------------------------------------------------------
+class TestDatabaseRevision:
+    def test(self, constants, logging_setup, output_directory):
+        """
+        Tests the sqlite implementation of XchemBeDatabase.
+        """
+
+        database_specification = {
+            "type": "dls_normsql.aiosqlite",
+            "filename": "%s/soakdb3.sqlite" % (output_directory),
+        }
+
+        # Test direct SQL access to the database.
+        DatabaseTesterRevision().main(
+            constants,
+            database_specification,
+            output_directory,
+        )
+
+
+# ----------------------------------------------------------------------------------------
+class DatabaseTesterRevision(BaseTester2):
+    """
+    Test direct SQL access to the database.
+    """
+
+    async def _main_coroutine(
+        self, constants, database_specification, output_directory
+    ):
+        """ """
+        databases = Databases()
+        database = databases.build_object(
+            database_specification,
+            DatabaseDefinition(),
+        )
+
+        try:
+            # Connect to database.
+            await database.connect(should_drop_database=True)
+
+            sql = f"SELECT * FROM {DlsNormsqlTablenames.REVISION}"
+            records = await database.query(sql)
+            assert len(records) == 1
+
+            sql = f"SELECT * FROM {Tablenames.VISIT}"
+            records = await database.query(sql)
+            assert len(records) == 0
+
+            await database.execute(f"DROP TABLE {DlsNormsqlTablenames.REVISION}")
+            await database.execute(f"DROP TABLE {Tablenames.VISIT}")
+
+        finally:
+            # Connect from the database... necessary to allow asyncio loop to exit.
+            await database.disconnect()
+
+        try:
+            # Connect to database, revision should be applied to add back the Visit and Revision tables.
+            await database.connect()
+
+            # Make sure we are up to date with the latest database schema revision.
+            await database.apply_revisions()
+
+            sql = f"SELECT * FROM {DlsNormsqlTablenames.REVISION}"
+            records = await database.query(sql)
+            assert len(records) == 1
+
+            sql = f"SELECT * FROM {Tablenames.VISIT}"
+            records = await database.query(sql)
+            assert len(records) == 0
+
+        finally:
+            # Connect from the database... necessary to allow asyncio loop to exit.
+            await database.disconnect()