WIP Feat/autocreate json testscripts

Pipfile

@@ -12,14 +12,14 @@ ipython = "~=8.11.0"
 
 [packages]
 fhirpy = "~=1.3.1"
-fhirpathpy = "~=0.1.2"
 pytest = "~=7.2.2"
 pytest-asyncio = "~=0.20.3"
 jsonpath-ng = "~=1.5.3"
 funcy = "~=1.18"
 aiofiles = "~=23.1.0"
 pyyaml = "~=6.0"
 pyyaml-include = "~=1.3"
+fhirpathpy = "~=2.2.4"
 
 [requires]
 python_version = "3.11"

README.md

@@ -23,7 +23,7 @@ docker run -v "$(pwd)/resources:/app" --env-file ./env/testscript bedasoftware/t
 ```
 ### Restrictions
 1. TestScript files should be stored in the TestScript folder;
-2. TestScript files should be in YAML format.
+2. TestScript files should be in YAML (`.yaml`) or JSON (`.json`) format.
 ## Recommendations
 Highly recommend using testscript-eval-py alongside [Kaitenzushi](https://github.com/beda-software/kaitenzushi/) to enhance the TestScript development experience.
 ## References

pyproject.toml

@@ -9,21 +9,19 @@ readme = "README.md"
 license = { file = "LICENSE" }
 keywords = ["fhir"]
 authors = [{ name = "beda.software", email = "ilya@beda.software" }]
-dependencies = ["requests>=2.25.1", "aiohttp>=3.6.3", "pytz", "aiofiles", "fhirpathpy", "pyyaml-include", "jsonpath_ng", "funcy", "fhirpy", "pytest_asyncio"]
+dependencies = ["requests>=2.25.1", "aiohttp>=3.6.3", "pytz", "aiofiles", "fhirpathpy~=2.2.4", "pyyaml-include", "jsonpath_ng", "funcy", "fhirpy", "pytest_asyncio"]
 dynamic = ["version"]
 classifiers = [
     "Development Status :: 4 - Beta",
     "Intended Audience :: Developers",
     "Operating System :: OS Independent",
     "Programming Language :: Python",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.8",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Topic :: Software Development :: Libraries :: Python Modules",
 ]
-requires-python = ">=3.8"
+requires-python = ">=3.10"
 
 [tool.black]
 line-length = 100
@@ -50,7 +48,7 @@ known_first_party = ["testscript_eval"]
 
 [tool.pytest.ini_options]
 minversion = "7.0"
-addopts = "-ra -q --color=yes --disable-warnings"
+addopts = "-ra -q --color=yes"
 testpaths = ["tests"]
 log_cli = true
 log_cli_level = "INFO"

testscript_eval/assertation.py

@@ -1,17 +1,23 @@
+import json
 import logging
 import re
+import warnings
 
-from fhirpathpy import evaluate as fhirpath
 from jsonpath_ng.ext import parse
 
 from . import operations
-from .misc import resolve_string_template
+from .misc import fhirpath, resolve_string_template
+from .operations import check
 
 
 class NotImplementedAssert(Exception):
     pass
 
 
+class TestScriptWarning(UserWarning):
+    """A failed warningOnly assert: reported, but does not fail the test script."""
+
+
 def not_implemented_assert(assert_name, _assertation, _result, _resource, _var):
     raise NotImplementedAssert(f"Assert rule {assert_name} is not implemented")
 
@@ -31,10 +37,11 @@ def response_code(_assert_name, assertation, result, _resource, _var):
 
 def resource(_assert_name, assertation, _result, resource, _var):
     resource_type = assertation["resource"]
-    if assertation.get("operator", "equals") == "equals":
-        assert resource_type == resource["resourceType"]
-    else:
-        assert resource_type == resource["resourceType"]
+    check(isinstance(resource, dict), "No resource in the response")
+    check(
+        resource_type == resource.get("resourceType"),
+        f"Expected resource {resource_type}, got {resource.get('resourceType')}",
+    )
 
 
 def path(_assert_name, assertation, result, resource, var):
@@ -50,19 +57,75 @@ def path(_assert_name, assertation, result, resource, var):
 def response(_assert_name, assertation, result, resource, var):
     response = assertation["response"]
     if response == "okay":
-        assert result.status == 200, f"Expected status 200 (okay), got {result.status}"
+        check(result.status == 200, f"Expected status 200 (okay), got {result.status}")
     elif response == "notFound":
-        assert result.status == 404, f"Expected status 404 (notFound), got {result.status}"
+        check(result.status == 404, f"Expected status 404 (notFound), got {result.status}")
     elif response == "bad":
-        assert result.status == 400, f"Expected status 400 (bad), got {result.status}"
+        check(result.status == 400, f"Expected status 400 (bad), got {result.status}")
     elif response == "created":
-        assert result.status == 201, f"Expected status 201 (created), got {result.status}"
+        check(result.status == 201, f"Expected status 201 (created), got {result.status}")
     else:
-        raise Exception(f"Response code {response.status} is not supported")
-
-
-def validate_profile_id(_assert_name, assertation, result, resource, var):
-    logging.warning("Profile validation is not implemented yet")
+        raise Exception(f"Response code {response} is not supported")
+
+
+def resolve_profile(profile_id, profiles):
+    # validateProfileId is the id of a TestScript.profile entry; a canonical URL given
+    # directly instead (as fhir-frog TestScripts do) is used as is
+    for profile in profiles:
+        if profile.get("id") == profile_id:
+            return profile["reference"]
+    if "/" in profile_id:
+        return profile_id
+    raise Exception(f"validateProfileId '{profile_id}' does not match any TestScript.profile id")
+
+
+async def validate_profile_id(assertation, resource, client, profiles):
+    profile_url = resolve_profile(assertation["validateProfileId"], profiles)
+    check(
+        isinstance(resource, dict) and "resourceType" in resource,
+        f"No resource to validate against {profile_url}",
+    )
+    # Pass the profile both ways: as the $validate `profile` parameter, and in meta.profile,
+    # which servers that ignore the parameter (e.g. Aidbox) validate against
+    meta = resource.get("meta", {})
+    payload = {
+        **resource,
+        "meta": {**meta, "profile": list(dict.fromkeys([*meta.get("profile", []), profile_url]))},
+    }
+    response, outcome = await client.execute(
+        f"{resource['resourceType']}/$validate",
+        method="post",
+        data=payload,
+        params={"profile": profile_url},
+    )
+    # No validation happened (auth, $validate unsupported, server error): an engine/infrastructure
+    # error, never a check failure, so warningOnly does not hide it
+    if response.status in (401, 403, 404, 405) or response.status >= 500:
+        raise Exception(
+            f"$validate against {profile_url} could not run: status {response.status}: {outcome}"
+        )
+    if not isinstance(outcome, dict):
+        # Non-2xx: the body comes back wrapped in an exception object
+        try:
+            outcome = json.loads(str(outcome))
+        except ValueError:
+            outcome = None
+    if not isinstance(outcome, dict) or outcome.get("resourceType") != "OperationOutcome":
+        raise Exception(
+            f"$validate against {profile_url} returned no OperationOutcome "
+            f"(status {response.status}): {outcome}"
+        )
+    errors = [
+        f"[{issue.get('severity')}] {', '.join(issue.get('expression', []))}: "
+        f"{issue.get('diagnostics') or issue.get('details', {}).get('text', '')}"
+        for issue in outcome.get("issue", [])
+        if issue.get("severity") in ("error", "fatal")
+    ]
+    # A 2xx, or a 4xx such as 400/422 with an OperationOutcome, is the validation result
+    check(
+        response.status < 400 and not errors,
+        f"Validation against {profile_url} failed (status {response.status}):\n" + "\n".join(errors),
+    )
 
 
 def parse_variable_in_string(string: str, variables):
@@ -92,6 +155,7 @@ def compare_response_expression_to_value(_assert_name, assertation, result, reso
     fixture = fixtures[id] if id else resource
     expression = assertation["expression"]
     expression_result = fhirpath(fixture, expression, {})
+    check(expression_result, f"Expression '{expression}' returned no value")
     if len(expression_result) > 1:
         logging.warning("Expression '%s' returns more then one value")
     operator = assertation.get("operator", "equals")
@@ -101,12 +165,29 @@ def compare_response_expression_to_value(_assert_name, assertation, result, reso
     operations.eval(operator, expression_result[0], value)
 
 
+def expression(_assert_name, assertation, result, resource, var):
+    # An expression with a value is compared by the "value" rule, and one with
+    # compareToSourceId by the "compareToSourceExpression" rule
+    if "value" in assertation or "compareToSourceId" in assertation:
+        return
+    # Without a value the expression is a boolean condition: operator "eval",
+    # which is also the default (https://hl7.org/fhir/R5/testing.html#assertion-table)
+    operator = assertation.get("operator", "eval")
+    if operator != "eval":
+        raise NotImplementedAssert(f"Operator {operator} without a value is not implemented")
+    fixtures = var["--fixtures--"]
+    id = assertation.get("sourceId")
+    source = fixtures[id] if id else resource
+    operations.eval(operator, fhirpath(source, assertation["expression"], {}), None)
+
+
 def minimum_id(_assert_name, assertation, result, resource, var):
     fixtures = var["--fixtures--"]
     id = assertation["minimumId"]
     fixture = fixtures[id]
+    check(isinstance(resource, dict), "No resource in the response")
     for k, v in fixture.items():
-        assert resource[k] == v
+        check(k in resource and resource[k] == v, f"minimumId {id}: {k} does not match")
 
 
 def content_type(_assert_name, assertation, result, resource, var):
@@ -117,16 +198,17 @@ def content_type(_assert_name, assertation, result, resource, var):
 
 
 def navigation_links(_assert_name, assertation, result, resource, var):
-    assert "link" in resource
+    check(isinstance(resource, dict), "No resource in the response")
+    check("link" in resource, "Response has no navigation links")
 
 
 assert_rules = {
     "headerField": header_field,
     "responseCode": response_code,
     "resource": resource,
     "path": path,
+    "expression": expression,
     "response": response,
-    "validateProfileId": validate_profile_id,
     "compareToSourceExpression": compare_to_source_expression,
     "minimumId": minimum_id,
     "contentType": content_type,
@@ -139,15 +221,26 @@ def navigation_links(_assert_name, assertation, result, resource, var):
 }
 
 
-def eval(assertation, result, resource, var, fixtures):
+async def eval(assertation, result, resource, var, fixtures, client, profiles):
     # TODO rid of temporary hack
     # pass fixtures as separated context item
     var["--fixtures--"] = fixtures
     logging.warning("%s", assertation["description"])
-    for assert_name, assert_eval in assert_rules.items():
-        if assertation.get("direction") == "request":
-            # Skip all request tests
-            logging.warning("Skip request validation")
-            continue
-        if assert_name in assertation:
-            assert_eval(assert_name, assertation, result, resource, var)
+    if assertation.get("direction") == "request":
+        # Skip all request tests
+        logging.warning("Skip request validation")
+        return
+    try:
+        for assert_name, assert_eval in assert_rules.items():
+            if assert_name in assertation:
+                assert_eval(assert_name, assertation, result, resource, var)
+        # Validation calls the server, so it is not one of the synchronous assert_rules
+        if "validateProfileId" in assertation:
+            await validate_profile_id(assertation, resource, client, profiles)
+    except AssertionError as e:
+        # A failed check of a warningOnly assert is reported and the script goes on; anything
+        # else (engine, script or infrastructure errors) is not an AssertionError and still raises
+        if not assertation.get("warningOnly", False):
+            raise
+        logging.warning("warningOnly assert failed: %s: %s", assertation["description"], e)
+        warnings.warn(TestScriptWarning(f"{assertation['description']}: {e}"))