How to use the cwltest.utils.TestResult function in cwltest
To help you get started, we’ve selected a few cwltest examples, based on popular ways it is used in public projects.
Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
common-workflow-language / cwltest / cwltest / __init__.py View on Github 
raise subprocess.CalledProcessError(return_code, " ".join(test_command))
out = json.loads(outstr)
except ValueError as err:
_logger.error(str(err))
_logger.error(outstr)
_logger.error(outerr)
except subprocess.CalledProcessError as err:
if err.returncode == UNSUPPORTED_FEATURE and REQUIRED not in test.get(
"tags", ["required"]
):
return TestResult(
UNSUPPORTED_FEATURE, outstr, outerr, duration, args.classname
)
if test.get("should_fail", False):
return TestResult(0, outstr, outerr, duration, args.classname)
_logger.error(
u"""Test %i failed: %s""",
test_number,
" ".join([quote(tc) for tc in test_command]),
)
_logger.error(test.get("doc"))
if err.returncode == UNSUPPORTED_FEATURE:
_logger.error(u"Does not support required feature")
else:
_logger.error(u"Returned non-zero")
_logger.error(outerr)
return TestResult(1, outstr, outerr, duration, args.classname, str(err))
except (yamlscanner.ScannerError, TypeError) as err:
_logger.error(
u"""Test %i failed: %s""",
test_number,outstr, outerr = process.communicate(timeout=timeout)
return_code = process.poll()
duration = time.time() - start_time
if return_code:
raise subprocess.CalledProcessError(return_code, " ".join(test_command))
out = json.loads(outstr)
except ValueError as err:
_logger.error(str(err))
_logger.error(outstr)
_logger.error(outerr)
except subprocess.CalledProcessError as err:
if err.returncode == UNSUPPORTED_FEATURE and REQUIRED not in test.get(
"tags", ["required"]
):
return TestResult(
UNSUPPORTED_FEATURE, outstr, outerr, duration, args.classname
)
if test.get("should_fail", False):
return TestResult(0, outstr, outerr, duration, args.classname)
_logger.error(
u"""Test %i failed: %s""",
test_number,
" ".join([quote(tc) for tc in test_command]),
)
_logger.error(test.get("doc"))
if err.returncode == UNSUPPORTED_FEATURE:
_logger.error(u"Does not support required feature")
else:
_logger.error(u"Returned non-zero")
_logger.error(outerr)
return TestResult(1, outstr, outerr, duration, args.classname, str(err))if process.poll() is not None:
break
if process.returncode is None:
process.kill()
fail_message = ""
if test.get("should_fail", False):
_logger.warning(
u"""Test %i failed: %s""",
test_number,
u" ".join([quote(tc) for tc in test_command]),
)
_logger.warning(test.get("doc"))
_logger.warning(u"Returned zero but it should be non-zero")
return TestResult(1, outstr, outerr, duration, args.classname)
try:
compare(test.get("output"), out)
except CompareFail as ex:
_logger.warning(
u"""Test %i failed: %s""",
test_number,
u" ".join([quote(tc) for tc in test_command]),
)
_logger.warning(test.get("doc"))
_logger.warning(u"Compare failure %s", ex)
fail_message = str(ex)
if outdir:
shutil.rmtree(outdir, True)_logger.error(outerr)
except KeyboardInterrupt:
_logger.error(
u"""Test %i interrupted: %s""",
test_number,
u" ".join([quote(tc) for tc in test_command]),
)
raise
except subprocess.TimeoutExpired:
_logger.error(
u"""Test %i timed out: %s""",
test_number,
u" ".join([quote(tc) for tc in test_command]),
)
_logger.error(test.get("doc"))
return TestResult(2, outstr, outerr, timeout, args.classname, "Test timed out")
finally:
if process is not None and process.returncode is None:
_logger.error(u"""Terminating lingering process""")
process.terminate()
for _ in range(0, 3):
time.sleep(1)
if process.poll() is not None:
break
if process.returncode is None:
process.kill()
fail_message = ""
if test.get("should_fail", False):
_logger.warning(
u"""Test %i failed: %s""",UNSUPPORTED_FEATURE, outstr, outerr, duration, args.classname
)
if test.get("should_fail", False):
return TestResult(0, outstr, outerr, duration, args.classname)
_logger.error(
u"""Test %i failed: %s""",
test_number,
" ".join([quote(tc) for tc in test_command]),
)
_logger.error(test.get("doc"))
if err.returncode == UNSUPPORTED_FEATURE:
_logger.error(u"Does not support required feature")
else:
_logger.error(u"Returned non-zero")
_logger.error(outerr)
return TestResult(1, outstr, outerr, duration, args.classname, str(err))
except (yamlscanner.ScannerError, TypeError) as err:
_logger.error(
u"""Test %i failed: %s""",
test_number,
u" ".join([quote(tc) for tc in test_command]),
)
_logger.error(outstr)
_logger.error(u"Parse error %s", str(err))
_logger.error(outerr)
except KeyboardInterrupt:
_logger.error(
u"""Test %i interrupted: %s""",
test_number,
u" ".join([quote(tc) for tc in test_command]),
)
raisetry:
compare(test.get("output"), out)
except CompareFail as ex:
_logger.warning(
u"""Test %i failed: %s""",
test_number,
u" ".join([quote(tc) for tc in test_command]),
)
_logger.warning(test.get("doc"))
_logger.warning(u"Compare failure %s", ex)
fail_message = str(ex)
if outdir:
shutil.rmtree(outdir, True)
return TestResult(
(1 if fail_message else 0),
outstr,
outerr,
duration,
args.classname,
fail_message,
)cwltest
Common Workflow Language testing framework
