JMS mk2 testing

src/ansys/hps/client/exceptions.py

@@ -21,8 +21,12 @@
 # SOFTWARE.
 """Module providing the base class for all client and server HPS-related errors."""
 
+import logging
+
 from requests.exceptions import RequestException
 
+log = logging.getLogger(__name__)
+
 
 class HPSError(RequestException):
     """Provides the base class for all HPS-related errors.
@@ -76,42 +80,47 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
 
+description_fields = ["description", "error_description", "detail"]
+
+
+def _build_error_message(category, response):
+    """Builds an error message from a response object."""
+    r_content = {}
+    try:
+        r_content = response.json()
+    except ValueError:
+        pass
+
+    reason = r_content.get("title", None)  # jms api
+    if not reason:
+        reason = r_content.get("error", None)  # auth api
+    if not reason:
+        reason = response.reason
+
+    description = None
+    for field in description_fields:
+        description = r_content.get(field, None)
+        if description:
+            break
+
+    msg = (
+        f"{response.status_code} {category}: {reason} for: {response.request.method} {response.url}"
+    )
+    if description:
+        msg += f"\n{description}"
+
+    return reason, description, msg
+
+
 def raise_for_status(response, *args, **kwargs):
     """Automatically checks HTTP errors.
 
     This method mimics the requests.Response.raise_for_status() method.
     """
-    if 400 <= response.status_code < 600:
-        r_content = {}
-        try:
-            r_content = response.json()
-        except ValueError:
-            pass
-
-        reason = r_content.get("title", None)  # jms api
-        if not reason:
-            reason = r_content.get("error", None)  # auth api
-        if not reason:
-            reason = response.reason
-
-        description = r_content.get("description", None)  # jms api
-        if not description:
-            description = r_content.get("error_description", None)  # auth api
-
-        if 400 <= response.status_code < 500:
-            error_msg = (
-                f"{response.status_code} Client Error: {reason} for:"
-                f" {response.request.method} {response.url}"
-            )
-            if description:
-                error_msg += f"\n{description}"
-            raise ClientError(error_msg, reason=reason, description=description, response=response)
-        elif 500 <= response.status_code < 600:
-            error_msg = (
-                f"{response.status_code} Server Error: {reason} for:"
-                f" {response.request.method} {response.url}"
-            )
-            if description:
-                error_msg += f"\n{description}"
-            raise APIError(error_msg, reason=reason, description=description, response=response)
+    if 400 <= response.status_code < 500:
+        reason, description, message = _build_error_message("Client Error", response)
+        raise ClientError(message, reason=reason, description=description, response=response)
+    elif 500 <= response.status_code < 600:
+        reason, description, message = _build_error_message("Server Error", response)
+        raise APIError(message, reason=reason, description=description, response=response)
     return response

tests/auth/test_api.py

@@ -102,6 +102,7 @@ def test_get_user_permissions(client):
     assert api.user_is_admin is not None
 
 
+@pytest.mark.skip(reason="no auth in mk2 yet")
 def test_impersonate_user(url, keycloak_client):
     """Test token exchange for impersonation, see https://www.rfc-editor.org/rfc/rfc8693.html.
 

tests/jms/test_job_definitions.py

@@ -104,8 +104,9 @@ def test_task_definition_fields(client, has_hps_version_ge_1_3_45, has_hps_versi
     if has_hps_version_gt_1_3_45:
         assert task_def.working_directory == "/tmp"
 
-    assert auth_api.get_user(id=task_def.created_by).username == client.username
-    assert auth_api.get_user(id=task_def.modified_by).username == client.username
+    if task_def.created_by is not None:
+        assert auth_api.get_user(id=task_def.created_by).username == client.username
+        assert auth_api.get_user(id=task_def.modified_by).username == client.username
 
     jms_api.delete_project(project)
 

tests/jms/test_jobs.py

@@ -184,8 +184,10 @@ def test_job_integration(client):
         assert job.executed_level is not None
         assert job.modified_by is not missing
         assert job.created_by is not missing
-        assert auth_api.get_user(id=job.created_by).username == client.username
-        assert auth_api.get_user(id=job.modified_by).username == client.username
+        if job.created_by is not None:
+            assert auth_api.get_user(id=job.created_by).username == client.username
+        if job.modified_by is not None:
+            assert auth_api.get_user(id=job.modified_by).username == client.username
         # fill some of them
         job.creator = "hps-client"
         job.note = f"test job{job.id} update"

tests/jms/test_project_permissions.py

@@ -24,6 +24,8 @@
 import time
 import uuid
 
+import pytest
+
 from ansys.hps.client import Client
 from ansys.hps.client.auth import AuthApi, User
 from ansys.hps.client.exceptions import ClientError
@@ -62,6 +64,7 @@ def remove_permissions(project_api: ProjectApi, user):
     log.info(f"Permissions after: {permissions}")
 
 
+@pytest.mark.skip("not available in mk2 at the moment")
 def test_get_project_permissions(client):
     jms_api = JmsApi(client)
     proj_name = "test_jms_get_permissions_test"
@@ -80,6 +83,7 @@ def test_get_project_permissions(client):
     jms_api.delete_project(proj)
 
 
+@pytest.mark.skip("not available in mk2 at the moment")
 def test_modify_project_permissions(client, keycloak_client):
     user_credentials = {
         "user1": {"username": f"testuser-{uuid.uuid4().hex[:8]}", "password": "test"},

tests/jms/test_projects.py

@@ -131,6 +131,7 @@ def test_project_replace(client):
     assert p.name == "Replaced Project"
 
 
+@pytest.mark.skip("not in mk2 at the moment")
 def test_project_copy(client):
     jms_api = JmsApi(client)
     proj_name = "test_jms_ProjectCopyTest"
@@ -228,7 +229,9 @@ def test_project_delete_job_definition(client):
     jms_api.delete_project(proj)
 
 
+@pytest.mark.skip("not in mk2 at the moment")
 def test_project_archive_restore(client):
+
     num_jobs = 2
     jms_api = JmsApi(client)
     proj_name = "test_jms_project_archive_restore"

tests/jms/test_task_definition_templates.py

@@ -24,6 +24,7 @@
 import logging
 import uuid
 
+import pytest
 from marshmallow.utils import missing
 
 from ansys.hps.client import HPSError
@@ -200,6 +201,7 @@ def test_template_integration(jms_api, request):
     jms_api.delete_task_definition_templates([new_template])
 
 
+@pytest.mark.skip("not in mk2 at the moment")
 def test_template_permissions(client, keycloak_client, is_admin, request):
     jms_api = JmsApi(client)
 
@@ -297,6 +299,7 @@ def test_template_permissions(client, keycloak_client, is_admin, request):
     delete_user(keycloak_client, user1)
 
 
+@pytest.mark.skip("not in mk2 at the moment")
 def test_template_permissions_update(jms_api, request):
     xfail_for_hps_version_under(HpsRelease.v1_3_45, jms_api, request)
 
@@ -322,6 +325,7 @@ def test_template_permissions_update(jms_api, request):
     jms_api.delete_task_definition_templates([template])
 
 
+@pytest.mark.skip("not in mk2 at the moment")
 def test_template_anyone_permission(client, keycloak_client, request):
     jms_api = JmsApi(client)
     xfail_for_hps_version_under(HpsRelease.v1_3_45, jms_api, request)
@@ -386,6 +390,7 @@ def test_template_anyone_permission(client, keycloak_client, request):
     delete_user(keycloak_client, user1)
 
 
+@pytest.mark.skip("not in mk2 at the moment")
 def test_template_delete(client, keycloak_client, request):
     xfail_for_hps_version_under(HpsRelease.v1_3_45, JmsApi(client), request)
 

tests/jms/test_tasks.py

@@ -305,11 +305,14 @@ def test_register_external_job(client):
 
     job = Job(
         name="Fluent Session",
-        eval_status="running",
+        eval_status="pending",
         job_definition_id=job_def.id,
     )
     job = project_api.create_jobs([job])[0]
 
+    job.eval_status = "running"
+    job = project_api.update_jobs([job])[0]
+
     # add custom data to the task
     tasks = project_api.get_tasks(job_id=job.id)
     assert len(tasks) == 1

tests/test_exceptions.py

@@ -36,12 +36,24 @@ def test_server_error(client):
     try:
         jms_api.get_projects(wrong_query_param="value")
     except APIError as e:
+        # JMS mk1
+        except_obj = e
+        log.error(str(e))
+    except ClientError as e:
+        # JMS mk2
         except_obj = e
         log.error(str(e))
 
-    assert except_obj.reason == "500 Internal Server Error"
-    assert except_obj.description == "type object 'Project' has no attribute 'wrong_query_param'"
-    assert except_obj.response.status_code == 500
+    if except_obj.response.status_code == 500:
+        assert except_obj.reason == "500 Internal Server Error"
+        assert (
+            except_obj.description == "type object 'Project' has no attribute 'wrong_query_param'"
+        )
+    elif except_obj.response.status_code == 400:
+        assert "Bad Request" in except_obj.reason
+        assert "invalid field filter" in except_obj.description
+    else:
+        raise AssertionError(f"Unexpected status code: {except_obj.response.status_code}")
 
 
 @pytest.mark.xfail