"""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 ```` 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