add test

Author: giancarloromeo

services/web/server/tests/unit/with_dbs/01/storage/test_storage.py

@@ -722,3 +722,78 @@ async def test_get_async_job_links(
     assert response.status == return_status
     if return_schema:
         Envelope[return_schema].model_validate(await response.json())
+
+
+@pytest.fixture
+def create_consume_events_mock(
+    mocker: MockerFixture,
+) -> Callable[[Any], None]:
+    def _(result_or_exception: Any):
+        async def mock_consume_events(*args, **kwargs):
+            if isinstance(result_or_exception, Exception):
+                raise result_or_exception
+            # Yield the mock events
+            for event_id, event_data in result_or_exception:
+                yield event_id, event_data
+
+        mock_task_manager = mocker.MagicMock()
+        mock_task_manager.consume_task_events = mock_consume_events
+        mocker.patch(
+            "simcore_service_webserver.tasks._controller._rest.get_task_manager",
+            return_value=mock_task_manager,
+        )
+
+    return _
+
+
+class MockEvent:
+    def __init__(self, event_type: str, event_data: dict[str, Any]):
+        self.type = event_type
+        self.data = event_data
+
+
+@pytest.mark.parametrize("user_role", _user_roles)
+@pytest.mark.parametrize(
+    "backend_result_or_exception, expected_status",
+    [
+        (
+            [
+                ("event-1", MockEvent("status", {"status": "running"})),
+                ("event-2", MockEvent("progress", {"percent": 50})),
+                ("event-3", MockEvent("status", {"status": "completed"})),
+            ],
+            status.HTTP_200_OK,
+        ),
+        ([], status.HTTP_200_OK),  # No events
+    ],
+    ids=lambda x: (
+        "with_events" if x and isinstance(x, list) and len(x) > 0 else "no_events"
+    ),
+)
+async def test_get_async_job_stream(
+    user_role: UserRole,
+    logged_user: UserInfoDict,
+    client: TestClient,
+    create_consume_events_mock: Callable[[Any], None],
+    faker: Faker,
+    backend_result_or_exception: Any,
+    expected_status: int,
+):
+    create_consume_events_mock(backend_result_or_exception)
+
+    _task_id = AsyncJobId(faker.uuid4())
+
+    response = await client.get(
+        f"/{API_VERSION}/tasks/{_task_id}/stream",
+        headers={"Accept": "text/event-stream"},
+    )
+    assert response.status == expected_status
+
+    if response.status == status.HTTP_200_OK:
+        assert response.headers.get("Content-Type") == "text/event-stream"
+
+        content = await response.text()
+        if backend_result_or_exception:
+            assert "data:" in content or len(backend_result_or_exception) == 0
+        else:
+            assert content == ""