address review comments

components/renku_data_services/message_queue/api.spec.yaml

@@ -9,17 +9,23 @@ servers:
   - url: /api/data
   - url: /ui-server/api/data
 paths:
-  /search/reprovision:
+  /message_queue/reprovision:
     post:
       summary: Start a new reprovisioning
       description: Only a single reprovisioning is active at any time
       responses:
         "201":
           description: The reprovisioning is/will be started
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/Reprovisioning"
+        "409":
+          description: A reprovisioning is already started
         default:
           $ref: "#/components/responses/Error"
       tags:
-        - search
+        - message_queue
     get:
       summary: Return status of reprovisioning
       responses:
@@ -38,7 +44,7 @@ paths:
         default:
           $ref: "#/components/responses/Error"
       tags:
-        - search
+        - message_queue
     delete:
       summary: Stop an active reprovisioning
       responses:
@@ -47,28 +53,37 @@ paths:
         default:
           $ref: "#/components/responses/Error"
       tags:
-        - search
+        - message_queue
 
 components:
   schemas:
-    ReprovisioningStatus:
-      description: Status of a reprovisioning
+    Reprovisioning:
+      description: A reprovisioning
       type: object
       properties:
-        active:
-          type: boolean
-          description: Whether a reprovisioning is in progress or not
+        id:
+          $ref: "#/components/schemas/Ulid"
         start_date:
-          description: The date and time the resource was created (in UTC and ISO-8601 format)
+          description: The date and time the reprovisioning was started (in UTC and ISO-8601 format)
           type: string
           format: date-time
           example: "2023-11-01T17:32:28Z"
       required:
-        - active
+        - id
+        - start_date
       example:
-        - active: true
+        - id: 01BX5ZZK2KAC4AV9WEV3EMM0S0
           start_date: "2023-11-01T17:32:28Z"
-        - active: false
+    ReprovisioningStatus:
+      description: Status of a reprovisioning
+      allOf:
+      - $ref: "#/components/schemas/Reprovisioning"
+    Ulid:
+      description: ULID identifier
+      type: string
+      minLength: 26
+      maxLength: 26
+      pattern: "^[0-7][0-9A-HJKMNP-TV-Z]{25}$" # This is case-insensitive
     ErrorResponse:
       type: object
       properties:

components/renku_data_services/message_queue/apispec.py

@@ -1,6 +1,6 @@
 # generated by datamodel-codegen:
 #   filename:  api.spec.yaml
-#   timestamp: 2024-09-10T00:17:57+00:00
+#   timestamp: 2024-10-03T07:51:44+00:00
 
 from __future__ import annotations
 
@@ -11,17 +11,6 @@
 from renku_data_services.message_queue.apispec_base import BaseAPISpec
 
 
-class ReprovisioningStatus(BaseAPISpec):
-    active: bool = Field(
-        ..., description="Whether a reprovisioning is in progress or not"
-    )
-    start_date: Optional[datetime] = Field(
-        None,
-        description="The date and time the resource was created (in UTC and ISO-8601 format)",
-        example="2023-11-01T17:32:28Z",
-    )
-
-
 class Error(BaseAPISpec):
     code: int = Field(..., example=1404, gt=0)
     detail: Optional[str] = Field(
@@ -32,3 +21,22 @@ class Error(BaseAPISpec):
 
 class ErrorResponse(BaseAPISpec):
     error: Error
+
+
+class Reprovisioning(BaseAPISpec):
+    id: str = Field(
+        ...,
+        description="ULID identifier",
+        max_length=26,
+        min_length=26,
+        pattern="^[0-7][0-9A-HJKMNP-TV-Z]{25}$",
+    )
+    start_date: datetime = Field(
+        ...,
+        description="The date and time the reprovisioning was started (in UTC and ISO-8601 format)",
+        example="2023-11-01T17:32:28Z",
+    )
+
+
+class ReprovisioningStatus(Reprovisioning):
+    pass

components/renku_data_services/message_queue/blueprints.py

@@ -36,7 +36,7 @@ def post(self) -> BlueprintFactoryResponse:
 
         @authenticate(self.authenticator)
         @only_admins
-        async def _post(request: Request, user: base_models.APIUser) -> HTTPResponse:
+        async def _post(request: Request, user: base_models.APIUser) -> HTTPResponse | JSONResponse:
             reprovisioning = await self.reprovisioning_repo.start()
 
             request.app.add_task(
@@ -54,19 +54,21 @@ async def _post(request: Request, user: base_models.APIUser) -> HTTPResponse:
                 name=f"reprovisioning-{reprovisioning.id}",
             )
 
-            return HTTPResponse(status=201)
+            return json({"id": str(reprovisioning.id), "start_date": reprovisioning.start_date.isoformat()}, 201)
 
-        return "/search/reprovision", ["POST"], _post
+        return "/message_queue/reprovision", ["POST"], _post
 
     def get_status(self) -> BlueprintFactoryResponse:
         """Get reprovisioning status."""
 
         @authenticate(self.authenticator)
         async def _get_status(_: Request, __: base_models.APIUser) -> JSONResponse | HTTPResponse:
-            active_reprovisioning = await self.reprovisioning_repo.get_active_reprovisioning()
-            return json({"active": bool(active_reprovisioning)}, 200)
+            reprovisioning = await self.reprovisioning_repo.get_active_reprovisioning()
+            if not reprovisioning:
+                return HTTPResponse(status=404)
+            return json({"id": str(reprovisioning.id), "start_date": reprovisioning.start_date.isoformat()})
 
-        return "/search/reprovision", ["GET"], _get_status
+        return "/message_queue/reprovision", ["GET"], _get_status
 
     def delete(self) -> BlueprintFactoryResponse:
         """Stop reprovisioning (if any)."""
@@ -77,4 +79,4 @@ async def _delete(_: Request, __: base_models.APIUser) -> HTTPResponse:
             await self.reprovisioning_repo.stop()
             return HTTPResponse(status=204)
 
-        return "/search/reprovision", ["DELETE"], _delete
+        return "/message_queue/reprovision", ["DELETE"], _delete

components/renku_data_services/message_queue/db.py

@@ -78,7 +78,7 @@ async def delete_event(self, id: int) -> None:
             stmt = delete(schemas.EventORM).where(schemas.EventORM.id == id)
             await session.execute(stmt)
 
-    async def clear(self) -> None:
+    async def delete_all_events(self) -> None:
         """Delete all events. This is only used when testing reprovisioning."""
         async with self.session_maker() as session, session.begin():
             await session.execute(delete(schemas.EventORM))

test/bases/renku_data_services/data_api/test_message_queue.py

@@ -6,9 +6,29 @@
 from test.bases.renku_data_services.data_api.utils import dataclass_to_str, deserialize_event
 
 
+@pytest.fixture
+def _reprovisioning(sanic_client, user_headers):
+    """Wait for the data service to finish the reprovisioning task."""
+
+    async def wait_helper():
+        total_wait_time = 0
+        while True:
+            await asyncio.sleep(0.1)
+            total_wait_time += 0.1
+
+            _, response = await sanic_client.get("/api/data/message_queue/reprovision", headers=user_headers)
+
+            if response.status_code == 404:
+                break
+            elif total_wait_time > 30:
+                assert False, "Reprovisioning was not finished after 30 seconds"
+
+    return wait_helper
+
+
 @pytest.mark.asyncio
-async def test_search_reprovisioning(
-    sanic_client, app_config, create_project, create_group, admin_headers, project_members
+async def test_message_queue_reprovisioning(
+    sanic_client, app_config, create_project, create_group, admin_headers, project_members, _reprovisioning
 ) -> None:
     await create_project("Project 1")
     await create_project("Project 2", visibility="public")
@@ -22,14 +42,15 @@ async def test_search_reprovisioning(
     events = await app_config.event_repo._get_pending_events()
 
     # NOTE: Clear all events before reprovisioning
-    await app_config.event_repo.clear()
+    await app_config.event_repo.delete_all_events()
 
-    _, response = await sanic_client.post("/api/data/search/reprovision", headers=admin_headers)
+    _, response = await sanic_client.post("/api/data/message_queue/reprovision", headers=admin_headers)
 
     assert response.status_code == 201, response.text
+    assert response.json["id"] is not None
+    assert response.json["start_date"] is not None
 
-    # NOTE: Wait for server to finish the reprovisioning task
-    await asyncio.sleep(2)
+    await _reprovisioning()
 
     reprovisioning_events = await app_config.event_repo._get_pending_events()
 
@@ -40,8 +61,8 @@ async def test_search_reprovisioning(
 
 
 @pytest.mark.asyncio
-async def test_search_only_admins_can_start_reprovisioning(sanic_client, user_headers) -> None:
-    _, response = await sanic_client.post("/api/data/search/reprovision", headers=user_headers)
+async def test_message_queue_only_admins_can_start_reprovisioning(sanic_client, user_headers) -> None:
+    _, response = await sanic_client.post("/api/data/message_queue/reprovision", headers=user_headers)
 
     assert response.status_code == 403, response.text
     assert "You do not have the required permissions for this operation." in response.json["error"]["message"]
@@ -53,48 +74,49 @@ async def long_reprovisioning_mock(*_, **__):
 
 
 @pytest.mark.asyncio
-async def test_search_multiple_reprovisioning_not_allowed(sanic_client, admin_headers, monkeypatch) -> None:
+async def test_message_queue_multiple_reprovisioning_not_allowed(sanic_client, admin_headers, monkeypatch) -> None:
     monkeypatch.setattr(renku_data_services.message_queue.blueprints, "reprovision", long_reprovisioning_mock)
 
-    _, response = await sanic_client.post("/api/data/search/reprovision", headers=admin_headers)
+    _, response = await sanic_client.post("/api/data/message_queue/reprovision", headers=admin_headers)
     assert response.status_code == 201, response.text
 
-    _, response = await sanic_client.post("/api/data/search/reprovision", headers=admin_headers)
+    _, response = await sanic_client.post("/api/data/message_queue/reprovision", headers=admin_headers)
 
     assert response.status_code == 409, response.text
     assert "A reprovisioning is already in progress" in response.json["error"]["message"]
 
 
 @pytest.mark.asyncio
-async def test_search_get_reprovisioning_status(sanic_client, admin_headers, user_headers, monkeypatch):
+async def test_message_queue_get_reprovisioning_status(sanic_client, admin_headers, user_headers, monkeypatch):
     monkeypatch.setattr(renku_data_services.message_queue.blueprints, "reprovision", long_reprovisioning_mock)
 
-    _, response = await sanic_client.get("/api/data/search/reprovision", headers=user_headers)
+    _, response = await sanic_client.get("/api/data/message_queue/reprovision", headers=user_headers)
 
-    assert response.status_code == 200, response.text
-    assert response.json["active"] is False
+    assert response.status_code == 404, response.text
 
     # NOTE: Start a reprovisioning
-    _, response = await sanic_client.post("/api/data/search/reprovision", headers=admin_headers)
+    _, response = await sanic_client.post("/api/data/message_queue/reprovision", headers=admin_headers)
     assert response.status_code == 201, response.text
 
-    _, response = await sanic_client.get("/api/data/search/reprovision", headers=user_headers)
+    _, response = await sanic_client.get("/api/data/message_queue/reprovision", headers=user_headers)
 
     assert response.status_code == 200, response.text
-    assert response.json["active"] is True
+    assert response.json["id"] is not None
+    assert response.json["start_date"] is not None
 
 
 @pytest.mark.asyncio
-async def test_search_can_stop_reprovisioning(sanic_client, admin_headers, monkeypatch) -> None:
+async def test_message_queue_can_stop_reprovisioning(sanic_client, admin_headers, monkeypatch) -> None:
     monkeypatch.setattr(renku_data_services.message_queue.blueprints, "reprovision", long_reprovisioning_mock)
 
-    _, response = await sanic_client.post("/api/data/search/reprovision", headers=admin_headers)
+    _, response = await sanic_client.post("/api/data/message_queue/reprovision", headers=admin_headers)
     assert response.status_code == 201, response.text
+    _, response = await sanic_client.get("/api/data/message_queue/reprovision", headers=admin_headers)
+    assert response.status_code == 200, response.text
 
-    _, response = await sanic_client.delete("/api/data/search/reprovision", headers=admin_headers)
+    _, response = await sanic_client.delete("/api/data/message_queue/reprovision", headers=admin_headers)
     assert response.status_code == 204, response.text
 
-    _, response = await sanic_client.get("/api/data/search/reprovision", headers=admin_headers)
+    _, response = await sanic_client.get("/api/data/message_queue/reprovision", headers=admin_headers)
 
-    assert response.status_code == 200, response.text
-    assert response.json["active"] is False
+    assert response.status_code == 404, response.text