test: move wait_task_completion function in utils.py for better r…

…euse

tests/integration/test_studies_upgrade.py

@@ -1,36 +1,13 @@
 import os
-import time
 
 import pytest
+from antarest.core.tasks.model import TaskStatus
 from starlette.testclient import TestClient
-
-from antarest.core.tasks.model import TaskDTO, TaskStatus
+from tests.integration.utils import wait_task_completion
 
 RUN_ON_WINDOWS = os.name == "nt"
 
 
-def wait_task_completion(
-    client: TestClient,
-    access_token: str,
-    task_id: str,
-    *,
-    timeout: float = 10,
-) -> TaskDTO:
-    end_time = time.time() + timeout
-    while time.time() < end_time:
-        time.sleep(0.1)
-        res = client.get(
-            f"/v1/tasks/{task_id}",
-            headers={"Authorization": f"Bearer {access_token}"},
-            json={"wait_for_completion": True},
-        )
-        assert res.status_code == 200
-        task = TaskDTO(**res.json())
-        if task.status not in {TaskStatus.PENDING, TaskStatus.RUNNING}:
-            return task
-    raise TimeoutError(f"{timeout} seconds")
-
-
 class TestStudyUpgrade:
     @pytest.mark.skipif(
         RUN_ON_WINDOWS, reason="This test runs randomly on Windows"

tests/integration/utils.py

@@ -2,6 +2,9 @@
 import time
 from typing import Callable
 
+from antarest.core.tasks.model import TaskDTO, TaskStatus
+from starlette.testclient import TestClient
+
 
 def wait_for(
     predicate: Callable[[], bool], timeout: float = 10, sleep_time: float = 1
@@ -13,3 +16,25 @@ def wait_for(
                 return
         time.sleep(sleep_time)
     raise TimeoutError(f"task is still in progress after {timeout} seconds")
+
+
+def wait_task_completion(
+    client: TestClient,
+    access_token: str,
+    task_id: str,
+    *,
+    timeout: float = 10,
+) -> TaskDTO:
+    end_time = time.time() + timeout
+    while time.time() < end_time:
+        time.sleep(0.1)
+        res = client.get(
+            f"/v1/tasks/{task_id}",
+            headers={"Authorization": f"Bearer {access_token}"},
+            json={"wait_for_completion": True},
+        )
+        assert res.status_code == 200
+        task = TaskDTO(**res.json())
+        if task.status not in {TaskStatus.PENDING, TaskStatus.RUNNING}:
+            return task
+    raise TimeoutError(f"{timeout} seconds")