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_impl(
    fixture_name: str,
    *,
    extra_args: tuple[str, ...] = (),
) -> tuple[str, str]:
    """Execute a fixture and return its ``(status, message)`` tuple.

    Args:
        fixture_name: Name of the fixture file without the ``.robot`` suffix.
        extra_args: Additional command-line arguments passed to ``robot``.

    Returns:
        A tuple containing the Robot test status and message for the first
        test case in the output XML.
    """
    _validate_fixture_name(fixture_name)

    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,
            *extra_args,
            "--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)

        try:
            tree = ET.parse(out_xml)
        except ET.ParseError as exc:
            print(
                f"ERROR: Failed to parse output.xml for {fixture_name!r}: {exc}",
                file=sys.stderr,
            )
            print(f"stderr: {proc.stderr}", file=sys.stderr)
            return "ERROR", f"Malformed Robot output: {exc}"

        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 _validate_fixture_name(fixture_name: str) -> None:
    """Ensure the fixture name does not contain directory traversal tokens."""
    if (
        not fixture_name
        or any(token in fixture_name for token in ("/", "\\", ".."))
        or fixture_name.startswith(":")
    ):
        raise ValueError(f"Invalid fixture name: {fixture_name!r}")


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.
    """
    return _run_fixture_impl(fixture_name)


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

    Dry-run mode (``--dryrun``) causes Robot Framework to validate
    keyword calls without executing them.  Tests get status ``NOT RUN``
    (or ``FAIL`` if keyword resolution fails).  This function is used
    to verify the listener's dry-run guard.

    Returns the status and message of the *first* test case found in
    the Robot output XML.
    """
    return _run_fixture_impl(fixture_name, extra_args=("--dryrun",))


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:
        _validate_fixture_name(name)
        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)

        try:
            tree = ET.parse(out_xml)
        except ET.ParseError as exc:
            print(
                f"ERROR: Failed to parse output.xml for multi-fixture run: {exc}",
                file=sys.stderr,
            )
            print(f"stderr: {proc.stderr}", file=sys.stderr)
            return {
                "__multi_fixture_parse_error__": (
                    "ERROR",
                    f"Malformed Robot output: {exc}",
                )
            }

        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