tests: add deletion tests

Author: giancarloromeo

services/web/server/tests/unit/with_dbs/04/functions/conftest.py

@@ -6,7 +6,7 @@
 from collections.abc import AsyncIterator, Awaitable, Callable
 from contextlib import AsyncExitStack
 from typing import Any
-from uuid import uuid4
+from uuid import UUID, uuid4
 
 import pytest
 import sqlalchemy
@@ -31,6 +31,13 @@
 from simcore_postgres_database.models.funcapi_api_access_rights_table import (
     funcapi_api_access_rights_table,
 )
+from simcore_postgres_database.models.funcapi_function_jobs_table import (
+    function_jobs_table,
+)
+from simcore_postgres_database.models.funcapi_functions_access_rights_table import (
+    functions_access_rights_table,
+)
+from simcore_postgres_database.models.funcapi_functions_table import functions_table
 from simcore_service_webserver.application_settings import ApplicationSettings
 from simcore_service_webserver.statics._constants import FRONTEND_APP_DEFAULT
 from sqlalchemy.ext.asyncio import AsyncEngine
@@ -253,3 +260,71 @@ async def logged_user_function_api_access_rights(
     async with AsyncExitStack() as stack:
         row = await stack.enter_async_context(cm)
         yield row
+
+
+@pytest.fixture
+async def fake_function_with_associated_job(
+    asyncpg_engine: AsyncEngine,
+    logged_user: UserInfoDict,
+) -> AsyncIterator[UUID]:
+    async with AsyncExitStack() as stack:
+        function_row = await stack.enter_async_context(
+            insert_and_get_row_lifespan(
+                asyncpg_engine,
+                table=functions_table,
+                values={
+                    "title": "Test Function",
+                    "function_class": FunctionClass.PROJECT.value,
+                    "description": "A test function",
+                    "input_schema": {
+                        "schema_class": "application/schema+json",
+                        "schema_content": {
+                            "type": "object",
+                            "properties": {"input1": {"type": "string"}},
+                        },
+                    },
+                    "output_schema": {
+                        "schema_class": "application/schema+json",
+                        "schema_content": {
+                            "type": "object",
+                            "properties": {"output1": {"type": "string"}},
+                        },
+                    },
+                    "class_specific_data": {"project_id": f"{uuid4()}"},
+                },
+                pk_col=functions_table.c.uuid,
+            )
+        )
+
+        await stack.enter_async_context(
+            insert_and_get_row_lifespan(
+                asyncpg_engine,
+                table=functions_access_rights_table,
+                values={
+                    "function_uuid": function_row["uuid"],
+                    "group_id": logged_user["primary_gid"],
+                    "product_name": "osparc",  # Default product name
+                    "read": True,
+                    "write": True,
+                    "execute": True,
+                },
+                pk_cols=[
+                    functions_access_rights_table.c.function_uuid,
+                    functions_access_rights_table.c.group_id,
+                    functions_access_rights_table.c.product_name,
+                ],
+            )
+        )
+
+        await stack.enter_async_context(
+            insert_and_get_row_lifespan(
+                asyncpg_engine,
+                table=function_jobs_table,
+                values={
+                    "function_uuid": function_row["uuid"],
+                    "status": "pending",
+                },
+                pk_col=function_jobs_table.c.uuid,
+            )
+        )
+        yield function_row["uuid"]

services/web/server/tests/unit/with_dbs/04/functions/test_functions_controller_rest.py

@@ -7,7 +7,7 @@
 from collections.abc import AsyncIterator
 from http import HTTPStatus
 from typing import Any
-from uuid import uuid4
+from uuid import UUID, uuid4
 
 import pytest
 from aiohttp.test_utils import TestClient
@@ -306,3 +306,43 @@ async def test_list_user_functions_permissions(
     function_permissions = MyFunctionPermissionsGet.model_validate(data)
     assert function_permissions.read_functions == expected_read_functions
     assert function_permissions.write_functions == expected_write_functions
+
+
+@pytest.mark.parametrize(
+    "user_role,expected_read_functions,expected_write_functions",
+    [(UserRole.USER, True, True)],
+)
+async def test_delete_function_with_associated_jobs(
+    client: TestClient,
+    logged_user: UserInfoDict,
+    fake_function_with_associated_job: UUID,
+    logged_user_function_api_access_rights: dict[str, Any],
+) -> None:
+    function_id = fake_function_with_associated_job
+
+    url = client.app.router["get_function"].url_for(function_id=f"{function_id}")
+    response = await client.get(url)
+    data, error = await assert_status(response, status.HTTP_200_OK)
+    assert not error
+    function = TypeAdapter(RegisteredFunctionGet).validate_python(data)
+    assert function.uid == function_id
+
+    url = client.app.router["delete_function"].url_for(function_id=f"{function_id}")
+    response = await client.delete(url)
+    data, error = await assert_status(response, status.HTTP_409_CONFLICT)
+    assert error is not None
+
+    url = client.app.router["get_function"].url_for(function_id=f"{function_id}")
+    response = await client.get(url)
+    data, error = await assert_status(response, status.HTTP_200_OK)
+    assert not error
+
+    url = client.app.router["delete_function"].url_for(function_id=f"{function_id}")
+    response = await client.delete(url, params={"force": "true"})
+    data, error = await assert_status(response, status.HTTP_204_NO_CONTENT)
+    assert not error
+
+    url = client.app.router["get_function"].url_for(function_id=f"{function_id}")
+    response = await client.get(url)
+    data, error = await assert_status(response, status.HTTP_404_NOT_FOUND)
+    assert error is not None