add test for logstreaming (#140)

* update workflow before publishing python package

* fix dependency issue and bump version

* point to website in project description

* fix broken dependency

* improve doc

* add github token to download artifacts

* ensure only read-access @wvangeit

* yet another attempt at downloading artifacts

* make sure to use repo that ran the trigger wf

* another attempt at fixing

* change owner

* allow publishing to testpypi also when pr

* minor change

* revert minor (but breaking) change

* minor fix

* add debug messages

* another debug message

* hopefully the final version

* final fix

* minor fix

* move master and tag to individual jobs

* add debug messages

* dev->post

* add python script for determining semantic version

* minor changes

* minor changes

* improve error handling and add version file to artifacts

* check if release

* minor fix

* ensure to enter venv

* also when tagging

* source venv in publishin workflow

* ensure only master

* add script for testing 'pure' semver

* adapt workflows to new python script

* minor change

* attempt to evaluate expressions correctly

* several fixes to fix tests

* ensure repo is checked out in publish workflow

* several small fixes

* cleanup

* debug

* minor cleanup

* mionr changes

* add debug message

* minor change

* minor change

* yet another try

* minor change

* minor change

* minor change

* mionr change

* minor changes

* correct workflow run id

* cosmetic change

* avoid using gh

* change to a single job for publishing

* minor cleanup

* swap loops in clean up jobs

* correction

* update server compatibility to new url

* minor change to trigger ci

* allow for async tests in e2e

* add test

* fix test

* clea up tests

Author: bisgaard-itis

clients/python/requirements/e2e-test.txt

@@ -10,6 +10,7 @@ pandas
 papermill<2.5.0 # due to bug in 2.5.0 (https://github.com/nteract/papermill/issues/735). Remove this requirement once the bug is fixed
 pydantic
 pytest
+pytest-asyncio
 pytest-env
 pytest-html
 packaging

clients/python/test/e2e/ci/e2e/e2e/_models.py

@@ -82,6 +82,7 @@ class PytestConfig(BaseModel):
     env: str
     required_plugins: str
     addopts: str
+    asyncio_mode: str
 
 
 class Artifacts(BaseModel):

clients/python/test/e2e/ci/e2e/e2e/preprocess.py

@@ -73,8 +73,9 @@ def generate_ini(
     )
     pytest_config: PytestConfig = PytestConfig(
         env="\n" + "\n".join(envs),
-        required_plugins="pytest-env pytest-html",
+        required_plugins="pytest-env pytest-html pytest-asyncio",
         addopts=add_opts,
+        asyncio_mode="auto",
     )
 
     config: PytestIniFile = PytestIniFile(

clients/python/test/e2e/conftest.py

@@ -1,9 +1,11 @@
 import logging
 import os
 from pathlib import Path
+from typing import Iterable
 
 import osparc
 import pytest
+from httpx import AsyncClient, BasicAuth
 from numpy import random
 from pydantic import ByteSize
 
@@ -13,18 +15,31 @@
 
 
 @pytest.fixture
-def cfg() -> osparc.Configuration:
-    """Configuration
-
-    Returns:
-        osparc.Configuration: The Configuration
-    """
-    cfg = osparc.Configuration(
-        host=os.environ["OSPARC_API_HOST"],
-        username=os.environ["OSPARC_API_KEY"],
-        password=os.environ["OSPARC_API_SECRET"],
+def configuration() -> Iterable[osparc.Configuration]:
+    assert (host := os.environ.get("OSPARC_API_HOST"))
+    assert (username := os.environ.get("OSPARC_API_KEY"))
+    assert (password := os.environ.get("OSPARC_API_SECRET"))
+    yield osparc.Configuration(
+        host=host,
+        username=username,
+        password=password,
     )
-    return cfg
+
+
+@pytest.fixture
+def api_client(configuration: osparc.Configuration) -> Iterable[osparc.ApiClient]:
+    with osparc.ApiClient(configuration=configuration) as api_client:
+        yield api_client
+
+
+@pytest.fixture
+def async_client(configuration) -> Iterable[AsyncClient]:
+    yield AsyncClient(
+        base_url=configuration.host,
+        auth=BasicAuth(
+            username=configuration.username, password=configuration.password
+        ),
+    )  # type: ignore
 
 
 @pytest.fixture
@@ -42,3 +57,12 @@ def tmp_file(tmp_path: Path, caplog) -> Path:
         tmp_file.stat().st_size == byte_size
     ), f"Could not create file of size: {byte_size}"
     return tmp_file
+
+
+@pytest.fixture
+def sleeper(api_client: osparc.ApiClient) -> Iterable[osparc.Solver]:
+    solvers_api = osparc.SolversApi(api_client=api_client)
+    sleeper: osparc.Solver = solvers_api.get_solver_release(
+        "simcore/services/comp/itis/sleeper", "2.0.2"
+    )  # type: ignore
+    yield sleeper

clients/python/test/e2e/test_files_api.py

@@ -20,58 +20,54 @@ def _hash_file(file: Path) -> str:
 
 
 @requires_dev_features
-def test_upload_file(tmp_file: Path, cfg: osparc.Configuration) -> None:
+def test_upload_file(tmp_file: Path, api_client: osparc.ApiClient) -> None:
     """Test that we can upload a file via the multipart upload"""
     tmp_path: Path = tmp_file.parent
-    with osparc.ApiClient(cfg) as api_client:
-        files_api: osparc.FilesApi = osparc.FilesApi(api_client=api_client)
-        uploaded_file1: osparc.File = files_api.upload_file(tmp_file)
-        uploaded_file2: osparc.File = files_api.upload_file(tmp_file)
-        assert (
-            uploaded_file1.id == uploaded_file2.id
-        ), "could not detect that file was already on server"
-        downloaded_file = files_api.download_file(
-            uploaded_file1.id, destination_folder=tmp_path
-        )
-        assert Path(downloaded_file).parent == tmp_path
-        assert _hash_file(Path(downloaded_file)) == _hash_file(tmp_file)
-        files_api.delete_file(uploaded_file1.id)
+    files_api: osparc.FilesApi = osparc.FilesApi(api_client=api_client)
+    uploaded_file1: osparc.File = files_api.upload_file(tmp_file)
+    uploaded_file2: osparc.File = files_api.upload_file(tmp_file)
+    assert (
+        uploaded_file1.id == uploaded_file2.id
+    ), "could not detect that file was already on server"
+    downloaded_file = files_api.download_file(
+        uploaded_file1.id, destination_folder=tmp_path
+    )
+    assert Path(downloaded_file).parent == tmp_path
+    assert _hash_file(Path(downloaded_file)) == _hash_file(tmp_file)
+    files_api.delete_file(uploaded_file1.id)
 
 
 @requires_dev_features
 @pytest.mark.parametrize("use_checksum", [True, False])
 @pytest.mark.parametrize("use_id", [True, False])
 def test_search_files(
-    tmp_file: Path, cfg: osparc.Configuration, use_checksum: bool, use_id: bool
+    tmp_file: Path, api_client: osparc.ApiClient, use_checksum: bool, use_id: bool
 ) -> None:
     checksum: str = _hash_file(tmp_file)
     results: osparc.PaginationGenerator
-    with osparc.ApiClient(configuration=cfg) as api_client:
-        files_api: osparc.FilesApi = osparc.FilesApi(api_client=api_client)
-        try:
-            results = files_api._search_files(sha256_checksum=checksum)
-            assert len(results) == 0, "Found file which shouldn't be there"
+    files_api: osparc.FilesApi = osparc.FilesApi(api_client=api_client)
+    try:
+        results = files_api._search_files(sha256_checksum=checksum)
+        assert len(results) == 0, "Found file which shouldn't be there"
 
-            uploaded_file: osparc.File = files_api.upload_file(tmp_file)
-            assert checksum == uploaded_file.checksum
+        uploaded_file: osparc.File = files_api.upload_file(tmp_file)
+        assert checksum == uploaded_file.checksum
 
-            results = files_api._search_files(
-                file_id=uploaded_file.id if use_id else None,
-                sha256_checksum=uploaded_file.checksum if use_checksum else None,
-            )
-            assert len(results) == 1, "Could not find file after it had been uploaded"
+        results = files_api._search_files(
+            file_id=uploaded_file.id if use_id else None,
+            sha256_checksum=uploaded_file.checksum if use_checksum else None,
+        )
+        assert len(results) == 1, "Could not find file after it had been uploaded"
 
-            files_api.delete_file(uploaded_file.id)
-            results = files_api._search_files(
-                file_id=uploaded_file.id if use_id else None,
-                sha256_checksum=uploaded_file.checksum if use_checksum else None,
-            )
-            assert (
-                len(results) == 0
-            ), "Could find file on server after it had been deleted"
+        files_api.delete_file(uploaded_file.id)
+        results = files_api._search_files(
+            file_id=uploaded_file.id if use_id else None,
+            sha256_checksum=uploaded_file.checksum if use_checksum else None,
+        )
+        assert len(results) == 0, "Could find file on server after it had been deleted"
 
-        except Exception:
-            # clean up in case of failure
-            results = files_api._search_files(sha256_checksum=checksum)
-            for file in results:
-                files_api.delete_file(file.id)
+    except Exception:
+        # clean up in case of failure
+        results = files_api._search_files(sha256_checksum=checksum)
+        for file in results:
+            files_api.delete_file(file.id)

clients/python/test/e2e/test_solvers_api.py

@@ -1,48 +1,87 @@
+import json
+
 import osparc
 from _utils import requires_dev_features
+from httpx import AsyncClient
+
+DEFAULT_TIMEOUT_SECONDS = 10 * 60  # 10 min
 
 
 @requires_dev_features
-def test_jobs(cfg: osparc.Configuration):
+def test_jobs(api_client: osparc.ApiClient, sleeper: osparc.Solver):
     """Test the jobs method
 
     Args:
         configuration (osparc.Configuration): The Configuration
     """
-    solver: str = "simcore/services/comp/itis/sleeper"
-    version: str = "2.0.2"
     n_jobs: int = 3
-    with osparc.ApiClient(cfg) as api_client:
-        solvers_api: osparc.SolversApi = osparc.SolversApi(api_client)
-        sleeper: osparc.Solver = solvers_api.get_solver_release(solver, version)
-
-        # initial iterator
-        init_iter = solvers_api.jobs(sleeper.id, sleeper.version)
-        n_init_iter: int = len(init_iter)
-        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.jobs(sleeper.id, sleeper.version)
-        solvers_api.jobs(sleeper.id, sleeper.version)
-
-        final_iter = solvers_api.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:
+    solvers_api = osparc.SolversApi(api_client=api_client)
+    # initial iterator
+    init_iter = solvers_api.jobs(sleeper.id, sleeper.version)
+    n_init_iter: int = len(init_iter)
+    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.jobs(sleeper.id, sleeper.version)
+    solvers_api.jobs(sleeper.id, sleeper.version)
+
+    final_iter = solvers_api.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
+
+    # cleanup
+    for elm in created_job_ids:
+        solvers_api.delete_job(sleeper.id, sleeper.version, elm)
+
+
+@requires_dev_features
+async def test_logstreaming(
+    api_client: osparc.ApiClient, sleeper: osparc.Solver, async_client: AsyncClient
+):
+    """Test the log streaming
+
+    Args:
+        configuration (osparc.Configuration): The Configuration
+    """
+    solvers_api: osparc.SolversApi = osparc.SolversApi(api_client)
+    job: osparc.Job = solvers_api.create_job(
+        sleeper.id, sleeper.version, osparc.JobInputs({"input1": 1.0})
+    )  # type: ignore
+
+    solvers_api.start_job(sleeper.id, sleeper.version, job.id)
+
+    nloglines: int = 0
+    print("starting logstreaming...")
+    async with async_client.stream(
+        "GET",
+        f"/v0/solvers/{sleeper.id}/releases/{sleeper.version}/jobs/{job.id}/logstream",
+        timeout=DEFAULT_TIMEOUT_SECONDS,
+    ) as response:
+        async for line in response.aiter_lines():
+            log = json.loads(line)
+            job_id = log.get("job_id")
+            assert job_id
+            assert job_id == job.id
+            nloglines += 1
+            print("\n".join(log.get("messages")))
+            if nloglines > 10:  # dont wait too long
+                await response.aclose()
                 break
 
-        # cleanup
-        for elm in created_job_ids:
-            solvers_api.delete_job(sleeper.id, sleeper.version, elm)
+    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)