[DOP-13019] - restrict deletion namespaces with hwms

docs/changelog/next_release/25.feature.rst

@@ -0,0 +1 @@
+Restrict deletion of ``Namespace`` if there are any  ``hwms`` related to it.

horizon/backend/api/v1/router/namespaces.py

@@ -1,7 +1,7 @@
 # SPDX-FileCopyrightText: 2023-2024 MTS (Mobile Telesystems)
 # SPDX-License-Identifier: Apache-2.0
 
-from fastapi import APIRouter, Depends, status
+from fastapi import APIRouter, Depends, HTTPException, status
 from typing_extensions import Annotated
 
 from horizon.backend.db.models import User
@@ -100,6 +100,13 @@ async def delete_namespace(
     unit_of_work: Annotated[UnitOfWork, Depends()],
 ) -> None:
     async with unit_of_work:
+        hwm_records = await unit_of_work.hwm.paginate(namespace_id=namespace_id, page=1, page_size=1)
+        if hwm_records.total_count:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Cannot delete namespace because it has related HWM records.",
+            )
+
         namespace = await unit_of_work.namespace.delete(namespace_id=namespace_id, user=user)
         await unit_of_work.namespace_history.create(
             namespace_id=namespace_id,

tests/factories/hwm.py

@@ -40,7 +40,7 @@ async def new_hwm(
     item = hwm_factory(**params)
     yield item
 
-    query = delete(HWM).where(HWM.name == item.name)
+    query = delete(HWM).where(HWM.name == item.name, HWM.namespace_id == item.namespace_id)
 
     # do not use the same session in tests and fixture teardown
     # see https://github.com/MobileTeleSystems/horizon/pull/6

tests/test_backend/test_hwm/test_update_hwm.py

@@ -144,7 +144,7 @@ async def test_update_hwm(
     assert updated_hwm_history.action == "Updated"
 
 
-async def test_update_hwm_already_exist(
+async def test_update_hwm_already_exists(
     test_client: AsyncClient,
     access_token: str,
     hwms: list[HWM],

tests/test_backend/test_namespace/test_delete_namespace.py

@@ -8,7 +8,7 @@
 import pytest
 from sqlalchemy import select
 
-from horizon.backend.db.models import Namespace, NamespaceHistory, User
+from horizon.backend.db.models import HWM, Namespace, NamespaceHistory, User
 
 if TYPE_CHECKING:
     from httpx import AsyncClient
@@ -89,3 +89,59 @@ async def test_delete_namespace(
     assert created_namespace_history.action == "Deleted"
     assert created_namespace_history.changed_by_user_id == user.id
     assert pre_delete_timestamp <= created_namespace_history.changed_at <= post_delete_timestamp
+
+
+async def test_delete_namespace_with_existing_hwm(
+    test_client: AsyncClient,
+    access_token: str,
+    user: User,
+    namespace: Namespace,
+    hwm: HWM,
+    async_session: AsyncSession,
+):
+    response = await test_client.delete(
+        f"v1/namespaces/{namespace.id}",
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+
+    assert response.status_code == 400
+    assert response.json() == {
+        "error": {
+            "code": "bad_request",
+            "message": "Cannot delete namespace because it has related HWM records.",
+            "details": None,
+        },
+    }
+
+    query = select(Namespace).where(Namespace.id == namespace.id)
+    result = await async_session.execute(query)
+    namespace_record = result.scalars().first()
+    assert namespace_record is not None
+
+    query_hwm = select(HWM).where(HWM.namespace_id == namespace.id)
+    result_hwm = await async_session.execute(query_hwm)
+    hwm_records = result_hwm.scalars().all()
+    assert len(hwm_records) > 0
+
+    response = await test_client.delete(
+        f"v1/hwm/{hwm.id}",
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+
+    assert response.status_code == 204
+
+    response = await test_client.delete(
+        f"v1/namespaces/{namespace.id}",
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    assert response.status_code == 204
+
+    query = select(Namespace).where(Namespace.id == namespace.id)
+    result = await async_session.execute(query)
+    namespace_record = result.scalars().first()
+    assert namespace_record is None
+
+    query_hwm = select(HWM).where(HWM.namespace_id == namespace.id)
+    result_hwm = await async_session.execute(query_hwm)
+    hwm_records = result_hwm.scalars().all()
+    assert len(hwm_records) == 0