Cleanup test fixtures (#238)

Author: bisgaard-itis

clients/python/test/e2e/test_solvers_api.py

@@ -7,16 +7,57 @@
 import json
 
 import osparc
+import tenacity
 from _utils import skip_if_osparc_version
 from httpx import AsyncClient
 from packaging.version import Version
 from uuid import UUID
-
-DEFAULT_TIMEOUT_SECONDS = 15 * 60  # 10 min
+import pytest
+from contextlib import contextmanager
+from typing import Callable, Iterator, Set
+from tenacity import Retrying
+
+DEFAULT_TIMEOUT_SECONDS = 15 * 60  # 15 min
+
+
+@pytest.fixture
+def create_sleeper_jobs(
+    api_client: osparc.ApiClient,
+    sleeper: osparc.Solver,
+) -> Callable[[int], Iterator[Set[UUID | str]]]:
+    @contextmanager
+    def sleeper_jobs(n_jobs: int = 1) -> Iterator[Set[UUID | str]]:
+        job_ids = set()
+        solvers_api = osparc.SolversApi(api_client=api_client)
+        try:
+            for _ in range(n_jobs):
+                job = solvers_api.create_job(
+                    sleeper.id, sleeper.version, osparc.JobInputs({"input1": 1.0})
+                )
+                assert isinstance(job, osparc.Job)
+                print(job.to_str())
+                job_ids.add(job.id)
+            yield job_ids
+        finally:
+            for job_id in job_ids:
+                for attempt in Retrying(
+                    reraise=True,
+                    wait=tenacity.wait_fixed(2),
+                    stop=tenacity.stop_after_delay(60),
+                ):
+                    with attempt:
+                        solvers_api.stop_job(sleeper.id, sleeper.version, job_id)
+                        solvers_api.delete_job(sleeper.id, sleeper.version, job_id)
+
+    return sleeper_jobs
 
 
 @skip_if_osparc_version(at_least=Version("0.8.3.post0.dev20"))
-def test_jobs(api_client: osparc.ApiClient, sleeper: osparc.Solver):
+def test_jobs(
+    api_client: osparc.ApiClient,
+    create_sleeper_jobs: Callable[[int], Iterator[Set[UUID | str]]],
+    sleeper: osparc.Solver,
+):
     """Test the jobs method
 
     Args:
@@ -30,68 +71,56 @@ def test_jobs(api_client: osparc.ApiClient, sleeper: osparc.Solver):
     assert n_init_iter >= 0
 
     # create n_jobs jobs
-    created_job_ids = []
-    for _ in range(n_jobs):
-        job: osparc.Job = solvers_api.create_job(
-            sleeper.id, sleeper.version, osparc.JobInputs({"input1": 1.0})
-        )
-        created_job_ids.append(job.id)
-
-    tmp_iter = solvers_api.iter_jobs(sleeper.id, sleeper.version)
-    solvers_api.iter_jobs(sleeper.id, sleeper.version)
-
-    final_iter = solvers_api.iter_jobs(sleeper.id, sleeper.version)
-    assert len(final_iter) > 0, "No jobs were available"
-    assert n_init_iter + n_jobs == len(
-        final_iter
-    ), "An incorrect number of jobs was recorded"
-
-    for ii, elm in enumerate(tmp_iter):
-        assert isinstance(elm, osparc.Job)
-        if ii > 100:
-            break
+    with create_sleeper_jobs(n_jobs):
+        tmp_iter = solvers_api.iter_jobs(sleeper.id, sleeper.version)
+        solvers_api.iter_jobs(sleeper.id, sleeper.version)
+        final_iter = solvers_api.iter_jobs(sleeper.id, sleeper.version)
+        assert len(final_iter) > 0, "No jobs were available"
+        assert n_init_iter + n_jobs == len(
+            final_iter
+        ), "An incorrect number of jobs was recorded"
 
-    # cleanup
-    for elm in created_job_ids:
-        solvers_api.delete_job(sleeper.id, sleeper.version, elm)
+        for ii, elm in enumerate(tmp_iter):
+            assert isinstance(elm, osparc.Job)
+            if ii > 100:
+                break
 
 
 @skip_if_osparc_version(at_least=Version("0.6.5"))
 async def test_logstreaming(
-    api_client: osparc.ApiClient, sleeper: osparc.Solver, async_client: AsyncClient
+    api_client: osparc.ApiClient,
+    sleeper: osparc.Solver,
+    create_sleeper_jobs: Callable[[int], Iterator[Set[UUID]]],
+    async_client: AsyncClient,
 ):
     """Test log streaming"""
     solvers_api: osparc.SolversApi = osparc.SolversApi(api_client)
-    job = solvers_api.create_job(
-        sleeper.id, sleeper.version, osparc.JobInputs({"input1": 1.0})
-    )
-    assert isinstance(job, osparc.Job)
-    print(job.to_str())
-
-    solvers_api.start_job(sleeper.id, sleeper.version, job.id)
-
-    nloglines: int = 0
-    url = f"/v0/solvers/{sleeper.id}/releases/{sleeper.version}/jobs/{job.id}/logstream"
-    print(f"starting logstreaming from {url}...")
-
-    async with async_client.stream(
-        "GET",
-        url,
-        timeout=DEFAULT_TIMEOUT_SECONDS,
-    ) as response:
-        print(response.headers)
-        async for line in response.aiter_lines():
-            log = json.loads(line)
-            job_id = log.get("job_id")
-            assert job_id
-            assert job_id == (
-                f"{job.id}" if isinstance(job.id, UUID) else job.id
-            )  # keep test backwards compatible
-            nloglines += 1
-            print("\n".join(log.get("messages")))
-            await response.aclose()
-            break
-
-    assert nloglines > 0, f"Could not stream log for {sleeper.id=}, \
-        {sleeper.version=} and {job.id=}"  # type: ignore
-    solvers_api.delete_job(sleeper.id, sleeper.version, job.id)
+    with create_sleeper_jobs() as jobs:
+        job_id = next(iter(jobs))
+
+        solvers_api.start_job(sleeper.id, sleeper.version, job_id)
+
+        nloglines: int = 0
+        url = f"/v0/solvers/{sleeper.id}/releases/{sleeper.version}/jobs/{job_id}/logstream"
+        print(f"starting logstreaming from {url}...")
+
+        async with async_client.stream(
+            "GET",
+            url,
+            timeout=DEFAULT_TIMEOUT_SECONDS,
+        ) as response:
+            print(response.headers)
+            async for line in response.aiter_lines():
+                log = json.loads(line)
+                log_job_id = log.get("job_id")
+                assert log_job_id
+                assert log_job_id == (
+                    f"{job_id}" if isinstance(job_id, UUID) else job_id
+                )  # keep test backwards compatible
+                nloglines += 1
+                print("\n".join(log.get("messages")))
+                await response.aclose()
+                break
+
+        assert nloglines > 0, f"Could not stream log for {sleeper.id=}, \
+            {sleeper.version=} and {job_id=}"  # type: ignore

clients/python/test/e2e/test_studies.py

@@ -4,9 +4,11 @@
 # pylint: disable=unused-argument
 # pylint: disable=unused-variable
 
+from contextlib import contextmanager
 import shutil
 from datetime import timedelta
 from pathlib import Path
+from typing import Iterator
 from uuid import UUID
 
 import osparc
@@ -18,50 +20,80 @@
 _WAIT_TIMEOUT = timedelta(minutes=15)
 
 
+@pytest.fixture
+def create_study_job(
+    api_client: osparc.ApiClient, file_with_number: osparc.File, sleeper_study_id: UUID
+):
+    studies_api = osparc.StudiesApi(api_client=api_client)
+
+    @contextmanager
+    def _create_study_job() -> Iterator[osparc.Job]:
+        job_inputs = osparc.JobInputs(
+            values={
+                "input_file": file_with_number,
+            }
+        )
+        job = studies_api.create_study_job(
+            study_id=f"{sleeper_study_id}", job_inputs=job_inputs
+        )
+        assert isinstance(job, osparc.Job)
+        print(job)
+        try:
+            yield job
+        finally:
+            for attempt in tenacity.Retrying(
+                reraise=True,
+                wait=tenacity.wait_fixed(2),
+                stop=tenacity.stop_after_delay(60),
+            ):
+                with attempt:
+                    studies_api.stop_study_job(
+                        study_id=f"{sleeper_study_id}", job_id=job.id
+                    )
+                    studies_api.delete_study_job(
+                        study_id=f"{sleeper_study_id}", job_id=job.id
+                    )
+
+    return _create_study_job
+
+
 @skip_if_osparc_version(at_least=Version("0.6.6.post7"))
 @pytest.mark.parametrize("download_dir", [True, False])
 async def test_studies_logs(
     api_client: osparc.ApiClient,
-    file_with_number: osparc.File,
     sleeper_study_id: UUID,
+    create_study_job: Iterator[osparc.Job],
     download_dir: bool,
     tmp_path: Path,
 ):
     studies_api = osparc.StudiesApi(api_client=api_client)
-    job_inputs = osparc.JobInputs(
-        values={
-            "input_file": file_with_number,
-        }
-    )
-    job = studies_api.create_study_job(
-        study_id=f"{sleeper_study_id}", job_inputs=job_inputs
-    )
-    assert isinstance(job, osparc.Job)
-    print(job)
-    status = studies_api.start_study_job(study_id=f"{sleeper_study_id}", job_id=job.id)
-    assert isinstance(status, osparc.JobStatus)
-    async for attempt in tenacity.AsyncRetrying(
-        reraise=True,
-        wait=tenacity.wait_fixed(timedelta(seconds=5)),
-        stop=tenacity.stop_after_delay(_WAIT_TIMEOUT),
-        retry=tenacity.retry_if_exception_type(AssertionError),
-    ):
-        with attempt:
-            status = studies_api.inspect_study_job(
-                study_id=f"{sleeper_study_id}", job_id=job.id
-            )
-            assert isinstance(status, osparc.JobStatus)
-            print(f"--- seconds idle: {attempt.retry_state.idle_for}\n", status)
-            assert status.stopped_at is not None
-    assert status.state == "SUCCESS"
-    try:
-        log_dir = await studies_api.get_study_job_output_logfiles_async(
-            study_id=f"{sleeper_study_id}",
-            job_id=job.id,
-            download_dir=tmp_path if download_dir else None,
+    with create_study_job() as job:
+        status = studies_api.start_study_job(
+            study_id=f"{sleeper_study_id}", job_id=job.id
         )
-        assert log_dir.is_dir()
-        n_logfiles = sum(1 for _ in log_dir.rglob("*") if _.is_file())
-        assert n_logfiles > 0
-    finally:
-        shutil.rmtree(log_dir)
+        assert isinstance(status, osparc.JobStatus)
+        async for attempt in tenacity.AsyncRetrying(
+            reraise=True,
+            wait=tenacity.wait_fixed(timedelta(seconds=5)),
+            stop=tenacity.stop_after_delay(_WAIT_TIMEOUT),
+            retry=tenacity.retry_if_exception_type(AssertionError),
+        ):
+            with attempt:
+                status = studies_api.inspect_study_job(
+                    study_id=f"{sleeper_study_id}", job_id=job.id
+                )
+                assert isinstance(status, osparc.JobStatus)
+                print(f"--- seconds idle: {attempt.retry_state.idle_for}\n", status)
+                assert status.stopped_at is not None
+        assert status.state == "SUCCESS"
+        try:
+            log_dir = await studies_api.get_study_job_output_logfiles_async(
+                study_id=f"{sleeper_study_id}",
+                job_id=job.id,
+                download_dir=tmp_path if download_dir else None,
+            )
+            assert log_dir.is_dir()
+            n_logfiles = sum(1 for _ in log_dir.rglob("*") if _.is_file())
+            assert n_logfiles > 0
+        finally:
+            shutil.rmtree(log_dir)