full coverat in utils

Author: pcrespov

services/web/server/src/simcore_service_webserver/utils.py

@@ -65,7 +65,7 @@ class TaskInfoDict(TypedDict):
 
 def get_task_info(task: asyncio.Task) -> TaskInfoDict:
     def _format_frame(f):
-        return StackInfoDict(f_code=f.f_code, f_lineno=f.f_lineno)
+        return StackInfoDict(f_code=str(f.f_code), f_lineno=str(f.f_lineno))
 
     info = TaskInfoDict(
         txt=str(task),

services/web/server/tests/unit/isolated/test_utils.py

@@ -1,10 +1,13 @@
+import asyncio
+import contextlib
 import time
 import urllib.parse
 from datetime import datetime
 
 from simcore_service_webserver.utils import (
     DATETIME_FORMAT,
     compose_support_error_msg,
+    get_task_info,
     now_str,
     to_datetime,
 )
@@ -70,3 +73,74 @@ def test_compose_support_error_msg():
         msg == "First sentence for Mr.X. Second sentence."
         " For more information please forward this message to [email protected] (supportID=OEC:139641204989600)"
     )
+
+
+async def test_get_task_info():
+    """Test get_task_info function with asyncio tasks"""
+
+    async def dummy_task():
+        await asyncio.sleep(0.1)
+        return "task_result"
+
+    # Create a named task
+    task = asyncio.create_task(dummy_task(), name="test_task")
+
+    task_info = get_task_info(task)
+
+    # Check that task_info is a dictionary
+    assert isinstance(task_info, dict)
+
+    # Check that it contains expected keys from TaskInfoDict
+    expected_keys = {"txt", "type", "done", "cancelled", "stack", "exception"}
+    assert all(key in task_info for key in expected_keys)
+
+    # Check basic types
+    assert isinstance(task_info["txt"], str)
+    assert isinstance(task_info["type"], str)
+    assert isinstance(task_info["done"], bool)
+    assert isinstance(task_info["cancelled"], bool)
+    assert isinstance(task_info["stack"], list)
+
+    # Check that task name is in the txt representation
+    assert "test_task" in task_info["txt"]
+
+    # Check that stack contains frame info when task is running
+    if not task_info["done"]:
+        assert len(task_info["stack"]) > 0
+        # Check stack frame structure
+        for frame_info in task_info["stack"]:
+            assert "f_code" in frame_info
+            assert "f_lineno" in frame_info
+            assert isinstance(frame_info["f_code"], str)
+            assert isinstance(frame_info["f_lineno"], str)
+
+    # Clean up
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task
+
+
+async def test_get_task_info_unnamed_task():
+    """Test get_task_info function with unnamed tasks"""
+
+    async def dummy_task():
+        await asyncio.sleep(0.1)
+
+    # Create an unnamed task
+    task = asyncio.create_task(dummy_task())
+
+    task_info = get_task_info(task)
+
+    # Check basic structure
+    assert isinstance(task_info, dict)
+    expected_keys = {"txt", "type", "done", "cancelled", "stack", "exception"}
+    assert all(key in task_info for key in expected_keys)
+
+    # Check that txt contains task representation
+    assert isinstance(task_info["txt"], str)
+    assert "Task" in task_info["txt"]
+
+    # Clean up
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task