cwltest plugin for pytest (#74)

Co-authored-by: GlassOfWhiskey <[email protected]>

Author: mr-c

.coveragerc

@@ -1,13 +1,25 @@
+[paths]
+source =
+  cwltest
+  */site-packages/cwltest
+  */cwltest
+
 [run]
 branch = True
-source = cwltest
-omit = tests/*
+source_pkgs = cwltest
+omit =
+  tests/*
+  */site-packages/cwltest/tests/*
 
 [report]
 exclude_lines =
     if self.debug:
     pragma: no cover
     raise NotImplementedError
     if __name__ == .__main__.:
+    if TYPE_CHECKING:
+    \.\.\.
 ignore_errors = True
-omit = tests/*
+omit =
+  tests/*
+  */site-packages/cwltest/tests/*

Makefile

@@ -76,7 +76,7 @@ docs: FORCE
 clean: FORCE
 	rm -f ${MODILE}/*.pyc tests/*.pyc
 	python setup.py clean --all || true
-	rm -Rf .coverage
+	rm -Rf .coverage\.* .coverage
 	rm -f diff-cover.html
 
 # Linting and code style related targets
@@ -104,11 +104,11 @@ codespell:
 	codespell -w $(shell git ls-files | grep -v mypy-stubs | grep -v gitignore)
 
 ## format                 : check/fix all code indentation and formatting (runs black)
-format:
-	black setup.py setup.py cwltest tests mypy-stubs
+format: $(PYSOURCES) mypy-stubs
+	black $^
 
-format-check:
-	black --diff --check setup.py cwltest tests mypy-stubs
+format-check: $(PYSOURCES) mypy-stubs
+	black --diff --check $^
 
 ## pylint                 : run static code analysis on Python code
 pylint: $(PYSOURCES)
@@ -123,7 +123,9 @@ diff_pylint_report: pylint_report.txt
 	diff-quality --compare-branch=main --violations=pylint pylint_report.txt
 
 .coverage: $(PYSOURCES)
-	python setup.py test --addopts "--cov --cov-config=.coveragerc --cov-report= ${PYTEST_EXTRA}"
+	COV_CORE_SOURCE=cwltest COV_CORE_CONFIG=.coveragerc COV_CORE_DATAFILE=.coverage \
+		python -m pytest --cov --cov-append --cov-report=
+	# https://pytest-cov.readthedocs.io/en/latest/plugins.html#plugin-coverage
 
 coverage.xml: .coverage
 	coverage xml

README.rst

@@ -1,6 +1,6 @@
-==========================================
+##########################################
 Common Workflow Language testing framework
-==========================================
+##########################################
 
 |Linux Build Status| |Code coverage|
 
@@ -34,8 +34,12 @@ conformance tests.
 
 This is written and tested for Python 3.6, 3.7, 3.8, 3.9, 3.10, and 3.11.
 
+.. contents:: Table of Contents
+   :local:
+
+*******
 Install
--------
+*******
 
 Installing the official package from PyPi
 
@@ -56,15 +60,17 @@ Or from source
   git clone https://github.com/common-workflow-language/cwltest.git
   cd cwltest && python setup.py install
 
+***********************
 Run on the command line
------------------------
+***********************
 
 Simple command::
 
   cwltest --test test-descriptions.yml --tool cwl-runner
 
+*****************************************
 Generate conformance badges using cwltest
------------------------------------------
+*****************************************
 
 To make badges that show the results of the conformance test,
 you can generate JSON files for https://badgen.net by using --badgedir option

cwltest/__init__.py

@@ -1,4 +1,4 @@
-"""Common Workflow Language testing framework."""
+"""Run CWL descriptions with a cwl-runner, and look for expected output."""
 
 import logging
 import threading

cwltest/argparser.py

@@ -7,7 +7,7 @@
 from cwltest import DEFAULT_TIMEOUT
 
 
-def arg_parser():  # type: () -> argparse.ArgumentParser
+def arg_parser() -> argparse.ArgumentParser:
     """Generate a command Line argument parser for cwltest."""
     parser = argparse.ArgumentParser(
         description="Common Workflow Language testing framework"
@@ -60,7 +60,8 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
     parser.add_argument(
         "--junit-verbose",
         action="store_true",
-        help="Store more verbose output to JUnit xml file",
+        help="Store more verbose output to JUnit XML file by not passing "
+        "'--quiet' to the CWL runner.",
     )
     parser.add_argument(
         "--test-arg",
@@ -101,7 +102,9 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
         ),
     )
     parser.add_argument(
-        "--badgedir", type=str, help="Directory that stores JSON files for badges."
+        "--badgedir",
+        type=str,
+        help="Create JSON badges and store them in this directory.",
     )
 
     pkg = pkg_resources.require("cwltest")

cwltest/compare.py

@@ -0,0 +1,164 @@
+"""Compare utilities for CWL objects."""
+
+import json
+from typing import Any, Dict, Set
+
+
+class CompareFail(Exception):
+    """Compared CWL objects are not equal."""
+
+    @classmethod
+    def format(cls, expected, actual, cause=None):
+        # type: (Any, Any, Any) -> CompareFail
+        """Load the difference details into the error message."""
+        message = "expected: {}\ngot: {}".format(
+            json.dumps(expected, indent=4, sort_keys=True),
+            json.dumps(actual, indent=4, sort_keys=True),
+        )
+        if cause:
+            message += "\ncaused by: %s" % cause
+        return cls(message)
+
+
+def _check_keys(keys, expected, actual):
+    # type: (Set[str], Dict[str,Any], Dict[str,Any]) -> None
+    for k in keys:
+        try:
+            compare(expected.get(k), actual.get(k))
+        except CompareFail as e:
+            raise CompareFail.format(
+                expected, actual, f"field '{k}' failed comparison: {str(e)}"
+            ) from e
+
+
+def _compare_contents(expected, actual):
+    # type: (Dict[str,Any], Dict[str,Any]) -> None
+    expected_contents = expected["contents"]
+    with open(actual["path"]) as f:
+        actual_contents = f.read()
+    if expected_contents != actual_contents:
+        raise CompareFail.format(
+            expected,
+            actual,
+            json.dumps(
+                "Output file contents do not match: actual '%s' is not equal to expected '%s'"
+                % (actual_contents, expected_contents)
+            ),
+        )
+
+
+def _compare_dict(expected, actual):
+    # type: (Dict[str,Any], Dict[str,Any]) -> None
+    for c in expected:
+        try:
+            compare(expected[c], actual.get(c))
+        except CompareFail as e:
+            raise CompareFail.format(
+                expected, actual, f"failed comparison for key '{c}': {e}"
+            ) from e
+    extra_keys = set(actual.keys()).difference(list(expected.keys()))
+    for k in extra_keys:
+        if actual[k] is not None:
+            raise CompareFail.format(expected, actual, "unexpected key '%s'" % k)
+
+
+def _compare_directory(expected, actual):
+    # type: (Dict[str,Any], Dict[str,Any]) -> None
+    if actual.get("class") != "Directory":
+        raise CompareFail.format(
+            expected, actual, "expected object with a class 'Directory'"
+        )
+    if "listing" not in actual:
+        raise CompareFail.format(
+            expected, actual, "'listing' is mandatory field in Directory object"
+        )
+    for i in expected["listing"]:
+        found = False
+        for j in actual["listing"]:
+            try:
+                compare(i, j)
+                found = True
+                break
+            except CompareFail:
+                pass
+        if not found:
+            raise CompareFail.format(
+                expected,
+                actual,
+                "%s not found" % json.dumps(i, indent=4, sort_keys=True),
+            )
+    _compare_file(expected, actual)
+
+
+def _compare_file(expected, actual):
+    # type: (Dict[str,Any], Dict[str,Any]) -> None
+    _compare_location(expected, actual)
+    if "contents" in expected:
+        _compare_contents(expected, actual)
+    other_keys = set(expected.keys()) - {"path", "location", "listing", "contents"}
+    _check_keys(other_keys, expected, actual)
+    _check_keys(other_keys, expected, actual)
+
+
+def _compare_location(expected, actual):
+    # type: (Dict[str,Any], Dict[str,Any]) -> None
+    if "path" in expected:
+        comp = "path"
+        if "path" not in actual:
+            actual["path"] = actual["location"]
+    elif "location" in expected:
+        comp = "location"
+    else:
+        return
+    if actual.get("class") == "Directory":
+        actual[comp] = actual[comp].rstrip("/")
+
+    if expected[comp] != "Any" and (
+        not (
+            actual[comp].endswith("/" + expected[comp])
+            or ("/" not in actual[comp] and expected[comp] == actual[comp])
+        )
+    ):
+        raise CompareFail.format(
+            expected,
+            actual,
+            f"{actual[comp]} does not end with {expected[comp]}",
+        )
+
+
+def compare(expected, actual):  # type: (Any, Any) -> None
+    """Compare two CWL objects."""
+    if expected == "Any":
+        return
+    if expected is not None and actual is None:
+        raise CompareFail.format(expected, actual)
+
+    try:
+        if isinstance(expected, dict):
+            if not isinstance(actual, dict):
+                raise CompareFail.format(expected, actual)
+
+            if expected.get("class") == "File":
+                _compare_file(expected, actual)
+            elif expected.get("class") == "Directory":
+                _compare_directory(expected, actual)
+            else:
+                _compare_dict(expected, actual)
+
+        elif isinstance(expected, list):
+            if not isinstance(actual, list):
+                raise CompareFail.format(expected, actual)
+
+            if len(expected) != len(actual):
+                raise CompareFail.format(expected, actual, "lengths don't match")
+            for c in range(0, len(expected)):
+                try:
+                    compare(expected[c], actual[c])
+                except CompareFail as e:
+                    raise CompareFail.format(expected, actual, e) from e
+        else:
+            if expected != actual:
+                raise CompareFail.format(expected, actual)
+
+    except Exception as e:
+        raise CompareFail(str(e)) from e

cwltest/hooks.py

@@ -0,0 +1,23 @@
+"""Hooks for pytest-cwl users."""
+
+from typing import Any, Dict, Optional, Tuple
+
+from cwltest import utils
+
+
+def pytest_cwl_execute_test(  # type: ignore[empty-body]
+    config: utils.CWLTestConfig, processfile: str, jobfile: Optional[str]
+) -> Tuple[int, Optional[Dict[str, Any]]]:
+    """
+    Execute CWL test using a Python function instead of a command line runner.
+
+    The return value is a tuple.
+     - status code
+        - 0 = success
+        - :py:attr:`cwltest.UNSUPPORTED_FEATURE` for an unsupported feature
+        - and any other number for failure
+     - CWL output object using plain Python objects.
+
+    :param processfile: a path to a CWL document
+    :param jobfile: an optionl path to JSON/YAML input object
+    """