First db tests (#114)

* First db tests

* Update database.py

Author: cdimauro

test/NOT-USED-HOST

@@ -0,0 +1,121 @@
+import logging
+
+import pytest
+from sqlalchemy import func, select
+
+from wadas._version import __dbversion__
+from wadas.domain.database import DataBase, SQLiteDataBase
+from wadas.domain.db_model import (
+    ActuationEvent,
+    Actuator,
+    Camera,
+    ClassifiedAnimals,
+    DBMetadata,
+    DetectionEvent,
+    FeederActuator,
+    FTPCamera,
+    RoadSignActuator,
+    USBCamera,
+    camera_actuator_association,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.fixture
+def init():
+    if DataBase.wadas_db_engine is not None:
+        DataBase.destroy_instance()
+        assert DataBase.wadas_db_engine is None
+        assert DataBase.wadas_db is None
+    if DataBase.wadas_db is not None:
+        logger.debug("Found an active db instance before test execution...")
+        DataBase.wadas_db = None
+
+
+@pytest.fixture()
+def db(init):
+    assert DataBase.initialize(DataBase.DBTypes.SQLITE, ":memory:", None, "", "") is True
+    db = DataBase.get_instance()
+    session = DataBase.create_session()
+    assert db.create_database() is True
+    yield db, session
+    DataBase.destroy_instance()
+
+
+def test_database_already_created(init):
+    DataBase.wadas_db = True
+    with pytest.raises(RuntimeError, match="Database instance already created."):
+        SQLiteDataBase("NOT-USED-HOST")
+
+
+def test_database_already_initialized(init):
+    DataBase.wadas_db = True
+    assert DataBase.initialize(DataBase.DBTypes.SQLITE, "NOT-USED-HOST", None, "", "") is False
+
+
+def test_no_host(init):
+    assert DataBase.initialize(DataBase.DBTypes.SQLITE, "", None, "", "") is False
+
+
+def test_unsupported_database(init):
+    with pytest.raises(ValueError, match="Unsupported database type: DUMMY-DB-TYPE"):
+        DataBase.initialize("DUMMY-DB-TYPE", "NOT-USED-HOST", None, "", "")
+
+
+def test_no_engine_and_no_database(init):
+    with pytest.raises(RuntimeError, match="The database and db engine have not been initialized."):
+        SQLiteDataBase.get_engine()
+
+
+def test_no_database(init):
+    assert SQLiteDataBase.get_instance() is None
+
+
+def test_failing_on_destroy_instance(init):
+    DataBase.wadas_db_engine = True
+    SQLiteDataBase.destroy_instance()
+    assert DataBase.wadas_db_engine is None
+
+
+def test_connection_string(db):
+    db, session = db
+    assert db is not None
+    assert DataBase.wadas_db_engine is not None
+    assert db.get_connection_string() == "sqlite:///:memory:"
+
+
+def test_database_not_initialized(init):
+    assert SQLiteDataBase("NOT-USED-HOST").create_database() is False
+
+
+def test_serialize(init):
+    assert SQLiteDataBase("NOT-USED-HOST").serialize() == {
+        "host": "NOT-USED-HOST",
+        "type": DataBase.DBTypes.SQLITE.value,
+        "enabled": True,
+        "version": __dbversion__,
+    }
+    assert SQLiteDataBase("NOT-USED-HOST2", False, "vX.Y.Z").serialize() == {
+        "host": "NOT-USED-HOST2",
+        "type": DataBase.DBTypes.SQLITE.value,
+        "enabled": False,
+        "version": "vX.Y.Z",
+    }
+
+
+def test_empty_database(db):
+    db, session = db
+    assert session.execute(select(func.count()).select_from(ActuationEvent)).scalar() == 0
+    assert session.execute(select(func.count()).select_from(Actuator)).scalar() == 0
+    assert session.execute(select(func.count()).select_from(Camera)).scalar() == 0
+    assert session.execute(select(func.count()).select_from(ClassifiedAnimals)).scalar() == 0
+    assert session.execute(select(func.count()).select_from(DBMetadata)).scalar() == 0
+    assert session.execute(select(func.count()).select_from(DetectionEvent)).scalar() == 0
+    assert session.execute(select(func.count()).select_from(FeederActuator)).scalar() == 0
+    assert session.execute(select(func.count()).select_from(FTPCamera)).scalar() == 0
+    assert session.execute(select(func.count()).select_from(RoadSignActuator)).scalar() == 0
+    assert session.execute(select(func.count()).select_from(USBCamera)).scalar() == 0
+    assert (
+        session.execute(select(func.count()).select_from(camera_actuator_association)).scalar() == 0
+    )

test/test_db_sqlite.py

@@ -325,7 +325,7 @@ def save_configuration_to_file(file_, project_uuid):
             if FastAPIActuatorServer.actuator_server
             else ""
         ),
-        "database": DataBase.wadas_db.serialize() if DataBase.wadas_db else "",
+        "database": db.serialize() if (db := DataBase.get_instance()) else "",
     }
 
     with open(file_, "w") as yaml_file:

wadas/domain/configuration.py

@@ -67,14 +67,14 @@ def __init__(self, description: str, project_uuid):
 class DataBase(ABC):
     """Base Class to handle DB object."""
 
-    wadas_db = None  # Singleton instance of the database
-    wadas_db_engine = None  # Singleton engine associated with the database
-
     class DBTypes(Enum):
         SQLITE = "SQLite"
         MYSQL = "MySQL"
         MARIADB = "MariaDB"
 
+    wadas_db = None  # Singleton instance of the database
+    wadas_db_engine = None  # Singleton engine associated with the database
+
     def __init__(self, host, enabled=True, version=__dbversion__):
         """Constructor is not public, no external code should call this directly"""
 
@@ -135,19 +135,17 @@ def initialize(
     ):
         """Initialize the singleton database instance."""
 
-        if cls.wadas_db is not None:
+        if DataBase.wadas_db is not None:
             logger.error("Database is already initialized.")
             return False
 
-        sb_instance = cls._create_instance(
+        DataBase.wadas_db = cls._create_instance(
             db_type, host, port, username, database_name, enabled=True, version=__dbversion__
         )
-        if sb_instance:
-            cls.wadas_db = sb_instance
-        else:
+        if not DataBase.wadas_db:
             return False
 
-        cls.wadas_db_engine = create_engine(cls.wadas_db.get_connection_string())
+        DataBase.wadas_db_engine = create_engine(DataBase.wadas_db.get_connection_string())
         if log:
             logger.info("%s database initialized.", db_type.value)
         return True
@@ -159,14 +157,14 @@ def get_engine(cls):
 
         :return: SQLAlchemy engine instance.
         """
-        if cls.wadas_db_engine is None:
-            if cls.wadas_db is None:
+        if DataBase.wadas_db_engine is None:
+            if DataBase.wadas_db is None:
                 logger.error("The database and db engine have not been initialized.")
                 raise RuntimeError("The database and db engine have not been initialized.")
             else:
                 logger.debug("Initializing engine...")
-                cls.wadas_db_engine = create_engine(cls.wadas_db.get_connection_string())
-        return cls.wadas_db_engine
+                DataBase.wadas_db_engine = create_engine(DataBase.wadas_db.get_connection_string())
+        return DataBase.wadas_db_engine
 
     @classmethod
     def get_instance(cls):
@@ -175,23 +173,23 @@ def get_instance(cls):
 
         :return: The current database instance.
         """
-        if cls.wadas_db is None:
+        if DataBase.wadas_db is None:
             logger.debug("The database has not been initialized. Call 'initialize' first.")
             return None
-        return cls.wadas_db
+        return DataBase.wadas_db
 
     @classmethod
     def destroy_instance(cls):
         """Destroy the current database instance and release resources."""
 
         logger.debug("Destroying db instance...")
-        if cls.wadas_db_engine:
+        if DataBase.wadas_db_engine:
             try:
-                cls.wadas_db_engine.dispose()
+                DataBase.wadas_db_engine.dispose()
             except Exception:
                 logger.warning("Failed to dispose the database engine.")
-        cls.wadas_db_engine = None
-        cls.wadas_db = None
+        DataBase.wadas_db_engine = None
+        DataBase.wadas_db = None
 
     @classmethod
     def create_session(cls):