feat: test dao get_deployment

tests/conftest.py

@@ -4,7 +4,7 @@
 from collections.abc import Generator
 from contextlib import contextmanager
 from pathlib import Path
-from typing import cast
+from typing import Any, cast
 
 import pytest
 import yaml
@@ -136,3 +136,16 @@ def generate_collection_at_path(
         for filename, vars in file_vars.items():
             with (service_dir / filename).open("w") as fd:
                 yaml.dump(vars, fd)
+
+
+def assert_equal_values_in_model(model1: Any, model2: Any) -> bool:
+    """SQLAlchemy asserts that two identical objects of type DeclarativeBase parent of the BaseModel class,
+    which is used in TDP as pattern for the table models, are identical if they are compared in the same session,
+    but different if compared in two different sessions.
+
+    This function therefore transforms the tables into dictionaries and by parsing the coulumns compares their values.
+    """
+    if isinstance(model1, BaseModel) and isinstance(model2, BaseModel):
+        return model1.to_dict() == model2.to_dict()
+    else:
+        return False

tests/unit/test_dao.py

@@ -10,11 +10,14 @@
 from sqlalchemy.engine import Engine
 
 from tdp.core.models import (
+    DeploymentModel,
+    OperationModel,
     SCHStatusLogModel,
     SCHStatusLogSourceEnum,
 )
+from tdp.core.models.enums import DeploymentStateEnum, OperationStateEnum
 from tdp.dao import Dao
-from tests.conftest import create_session
+from tests.conftest import assert_equal_values_in_model, create_session
 
 logger = logging.getLogger(__name__)
 
@@ -158,3 +161,115 @@ def test_multiple_service_component_status(db_engine: Engine):
 
     with Dao(db_engine) as dao:
         assert set(dao.get_cluster_status()) == last_values
+
+
+@pytest.mark.parametrize("db_engine", [True], indirect=True)
+def test_get_deployment(db_engine):
+    with create_session(db_engine) as session:
+        session.add(
+            DeploymentModel(
+                id=1,
+                state=DeploymentStateEnum.RUNNING,
+            )
+        )
+        session.commit()
+    with Dao(db_engine) as dao:
+        assert assert_equal_values_in_model(
+            dao.get_deployment(1),
+            DeploymentModel(
+                id=1,
+                state=DeploymentStateEnum.RUNNING,
+            ),
+        )
+
+
+@pytest.mark.parametrize("db_engine", [True], indirect=True)
+def test_get_planned_deployment(db_engine):
+    with create_session(db_engine) as session:
+        session.add(
+            DeploymentModel(
+                id=1,
+                state=DeploymentStateEnum.PLANNED,
+            )
+        )
+        session.commit()
+    with Dao(db_engine) as dao:
+        assert assert_equal_values_in_model(
+            dao.get_planned_deployment(),
+            DeploymentModel(
+                id=1,
+                state=DeploymentStateEnum.PLANNED,
+            ),
+        )
+
+
+@pytest.mark.parametrize("db_engine", [True], indirect=True)
+def test_get_last_deployment(db_engine):
+    with create_session(db_engine) as session:
+        session.add(
+            DeploymentModel(
+                id=3,
+                state=DeploymentStateEnum.SUCCESS,
+            )
+        )
+        session.commit()
+    with Dao(db_engine) as dao:
+        assert assert_equal_values_in_model(
+            dao.get_last_deployment(),
+            DeploymentModel(
+                id=3,
+                state=DeploymentStateEnum.SUCCESS,
+            ),
+        )
+
+
+@pytest.mark.parametrize("db_engine", [True], indirect=True)
+def test_get_deployments(db_engine):
+    with create_session(db_engine) as session:
+        session.add(
+            DeploymentModel(
+                id=1,
+                state=DeploymentStateEnum.SUCCESS,
+            )
+        )
+        session.add(
+            DeploymentModel(
+                id=2,
+                state=DeploymentStateEnum.PLANNED,
+            )
+        )
+        session.commit()
+    with Dao(db_engine) as dao:
+        assert assert_equal_values_in_model(
+            list(dao.get_deployments())[0],
+            DeploymentModel(id=1, state=DeploymentStateEnum.SUCCESS),
+        )
+        assert assert_equal_values_in_model(
+            list(dao.get_deployments())[1],
+            DeploymentModel(id=2, state=DeploymentStateEnum.PLANNED),
+        )
+
+
+@pytest.mark.parametrize("db_engine", [True], indirect=True)
+def test_operation(db_engine):
+    with create_session(db_engine) as session:
+        session.add(DeploymentModel(id=1, state=DeploymentStateEnum.SUCCESS))
+        session.add(
+            OperationModel(
+                deployment_id=1,
+                operation_order=1,
+                operation="test_operation",
+                state=OperationStateEnum.SUCCESS,
+            )
+        )
+        session.commit()
+    with Dao(db_engine) as dao:
+        assert assert_equal_values_in_model(
+            dao.get_operation(deployment_id=1, operation_name="test_operation")[0],
+            OperationModel(
+                deployment_id=1,
+                operation_order=1,
+                operation="test_operation",
+                state=OperationStateEnum.SUCCESS,
+            ),
+        )