cleveragents-core/robot/_tdd_fixture_runner.py

"""Subprocess runner for TDD Robot Framework fixture files.

Runs ``.robot`` fixture files with the ``tdd_expected_fail_listener``
active and inspects the resulting ``output.xml`` to determine the final
test status after listener processing.

Extracted from ``helper_tdd_tag_validation.py`` to keep file lengths
under the 500-line project limit.
"""

from __future__ import annotations

__all__: list[str] = []

import subprocess
import sys
import tempfile

# Robot Framework generates the output XML; we parse it to inspect test
# results.  The XML is always self-generated (never from untrusted input),
# so the standard library parser is sufficient — no XXE risk.
import xml.etree.ElementTree as ET
from pathlib import Path

_ROBOT_DIR = Path(__file__).resolve().parent
_LISTENER = str(_ROBOT_DIR / "tdd_expected_fail_listener.py")
_FIXTURES = _ROBOT_DIR / "fixtures"

_SUBPROCESS_TIMEOUT = 120  # seconds — matches project convention


def _extract_message(
    test_el: ET.Element,
    status_el: ET.Element,
) -> str:
    """Extract the test message from an RF output XML status element.

    RF may place the message as element text, a ``message`` attribute
    (RF 7.3+), or in a child ``<msg>`` element.  Check all three.
    """
    message = status_el.text or ""
    if not message:
        message = status_el.get("message", "")
    if not message:
        msg_el = test_el.find(".//msg")
        if msg_el is not None and msg_el.text:
            message = msg_el.text
    return message


def run_fixture(fixture_name: str) -> tuple[str, str]:
    """Run a fixture ``.robot`` file and return ``(status, message)``.

    Returns the status and message of the *first* test case found in
    the Robot output XML.
    """
    fixture_path = _FIXTURES / f"{fixture_name}.robot"
    if not fixture_path.exists():
        print(f"ERROR: fixture not found: {fixture_path}", file=sys.stderr)
        sys.exit(1)

    with tempfile.TemporaryDirectory() as tmpdir:
        out_xml = str(Path(tmpdir) / "output.xml")
        cmd = [
            sys.executable,
            "-m",
            "robot",
            "--listener",
            _LISTENER,
            "--outputdir",
            tmpdir,
            "--loglevel",
            "INFO",
            "--report",
            "NONE",
            "--log",
            "NONE",
            str(fixture_path),
        ]
        try:
            proc = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=_SUBPROCESS_TIMEOUT,
            )
        except subprocess.TimeoutExpired:
            print(
                f"ERROR: fixture {fixture_name!r} timed out after "
                f"{_SUBPROCESS_TIMEOUT}s",
                file=sys.stderr,
            )
            sys.exit(1)

        if not Path(out_xml).exists():
            print(
                f"ERROR: Robot did not produce output.xml (rc={proc.returncode})",
                file=sys.stderr,
            )
            print(f"stderr: {proc.stderr}", file=sys.stderr)
            sys.exit(1)

        tree = ET.parse(out_xml)
        root = tree.getroot()
        test_el = root.find(".//test")
        if test_el is None:
            return "ERROR", "No test found in output XML"
        status_el = test_el.find("status")
        if status_el is None:
            return "ERROR", "No status element found"
        status = status_el.get("status", "UNKNOWN")
        message = _extract_message(test_el, status_el)
        return status, message


def run_multi_fixture(
    *fixture_names: str,
) -> dict[str, tuple[str, str]]:
    """Run multiple fixture ``.robot`` files in a single Robot invocation.

    Returns a dict mapping test name to ``(status, message)``.  Running
    multiple fixtures together proves the listener is loaded and
    selectively applies (normal tests are unaffected *in the same run*
    as ``tdd_expected_fail`` tests).
    """
    fixture_paths = []
    for name in fixture_names:
        fp = _FIXTURES / f"{name}.robot"
        if not fp.exists():
            print(f"ERROR: fixture not found: {fp}", file=sys.stderr)
            sys.exit(1)
        fixture_paths.append(str(fp))

    with tempfile.TemporaryDirectory() as tmpdir:
        out_xml = str(Path(tmpdir) / "output.xml")
        cmd = [
            sys.executable,
            "-m",
            "robot",
            "--listener",
            _LISTENER,
            "--outputdir",
            tmpdir,
            "--loglevel",
            "INFO",
            "--report",
            "NONE",
            "--log",
            "NONE",
            *fixture_paths,
        ]
        try:
            proc = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=_SUBPROCESS_TIMEOUT,
            )
        except subprocess.TimeoutExpired:
            print(
                f"ERROR: multi-fixture run timed out after {_SUBPROCESS_TIMEOUT}s",
                file=sys.stderr,
            )
            sys.exit(1)

        if not Path(out_xml).exists():
            print(
                f"ERROR: Robot did not produce output.xml (rc={proc.returncode})",
                file=sys.stderr,
            )
            print(f"stderr: {proc.stderr}", file=sys.stderr)
            sys.exit(1)

        tree = ET.parse(out_xml)
        root = tree.getroot()
        results: dict[str, tuple[str, str]] = {}
        for test_el in root.findall(".//test"):
            test_name = test_el.get("name", "UNKNOWN")
            status_el = test_el.find("status")
            if status_el is None:
                results[test_name] = ("ERROR", "No status element found")
                continue
            status = status_el.get("status", "UNKNOWN")
            message = _extract_message(test_el, status_el)
            results[test_name] = (status, message)
        return results