harness.py

"""Runner emits real JUnit and a nonzero exit status for any failure/error."""

import argparse
import json
from pathlib import Path
import time
import unittest
import xml.etree.ElementTree as ET

from .cases import CASE_IDS, PaymentContract


class JUnitResult(unittest.TestResult):
    def __init__(self):
        super().__init__()
        self.records = []

    def startTest(self, test):
        super().startTest(test)
        self.started = time.monotonic()

    def record(self, test, outcome, detail=""):
        name = test._testMethodName.removeprefix("test_")
        self.records.append({"id": CASE_IDS[name], "classname": "payment_api", "name": name,
                             "status": outcome, "detail": detail,
                             "seconds": round(time.monotonic() - self.started, 6)})

    def addSuccess(self, test):
        super().addSuccess(test)
        self.record(test, "passed")

    def addFailure(self, test, error):
        super().addFailure(test, error)
        self.record(test, "failed", self._exc_info_to_string(error, test))

    def addError(self, test, error):
        super().addError(test, error)
        self.record(test, "error", self._exc_info_to_string(error, test))

    def addSkip(self, test, reason):
        super().addSkip(test, reason)
        self.record(test, "skipped", reason)


def run(implementation, fault, output):
    PaymentContract.implementation = implementation
    PaymentContract.fault = fault
    result = JUnitResult()
    unittest.defaultTestLoader.loadTestsFromTestCase(PaymentContract).run(result)
    suite = ET.Element("testsuite", name="quality-workflow-payment-demo", tests=str(result.testsRun),
                       failures=str(len(result.failures)), errors=str(len(result.errors)),
                       skipped=str(len(result.skipped)))
    for record in result.records:
        node = ET.SubElement(suite, "testcase", classname=record["classname"], name=record["name"],
                             time=str(record["seconds"]), file="independent/cases.py")
        if record["status"] != "passed":
            tag = {"failed": "failure", "error": "error", "skipped": "skipped"}[record["status"]]
            ET.SubElement(node, tag).text = record["detail"]
    output.parent.mkdir(parents=True, exist_ok=True)
    ET.indent(suite)
    ET.ElementTree(suite).write(output, encoding="utf-8", xml_declaration=True)
    summary = {"fault": fault, "tests": result.testsRun, "passed": sum(
        record["status"] == "passed" for record in result.records),
        "failures": [record["id"] for record in result.records if record["status"] == "failed"],
        "errors": [record["id"] for record in result.records if record["status"] == "error"],
        "skipped": [record["id"] for record in result.records if record["status"] == "skipped"]}
    print(json.dumps(summary, ensure_ascii=False))
    return 0 if result.wasSuccessful() and not result.skipped else 1


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--implementation", type=Path, default=Path("demo_app/payment_service.py"))
    parser.add_argument("--fault", choices=["none", "duplicate-write", "wrong-error"], default="none")
    parser.add_argument("--junit", type=Path, required=True)
    args = parser.parse_args()
    raise SystemExit(run(args.implementation.resolve(), args.fault, args.junit))