orchestrata_ultimate.harness.evaluators.trace

Reference › orchestrata_ultimate › harness › evaluators › trace

TraceEvaluator, registered as trace_validator (SPEC-102 US-005, REQ-102-066). Despite the name it does not check @traces annotations: it scans execution trace text for unhandled exceptions. The trace is the scenario output, or the contents of the file when the output is a path to an existing file. It looks for Traceback (most recent call last): blocks and standalone Exception: or Error: lines; any whose text matches a regular expression in the allowed_patterns param is treated as handled. Unhandled exceptions give a failure with score 0.0 and a list, otherwise the evaluator passes with 1.0. See the evaluator registry.

At a glance

Class diagram

Dependency diagram

API

Classes

TraceEvaluator

Bases: Evaluator · Decorators: register_evaluator, traces(spec='102', story='US-005', reqs=['REQ-102-066']) · Line 21

Parse execution traces for unhandled exceptions.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/harness/evaluators/trace.py

"""Trace evaluator for the Volk Test Harness."""

from __future__ import annotations

import re
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.harness.evaluators import register_evaluator
from orchestrata_ultimate.harness.models import (
    Evaluator,
    EvaluatorConfig,
    EvaluatorResult,
    HarnessContext,
)


@register_evaluator
@traces(spec="102", story="US-005", reqs=["REQ-102-066"])
class TraceEvaluator(Evaluator):
    """Parse execution traces for unhandled exceptions."""

    name = "trace_validator"

    @traces(spec="102", story="US-005", reqs=["REQ-102-066"])
    async def evaluate(
        self,
        output: Any,
        context: HarnessContext,
        config: EvaluatorConfig,
    ) -> EvaluatorResult:
        """Parse trace text for exception patterns.

        The evaluator looks for:

        - ``Traceback (most recent call last):`` blocks
        - Standalone ``Exception:`` or ``Error:`` lines

        Exceptions whose text matches any regex in
        ``config.params["allowed_patterns"]`` are treated as *handled* and do
        not cause a failure.
        """
        trace_text = self._resolve_trace(output)
        allowed_patterns: list[str] = config.params.get("allowed_patterns", [])

        unhandled = self._find_unhandled_exceptions(trace_text, allowed_patterns)

        if unhandled:
            return EvaluatorResult(
                name=self.name,
                passed=False,
                score=0.0,
                details="Unhandled exceptions found:\n"
                + "\n".join(f"  - {exc}" for exc in unhandled),
            )

        return EvaluatorResult(
            name=self.name,
            passed=True,
            score=1.0,
            details="No unhandled exceptions found.",
        )

    @traces(spec="102", story="US-005", reqs=["REQ-102-066"])
    def _resolve_trace(self, output: Any) -> str:
        """Resolve *output* to a trace text string.

        If *output* is a path to an existing file, its contents are read.
        Otherwise the string representation of *output* is used directly.
        """
        if isinstance(output, (str, Path)):
            path = Path(output)
            if path.exists() and path.is_file():
                return path.read_text(encoding="utf-8")
            return str(output)
        if output is None:
            return ""
        return str(output)

    @traces(spec="102", story="US-005", reqs=["REQ-102-066"])
    def _find_unhandled_exceptions(
        self,
        trace_text: str,
        allowed_patterns: list[str],
    ) -> list[str]:
        """Find unhandled exceptions in *trace_text*."""
        unhandled: list[str] = []
        lines = trace_text.splitlines()

        # Find traceback blocks and extract the exception line.
        i = 0
        while i < len(lines):
            if "Traceback (most recent call last):" in lines[i]:
                j = i + 1
                while j < len(lines):
                    if lines[j] and not lines[j].startswith(" "):
                        exc_line = lines[j].strip()
                        if not self._is_allowed(exc_line, allowed_patterns):
                            unhandled.append(exc_line)
                        break
                    j += 1
                i = j if j < len(lines) else i + 1
            i += 1

        # Find standalone Exception: / Error: lines.
        for line in lines:
            stripped = line.strip()
            if re.match(r"^(?:Exception|Error):\s*", stripped):
                if not self._is_allowed(stripped, allowed_patterns):
                    unhandled.append(stripped)

        return unhandled

    @traces(spec="102", story="US-005", reqs=["REQ-102-066"])
    def _is_allowed(self, exc_line: str, patterns: list[str]) -> bool:
        """Check whether *exc_line* matches any allowed regex pattern."""
        for pattern in patterns:
            if re.search(pattern, exc_line):
                return True
        return False