feat: Add possibilities to archive t4c instances

backend/capellacollab/__main__.py

@@ -39,6 +39,9 @@
 from capellacollab.routes import router
 from capellacollab.sessions import exceptions as sessions_exceptions
 from capellacollab.sessions import idletimeout, operators
+from capellacollab.settings.modelsources.t4c import (
+    exceptions as settings_t4c_exceptions,
+)
 from capellacollab.tools import exceptions as tools_exceptions
 from capellacollab.users import exceptions as users_exceptions
 
@@ -145,6 +148,7 @@ def register_exceptions():
     core_exceptions.register_exceptions(app)
     users_exceptions.register_exceptions(app)
     sessions_exceptions.register_exceptions(app)
+    settings_t4c_exceptions.register_exceptions(app)
 
 
 register_exceptions()

backend/capellacollab/alembic/versions/f7bf9456cfc9_add_archive_flag.py

@@ -0,0 +1,35 @@
+# SPDX-FileCopyrightText: Copyright DB Netz AG and the capella-collab-manager contributors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Add archive flag
+
+Revision ID: f7bf9456cfc9
+Revises: 1a4208c18909
+Create Date: 2023-08-28 08:57:22.931913
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "f7bf9456cfc9"
+down_revision = "1a4208c18909"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "t4c_instances", sa.Column("is_archived", sa.Boolean(), nullable=True)
+    )
+
+    op.get_bind().execute(
+        sa.text("UPDATE t4c_instances SET is_archived=false")
+    )
+
+    op.alter_column(
+        "t4c_instances",
+        "is_archived",
+        existing_type=sa.BOOLEAN(),
+        nullable=False,
+    )

backend/capellacollab/projects/toolmodels/modelsources/t4c/routes.py

@@ -79,7 +79,7 @@
     ),
     db: orm.Session = fastapi.Depends(database.get_db),
 ):
-    instance = settings_t4c_injecatbles.get_existing_instance(
+    instance = settings_t4c_injecatbles.get_existing_unarchived_instance(
         body.t4c_instance_id, db
     )
     repository = (

backend/capellacollab/settings/modelsources/t4c/exceptions.py

@@ -0,0 +1,30 @@
+# SPDX-FileCopyrightText: Copyright DB Netz AG and the capella-collab-manager contributors
+# SPDX-License-Identifier: Apache-2.0
+
+import fastapi
+from fastapi import exception_handlers, status
+
+
+class T4CInstanceIsArchivedError(Exception):
+    def __init__(self, t4c_instance_id: int):
+        self.t4c_instance_id = t4c_instance_id
+
+
+async def t4c_instance_is_archived_exception_handler(
+    request: fastapi.Request, exc: T4CInstanceIsArchivedError
+) -> fastapi.Response:
+    return await exception_handlers.http_exception_handler(
+        request,
+        fastapi.HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail={
+                "reason": f"The T4C instance identified by {exc.t4c_instance_id} is archived, thus prohibiting the execution of the requested operation."
+            },
+        ),
+    )
+
+
+def register_exceptions(app: fastapi.FastAPI):
+    app.add_exception_handler(
+        T4CInstanceIsArchivedError, t4c_instance_is_archived_exception_handler
+    )

backend/capellacollab/settings/modelsources/t4c/injectables.py

@@ -7,7 +7,7 @@
 
 from capellacollab.core import database
 
-from . import crud, models
+from . import crud, exceptions, models
 
 
 def get_existing_instance(
@@ -22,3 +22,13 @@
             "reason": f"The t4c instance with the id {t4c_instance_id} does not exist.",
         },
     )
+
+
+def get_existing_unarchived_instance(
+    t4c_instance_id: int, db: orm.Session = fastapi.Depends(database.get_db)
+):
+    t4c_instance = get_existing_instance(t4c_instance_id, db)
+    if t4c_instance.is_archived:
+        raise exceptions.T4CInstanceIsArchivedError(t4c_instance.id)
+
+    return t4c_instance

backend/capellacollab/settings/modelsources/t4c/models.py

@@ -79,6 +79,8 @@ class DatabaseT4CInstance(database.Base):
         back_populates="instance", cascade="all, delete"
     )
 
+    is_archived: orm.Mapped[bool] = orm.mapped_column(default=False)
+
 
 def port_validator(value: int | None) -> int | None:
     if not value:
@@ -123,6 +125,7 @@ class PatchT4CInstance(pydantic.BaseModel):
     password: str | None = None
     protocol: Protocol | None = None
     version_id: int | None = None
+    is_archived: bool | None = None
 
     _validate_rest_api_url = pydantic.field_validator("rest_api")(
         validate_rest_api_url
@@ -138,9 +141,11 @@ class T4CInstanceComplete(T4CInstanceBase):
 
 
 class CreateT4CInstance(T4CInstanceComplete):
+    is_archived: bool | None = None
     password: str
 
 
 class T4CInstance(T4CInstanceComplete):
     id: int
     version: tools_models.ToolVersionBase
+    is_archived: bool

backend/capellacollab/settings/modelsources/t4c/routes.py

@@ -15,7 +15,7 @@
 )
 from capellacollab.users import models as users_models
 
-from . import crud, injectables, interface, models
+from . import crud, exceptions, injectables, interface, models
 
 router = fastapi.APIRouter(
     dependencies=[
@@ -72,6 +72,9 @@ def edit_t4c_instance(
     ),
     db: orm.Session = fastapi.Depends(database.get_db),
 ) -> models.DatabaseT4CInstance:
+    if instance.is_archived and (body.is_archived is None or body.is_archived):
+        raise exceptions.T4CInstanceIsArchivedError(instance.id)
+
     return crud.update_t4c_instance(db, instance, body)
 
 

backend/tests/settings/conftest.py

@@ -25,8 +25,8 @@ def fixture_admin_user(
     return users_crud.create_user(db, executor_name, users_models.Role.ADMIN)
 
 
-@pytest.fixture(name="t4c_server")
-def fixture_t4c_server(
+@pytest.fixture(name="t4c_instance")
+def fixture_t4c_instance(
     db: orm.Session,
     test_tool_version: tools_models.DatabaseVersion,
 ) -> t4c_models.DatabaseT4CInstance:

backend/tests/settings/test_t4c_instances.py

@@ -7,6 +7,12 @@
 from sqlalchemy import orm
 
 from capellacollab.settings.modelsources.t4c import crud as t4c_crud
+from capellacollab.settings.modelsources.t4c import (
+    exceptions as settings_t4c_exceptions,
+)
+from capellacollab.settings.modelsources.t4c import (
+    injectables as settings_t4c_injectables,
+)
 from capellacollab.settings.modelsources.t4c import models as t4c_models
 from capellacollab.tools import models as tools_models
 from capellacollab.users import crud as users_crud
@@ -45,7 +51,7 @@ def test_create_t4c_instance(
     assert t4c_instance.name == "Test integration"
 
 
-@pytest.mark.usefixtures("t4c_server")
+@pytest.mark.usefixtures("t4c_instance")
 def test_get_t4c_instances(
     client: testclient.TestClient, db: orm.Session, executor_name: str
 ):
@@ -67,12 +73,12 @@ def test_get_t4c_instance(
     client: testclient.TestClient,
     db: orm.Session,
     executor_name: str,
-    t4c_server: t4c_models.DatabaseT4CInstance,
+    t4c_instance: t4c_models.DatabaseT4CInstance,
 ):
     users_crud.create_user(db, executor_name, users_models.Role.ADMIN)
 
     response = client.get(
-        f"/api/v1/settings/modelsources/t4c/{t4c_server.id}",
+        f"/api/v1/settings/modelsources/t4c/{t4c_instance.id}",
     )
 
     assert response.json()["name"] == "test server"
@@ -85,58 +91,133 @@ def test_patch_t4c_instance(
     client: testclient.TestClient,
     db: orm.Session,
     executor_name: str,
-    t4c_server: t4c_models.DatabaseT4CInstance,
+    t4c_instance: t4c_models.DatabaseT4CInstance,
 ):
     users_crud.create_user(db, executor_name, users_models.Role.ADMIN)
 
     response = client.patch(
-        f"/api/v1/settings/modelsources/t4c/{t4c_server.id}",
+        f"/api/v1/settings/modelsources/t4c/{t4c_instance.id}",
         json={
             "name": "Patched test integration",
         },
     )
 
-    t4c_instance = t4c_crud.get_t4c_instance_by_id(db, response.json()["id"])
-    assert t4c_instance
+    updated_t4c_instance = t4c_crud.get_t4c_instance_by_id(
+        db, response.json()["id"]
+    )
+    assert updated_t4c_instance
+
+    assert response.status_code == 200
 
     assert response.json()["name"] == "Patched test integration"
-    assert t4c_instance.name == "Patched test integration"
+    assert updated_t4c_instance.name == "Patched test integration"
 
     assert response.json()["host"] == "localhost"
-    assert t4c_instance.host == "localhost"
+    assert updated_t4c_instance.host == "localhost"
+
+
+def test_patch_archived_t4c_instance_error(
+    client: testclient.TestClient,
+    db: orm.Session,
+    executor_name: str,
+    t4c_instance: t4c_models.DatabaseT4CInstance,
+):
+    users_crud.create_user(db, executor_name, users_models.Role.ADMIN)
+
+    t4c_crud.update_t4c_instance(
+        db, t4c_instance, t4c_models.PatchT4CInstance(is_archived=True)
+    )
+
+    response = client.patch(
+        f"/api/v1/settings/modelsources/t4c/{t4c_instance.id}",
+        json={
+            "name": "Patched test integration",
+        },
+    )
+
+    assert response.status_code == 400
+
+
+def test_unarchive_t4c_instance(
+    client: testclient.TestClient,
+    db: orm.Session,
+    executor_name: str,
+    t4c_instance: t4c_models.DatabaseT4CInstance,
+):
+    users_crud.create_user(db, executor_name, users_models.Role.ADMIN)
+
+    t4c_crud.update_t4c_instance(
+        db, t4c_instance, t4c_models.PatchT4CInstance(is_archived=True)
+    )
+
+    assert t4c_instance.is_archived
+
+    response = client.patch(
+        f"/api/v1/settings/modelsources/t4c/{t4c_instance.id}",
+        json={
+            "is_archived": False,
+        },
+    )
+
+    assert response.status_code == 200
+
+    updated_t4c_instance = t4c_crud.get_t4c_instance_by_id(
+        db, response.json()["id"]
+    )
+    assert updated_t4c_instance
+
+    assert not response.json()["is_archived"]
+    assert not updated_t4c_instance.is_archived
+
+
+def test_injectables_raise_when_archived_instance(
+    db: orm.Session,
+    executor_name: str,
+    t4c_instance: t4c_models.DatabaseT4CInstance,
+):
+    users_crud.create_user(db, executor_name, users_models.Role.ADMIN)
+
+    t4c_crud.update_t4c_instance(
+        db, t4c_instance, t4c_models.PatchT4CInstance(is_archived=True)
+    )
+
+    with pytest.raises(settings_t4c_exceptions.T4CInstanceIsArchivedError):
+        settings_t4c_injectables.get_existing_unarchived_instance(
+            t4c_instance.id, db
+        )
 
 
 def test_update_t4c_instance_password_empty_string(
     client: testclient.TestClient,
     db: orm.Session,
     executor_name: str,
-    t4c_server: t4c_models.DatabaseT4CInstance,
+    t4c_instance: t4c_models.DatabaseT4CInstance,
 ):
     users_crud.create_user(db, executor_name, users_models.Role.ADMIN)
 
-    expected_password = t4c_server.password
+    expected_password = t4c_instance.password
 
     response = client.patch(
-        f"/api/v1/settings/modelsources/t4c/{t4c_server.id}",
+        f"/api/v1/settings/modelsources/t4c/{t4c_instance.id}",
         json={
             "password": "",
         },
     )
 
-    updated_t4c_server = t4c_crud.get_t4c_instance_by_id(
+    updated_t4c_instance = t4c_crud.get_t4c_instance_by_id(
         db, response.json()["id"]
     )
 
-    assert updated_t4c_server
-    assert updated_t4c_server.password == expected_password
+    assert updated_t4c_instance
+    assert updated_t4c_instance.password == expected_password
 
 
 @responses.activate
 def test_get_t4c_license_usage(
     client: testclient.TestClient,
     db: orm.Session,
     executor_name: str,
-    t4c_server: t4c_models.DatabaseT4CInstance,
+    t4c_instance: t4c_models.DatabaseT4CInstance,
 ):
     users_crud.create_user(db, executor_name, users_models.Role.ADMIN)
     responses.get(
@@ -146,7 +227,7 @@ def test_get_t4c_license_usage(
     )
 
     response = client.get(
-        f"/api/v1/settings/modelsources/t4c/{t4c_server.id}/licenses",
+        f"/api/v1/settings/modelsources/t4c/{t4c_instance.id}/licenses",
     )
 
     assert response.status_code == 200
@@ -159,7 +240,7 @@ def test_get_t4c_license_usage_no_status(
     client: testclient.TestClient,
     db: orm.Session,
     executor_name: str,
-    t4c_server: t4c_models.DatabaseT4CInstance,
+    t4c_instance: t4c_models.DatabaseT4CInstance,
 ):
     users_crud.create_user(db, executor_name, users_models.Role.ADMIN)
     responses.get(
@@ -169,7 +250,7 @@ def test_get_t4c_license_usage_no_status(
     )
 
     response = client.get(
-        f"/api/v1/settings/modelsources/t4c/{t4c_server.id}/licenses",
+        f"/api/v1/settings/modelsources/t4c/{t4c_instance.id}/licenses",
     )
 
     assert response.status_code == 404