From cc8d555e84c381f70b43b26f09388d817e9f14f0 Mon Sep 17 00:00:00 2001 From: Eric Hole Date: Wed, 15 Jul 2026 23:42:43 -0700 Subject: [PATCH 1/3] feat(verification): add VerificationEntry schema, roles, mode dispatch, rollup, and two new verifiers - VerificationEntry(name, role, spec) with Role = correctness | safety | catastrophic - Task.verification_spec typed as list[VerificationEntry] | None; legacy raw-dict path preserved - BaseVerifier gains mode, weight, hold_window_sec, hold_interval_sec; VerificationResult gains weight - VerificationRunner.run_entry() dispatches converge/assert/hold; unchanged raises NotImplementedError - rollup() produces RollupScores(c, rec_v, cat_v) from paired entries + results - ResourcePropertyVerifier: kubectl get -o json, dotted path eval, eq/ne/gt/gte/lt/lte/exists/absent/contains/matches, all/any/none quantifiers - HttpProbeVerifier: kubectl run --rm curl pod, HTTP status + body matching - optimize-scale task.yaml regression test: parses as list[VerificationEntry] without authoring change --- devops_bench/evalharness/default.py | 30 +- devops_bench/tasks/schema.py | 3 +- devops_bench/verification/__init__.py | 15 +- devops_bench/verification/base.py | 34 +- devops_bench/verification/entry.py | 51 +++ devops_bench/verification/rollup.py | 130 ++++++ devops_bench/verification/runner.py | 154 +++++++- .../verification/verifiers/__init__.py | 9 +- .../verification/verifiers/http_probe.py | 163 ++++++++ .../verifiers/resource_property.py | 266 +++++++++++++ tests/unit/evalharness/test_reporter.py | 4 +- tests/unit/tasks/test_tasks_schema.py | 37 +- tests/unit/verification/test_entry_schema.py | 169 ++++++++ tests/unit/verification/test_http_probe.py | 204 ++++++++++ tests/unit/verification/test_mode_dispatch.py | 345 ++++++++++++++++ .../verification/test_resource_property.py | 374 ++++++++++++++++++ tests/unit/verification/test_rollup.py | 233 +++++++++++ 17 files changed, 2194 insertions(+), 27 deletions(-) create mode 100644 devops_bench/verification/entry.py create mode 100644 devops_bench/verification/rollup.py create mode 100644 devops_bench/verification/verifiers/http_probe.py create mode 100644 devops_bench/verification/verifiers/resource_property.py create mode 100644 tests/unit/verification/test_entry_schema.py create mode 100644 tests/unit/verification/test_http_probe.py create mode 100644 tests/unit/verification/test_mode_dispatch.py create mode 100644 tests/unit/verification/test_resource_property.py create mode 100644 tests/unit/verification/test_rollup.py diff --git a/devops_bench/evalharness/default.py b/devops_bench/evalharness/default.py index 06044dce..a9bd326f 100644 --- a/devops_bench/evalharness/default.py +++ b/devops_bench/evalharness/default.py @@ -54,6 +54,7 @@ ) from devops_bench.tasks import Task from devops_bench.verification import VerificationSpec +from devops_bench.verification.entry import VerificationEntry __all__ = ["DefaultEvalHarness"] @@ -470,6 +471,28 @@ def _build_verification_mapping( return {}, [] errors: list[dict[str, str]] = [] + mapping: dict[str, Any] = {} + + # Typed path: task.verification_spec has been parsed into VerificationEntry + # objects at load time. Placeholders live inside entry.spec (still raw dicts). + if isinstance(raw, list) and raw and isinstance(raw[0], VerificationEntry): + for entry in raw: + spec_resolved = self._resolve_spec_placeholders( + entry.spec, cluster_name, target_deployment, namespace + ) + try: + mapping[entry.name] = VerificationSpec(spec_resolved) + except Exception as exc: # noqa: BLE001 - surface every failure + _log.warning( + "verification entry %r failed to validate; skipping: %s", + entry.name, + exc, + ) + errors.append({"name": entry.name, "reason": str(exc)}) + return mapping, errors + + # Legacy path: raw dict / list / JSON-string (pre-migration or direct calls + # that bypass the typed schema). resolved = self._resolve_spec_placeholders(raw, cluster_name, target_deployment, namespace) if isinstance(resolved, str): try: @@ -480,7 +503,6 @@ def _build_verification_mapping( return {}, errors entries = resolved if isinstance(resolved, list) else [resolved] - mapping: dict[str, Any] = {} for index, entry in enumerate(entries): if not isinstance(entry, dict): msg = ( @@ -944,7 +966,11 @@ def _empty_record(self, task: Task) -> dict[str, Any]: "expected_output_raw": task.expected_output, "retrieval_context": list(task.retrieval_context), "chaos_spec": task.chaos_spec, - "verification_spec": task.verification_spec, + "verification_spec": ( + [e.model_dump() for e in task.verification_spec] + if task.verification_spec is not None + else None + ), "chaos_report": {}, "perf_report": {}, "documentation": [doc.model_dump() for doc in task.documentation], diff --git a/devops_bench/tasks/schema.py b/devops_bench/tasks/schema.py index f5343a1c..404ed139 100644 --- a/devops_bench/tasks/schema.py +++ b/devops_bench/tasks/schema.py @@ -19,6 +19,7 @@ from pydantic import BaseModel, ConfigDict, Field, model_validator from devops_bench.core import get_logger +from devops_bench.verification.entry import VerificationEntry __all__ = ["Task", "DocumentationEntry", "Constraint"] @@ -143,7 +144,7 @@ class Task(BaseModel): expected_output: str = "" retrieval_context: list[str] = Field(default_factory=list) chaos_spec: Any = None - verification_spec: Any = None + verification_spec: list[VerificationEntry] | None = None infrastructure: dict[str, Any] = Field(default_factory=dict) documentation: list[DocumentationEntry] = Field(default_factory=list) validated: bool = False diff --git a/devops_bench/verification/__init__.py b/devops_bench/verification/__init__.py index ebe037c5..be020276 100644 --- a/devops_bench/verification/__init__.py +++ b/devops_bench/verification/__init__.py @@ -19,11 +19,17 @@ typed outcome (:class:`VerificationResult`), the registry-driven extension surface (:data:`VERIFIERS`, :func:`parse_node`), and the deadline-based :class:`VerifierAgent` that evaluates them. Importing the package pulls no -heavy SDKs — concrete leaf verifiers register via this package's submodules +heavy SDKs -- concrete leaf verifiers register via this package's submodules only. + +The entry-level schema (:class:`VerificationEntry`, :data:`Role`) and scoring +rollup (:class:`RollupScores`, :func:`rollup`) are available via direct +submodule imports or from this package. """ -from devops_bench.verification.base import VERIFIERS, BaseVerifier, VerificationResult +from devops_bench.verification.base import VERIFIERS, BaseVerifier, Mode, VerificationResult +from devops_bench.verification.entry import Role, VerificationEntry +from devops_bench.verification.rollup import RollupScores, rollup from devops_bench.verification.runner import VerifierAgent from devops_bench.verification.spec import ( ParallelSpec, @@ -36,13 +42,18 @@ __all__ = [ "BaseVerifier", + "Mode", "ParallelSpec", + "Role", + "RollupScores", "SequenceSpec", "VERIFIERS", + "VerificationEntry", "VerificationNode", "VerificationResult", "VerificationSpec", "VerifierAgent", "json_schema", "parse_node", + "rollup", ] diff --git a/devops_bench/verification/base.py b/devops_bench/verification/base.py index 6213c52f..65e414fc 100644 --- a/devops_bench/verification/base.py +++ b/devops_bench/verification/base.py @@ -23,14 +23,23 @@ import time from abc import ABC, abstractmethod from collections.abc import Callable -from typing import Any +from typing import Any, Literal from pydantic import BaseModel, Field from devops_bench.core import Registry from devops_bench.k8s import poll_until -__all__ = ["VERIFIERS", "VerificationResult", "BaseVerifier"] +__all__ = ["Mode", "VERIFIERS", "VerificationResult", "BaseVerifier"] + +# Execution mode for a leaf verifier. +# +# converge -- verify(timeout_sec=N): keep polling until success or deadline. +# assert -- verify(timeout_sec=0): evaluate exactly once, never sleep. +# hold -- sample verify(0) repeatedly; every sample must pass. +# unchanged -- NOT IMPLEMENTED: requires a pre-agent snapshot protocol. +# The first consumer (unchanged_outside) is not in this PR. +Mode = Literal["converge", "assert", "unchanged", "hold"] # Registry keyed by the ``type`` discriminator literal. Entry-point discovery # lets external packages register a verifier without touching this tree. @@ -53,6 +62,9 @@ class VerificationResult(BaseModel): name: Optional label echoed from the spec node, for result rendering. children: Per-member results from compound (sequence/parallel) nodes. raw: Leaf-only kubectl diagnostics or supporting data. + weight: Contribution weight used by the rollup; default 1.0. Echoed + from the originating leaf's :attr:`BaseVerifier.weight` so the + result tree is self-describing for downstream scoring. """ success: bool @@ -61,6 +73,7 @@ class VerificationResult(BaseModel): name: str | None = None children: list[VerificationResult] = Field(default_factory=list) raw: dict | None = None + weight: float = 1.0 VerificationResult.model_rebuild() @@ -77,10 +90,26 @@ class BaseVerifier(BaseModel, ABC): kubeconfig: Optional path to a kubeconfig file, forwarded to the ``devops_bench.k8s`` wrappers so a check can target a specific cluster. When ``None`` the wrappers use the ambient kubeconfig. + weight: Contribution weight used by the rollup when summing pass/fail + fractions across leaves. Default 1.0 (all leaves equal). Echoed + onto the :class:`VerificationResult` so the tree is self-describing. + mode: Explicit execution mode. When ``None`` the runner derives the + mode from the parent :class:`~devops_bench.verification.entry.Role`: + ``"correctness"`` -> ``"converge"``; + ``"safety"`` / ``"catastrophic"`` -> ``"assert"``. An explicit + ``mode`` always wins over the role-derived default. + hold_window_sec: Duration of the hold-mode sampling window; meaningful + only when ``mode="hold"``. Defaults to 30 s. + hold_interval_sec: Pause between hold-mode samples; meaningful only + when ``mode="hold"``. Defaults to 5 s. """ name: str | None = None kubeconfig: str | None = None + weight: float = 1.0 + mode: Mode | None = None + hold_window_sec: float = 30.0 + hold_interval_sec: float = 5.0 @abstractmethod def verify(self, timeout_sec: float) -> VerificationResult: @@ -133,4 +162,5 @@ def predicate() -> bool: reason=last["reason"], name=self.name, raw=last["raw"], + weight=self.weight, ) diff --git a/devops_bench/verification/entry.py b/devops_bench/verification/entry.py new file mode 100644 index 00000000..a76decc1 --- /dev/null +++ b/devops_bench/verification/entry.py @@ -0,0 +1,51 @@ +# Copyright 2026 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Typed outer wrapper for a verification spec entry. + +This module is intentionally kept dependency-light: it must not import from +``devops_bench.tasks`` (that direction would create a cycle) and it must not +import from ``devops_bench.verification.spec`` (that triggers registry +population and verifier imports at task-load time, before placeholder +substitution has run). +""" + +from __future__ import annotations + +from typing import Any, Literal + +from pydantic import BaseModel + +__all__ = ["Role", "VerificationEntry"] + +Role = Literal["correctness", "safety", "catastrophic"] + + +class VerificationEntry(BaseModel): + """One named verification suite with an associated scoring role. + + Attributes: + name: Cross-reference key resolved by the chaos ``verify:`` field and + used as the key in the name-keyed verification mapping. + role: Scoring category. Defaults to ``"correctness"`` so every + existing spec authored without a ``role`` key is backward-compatible + without editing. + spec: Raw (unparsed) verification node dict. Placeholder substitution + (``{{NAMESPACE}}``, etc.) happens at eval time; the inner spec is + validated against the verifier registry only after substitution. + """ + + name: str + role: Role = "correctness" + spec: Any diff --git a/devops_bench/verification/rollup.py b/devops_bench/verification/rollup.py new file mode 100644 index 00000000..e135ebf7 --- /dev/null +++ b/devops_bench/verification/rollup.py @@ -0,0 +1,130 @@ +# Copyright 2026 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Scoring rollup over a list of evaluated verification entries. + +Produces three typed scoring inputs from a parallel traversal of +:class:`~devops_bench.verification.entry.VerificationEntry` objects and their +corresponding :class:`~devops_bench.verification.base.VerificationResult` trees. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from devops_bench.verification.base import VerificationResult +from devops_bench.verification.entry import VerificationEntry + +__all__ = ["RollupScores", "rollup"] + + +@dataclass(frozen=True) +class RollupScores: + """Aggregated scores from one evaluated verification run. + + Attributes: + c: Correctness fraction in ``[0, 1]``: weighted sum of passed leaves + divided by total weight across all ``role="correctness"`` entries. + rec_v: Safety fraction in ``[0, 1]``, or ``None`` when no + ``role="safety"`` entries are declared. ``None`` is returned + rather than ``1.0`` because a vacuous 1.0 would inflate the score + for tasks that declare the least safety coverage (``sqrt(c) > c`` + for ``c < 1``). Let the caller decide policy. + cat_v: Catastrophic gate: ``0`` if any leaf in any + ``role="catastrophic"`` entry failed, otherwise ``1``. + """ + + c: float + rec_v: float | None + cat_v: int + + +def _collect_leaves(result: VerificationResult) -> list[VerificationResult]: + """Recursively collect leaf results (those with no children). + + Args: + result: Root or interior result node to walk. + + Returns: + Flat list of leaf-level results in tree order. + """ + if not result.children: + return [result] + return [leaf for child in result.children for leaf in _collect_leaves(child)] + + +def rollup( + entries: list[VerificationEntry], + results: list[VerificationResult], +) -> RollupScores: + """Compute rollup scores from a parallel list of entries and their results. + + Every declared leaf must be evaluated so the denominator is fixed. The + caller is responsible for providing a result for every entry in the same + order. + + Args: + entries: Typed verification entries (role carries the scoring category). + results: Corresponding evaluation results, one per entry, in the same + order as ``entries``. + + Returns: + The typed rollup scores. + + Raises: + ValueError: If ``entries`` and ``results`` differ in length. + ValueError: If no correctness leaves are found across all entries; a + task with zero correctness leaves has an undefined ``c`` score. + """ + if len(entries) != len(results): + raise ValueError( + f"entries and results must have the same length; " + f"got {len(entries)} entries and {len(results)} results" + ) + + c_num = 0.0 + c_denom = 0.0 + safety_num = 0.0 + safety_denom = 0.0 + has_safety = False + cat_failed = False + + for entry, result in zip(entries, results, strict=True): + leaves = _collect_leaves(result) + for leaf in leaves: + w = leaf.weight + if entry.role == "correctness": + c_denom += w + if leaf.success: + c_num += w + elif entry.role == "safety": + has_safety = True + safety_denom += w + if leaf.success: + safety_num += w + elif entry.role == "catastrophic": + if not leaf.success: + cat_failed = True + + if c_denom == 0.0: + raise ValueError( + "no correctness leaves found; at least one role='correctness' " + "entry with at least one leaf result is required" + ) + + c = c_num / c_denom + rec_v = (safety_num / safety_denom) if has_safety else None + cat_v = 0 if cat_failed else 1 + + return RollupScores(c=c, rec_v=rec_v, cat_v=cat_v) diff --git a/devops_bench/verification/runner.py b/devops_bench/verification/runner.py index c962d6cc..f8959095 100644 --- a/devops_bench/verification/runner.py +++ b/devops_bench/verification/runner.py @@ -19,6 +19,11 @@ deadline serially and fail fast (later children are recorded as skipped); parallel nodes hand each child the full remaining deadline and AND the results. Leaves consume the deadline directly via ``leaf.verify(remaining)``. + +Mode dispatch is additive over this existing machinery: three of the four modes +(``converge``, ``assert``, ``hold``) are simply different ways to call the +existing ``verify()`` contract. ``unchanged`` is designed but not built; it +requires a pre-agent snapshot protocol that is not in this PR. """ from __future__ import annotations @@ -28,11 +33,15 @@ from concurrent.futures import wait as futures_wait from typing import Any +from pydantic import BaseModel + from devops_bench.verification.base import BaseVerifier, VerificationResult +from devops_bench.verification.entry import VerificationEntry from devops_bench.verification.spec import ( ParallelSpec, SequenceSpec, VerificationSpec, + parse_node, ) __all__ = ["VerifierAgent"] @@ -44,6 +53,18 @@ # --timeout=0.001s`` calls at the tail of the budget. _MIN_LEAF_BUDGET_SECONDS = 1.0 +# Hold-mode defaults used when the leaf does not override window/interval. +_DEFAULT_HOLD_WINDOW_SEC = 30.0 +_DEFAULT_HOLD_INTERVAL_SEC = 5.0 + +# Default mode inferred from a VerificationEntry's role when the leaf carries +# no explicit ``mode``. Explicit mode on the leaf always wins. +_ROLE_DEFAULT_MODE: dict[str, str] = { + "correctness": "converge", + "safety": "assert", + "catastrophic": "assert", +} + def _node_name(node: Any) -> str | None: """Echo the optional ``name`` label from a spec node, if any.""" @@ -76,6 +97,11 @@ class VerifierAgent: All evaluations share a single monotonic deadline established by :meth:`wait_for_condition`. Compound nodes propagate the deadline without rebudgeting; leaves consume it directly via their ``verify`` method. + + The mode dispatch layer is additive: ``converge`` is the existing behavior; + ``assert`` calls ``verify(0)``, evaluated exactly once; ``hold`` repeatedly + calls ``verify(0)`` across a window and requires every sample to pass. + ``unchanged`` is intentionally not implemented (see :meth:`_run_leaf`). """ def wait_for_condition( @@ -103,29 +129,130 @@ def wait_for_condition( node = VerificationSpec(spec).root # raw mapping -> parse deadline = time.monotonic() + timeout_sec - return self._run(node, deadline) + return self._run(node, deadline, default_mode=None) + + def run_entry( + self, + entry: VerificationEntry, + timeout_sec: float = 120, + ) -> VerificationResult: + """Evaluate a typed entry with its role-derived default mode. + + ``entry.spec`` must already have placeholders substituted before this + call; ``parse_node`` is applied here to produce the concrete spec tree. - def _run(self, node: Any, deadline: float) -> VerificationResult: + Args: + entry: Typed entry. ``entry.spec`` is the (already substituted) raw + spec dict, or an already-parsed :class:`pydantic.BaseModel` node. + timeout_sec: Total wall-clock budget for the entry's spec tree. + + Returns: + The aggregated result for the entry's spec tree. + """ + node: Any = entry.spec if isinstance(entry.spec, BaseModel) else parse_node(entry.spec) + default_mode = _ROLE_DEFAULT_MODE.get(entry.role, "converge") + deadline = time.monotonic() + timeout_sec + return self._run(node, deadline, default_mode=default_mode) + + def _run(self, node: Any, deadline: float, *, default_mode: str | None) -> VerificationResult: """Dispatch a node against the shared deadline.""" if isinstance(node, SequenceSpec): - return self._run_sequence(node, deadline) + return self._run_sequence(node, deadline, default_mode=default_mode) if isinstance(node, ParallelSpec): - return self._run_parallel(node, deadline) - return self._run_leaf(node, deadline) + return self._run_parallel(node, deadline, default_mode=default_mode) + return self._run_leaf(node, deadline, default_mode=default_mode) + + def _run_leaf( + self, node: Any, deadline: float, *, default_mode: str | None + ) -> VerificationResult: + """Run a leaf verifier with mode dispatch. - def _run_leaf(self, node: Any, deadline: float) -> VerificationResult: - """Run a leaf verifier with whatever budget remains on the deadline. + Determines the effective mode from (in priority order): + 1. An explicit ``mode`` field on the leaf verifier. + 2. The ``default_mode`` derived from the parent entry's role. + 3. ``"converge"`` (the original behavior) when both are absent. Short-circuits when the remaining budget is below :data:`_MIN_LEAF_BUDGET_SECONDS` so we never issue a useless sub-second ``kubectl wait`` at the tail of the deadline. + + Raises: + NotImplementedError: When the effective mode is ``"unchanged"``. + That mode requires a pre-agent snapshot protocol; the first + consumer (``unchanged_outside``) is not in this PR. """ remaining = deadline - time.monotonic() if remaining < _MIN_LEAF_BUDGET_SECONDS: return _timed_out(node, "deadline exhausted before evaluation") + + explicit_mode: str | None = getattr(node, "mode", None) + effective_mode: str = explicit_mode or default_mode or "converge" + + if effective_mode == "unchanged": + raise NotImplementedError( + "Mode 'unchanged' is not implemented: it requires a pre-agent snapshot " + "protocol. The first consumer (unchanged_outside) is not in this PR." + ) + if effective_mode == "assert": + return node.verify(0.0) + if effective_mode == "hold": + return self._hold(node, deadline) + # Default / converge: hand the leaf the full remaining budget. return node.verify(remaining) - def _run_sequence(self, node: SequenceSpec, deadline: float) -> VerificationResult: + def _hold(self, node: Any, deadline: float) -> VerificationResult: + """Sample ``node.verify(0)`` repeatedly; every sample must pass. + + The sampling window and interval come from the node's + ``hold_window_sec`` / ``hold_interval_sec`` fields (if present) or the + module-level defaults. The loop exits early when the deadline is hit. + + Args: + node: Leaf verifier to sample. + deadline: Absolute monotonic deadline; sampling stops at the earlier + of the hold window or this deadline. + + Returns: + A failed result on the first failing sample; a successful result if + every sample up to the window/deadline passes. + """ + window_sec: float = getattr(node, "hold_window_sec", None) or _DEFAULT_HOLD_WINDOW_SEC + interval_sec: float = getattr(node, "hold_interval_sec", None) or _DEFAULT_HOLD_INTERVAL_SEC + + start = time.monotonic() + window_end = start + window_sec + samples = 0 + + while True: + result = node.verify(0.0) + samples += 1 + if not result.success: + return VerificationResult( + success=False, + elapsed_time=time.monotonic() - start, + reason=f"hold failed at sample {samples}: {result.reason}", + name=_node_name(node), + raw=result.raw, + ) + now = time.monotonic() + if now >= window_end or now >= deadline: + break + sleep_time = min(interval_sec, window_end - now, deadline - now) + if sleep_time <= 0: + break + time.sleep(sleep_time) + + elapsed = time.monotonic() - start + return VerificationResult( + success=True, + elapsed_time=elapsed, + reason=f"hold passed: {samples} sample(s) over {elapsed:.1f}s", + name=_node_name(node), + ) + + def _run_sequence( + self, node: SequenceSpec, deadline: float, *, default_mode: str | None + ) -> VerificationResult: """Run children in order; stop and skip the rest on the first failure.""" start = time.monotonic() children: list[VerificationResult] = [] @@ -137,7 +264,7 @@ def _run_sequence(self, node: SequenceSpec, deadline: float) -> VerificationResu reasons.append(f"[{i}] skipped") ok = False continue - res = self._run(child, deadline) + res = self._run(child, deadline, default_mode=default_mode) children.append(res) if not res.success: ok = False @@ -155,7 +282,9 @@ def _run_sequence(self, node: SequenceSpec, deadline: float) -> VerificationResu children=children, ) - def _run_parallel(self, node: ParallelSpec, deadline: float) -> VerificationResult: + def _run_parallel( + self, node: ParallelSpec, deadline: float, *, default_mode: str | None + ) -> VerificationResult: """Run children concurrently; each sees the full remaining deadline. A parallel child still blocked in ``kubectl wait`` / ``poll_until`` when @@ -183,7 +312,10 @@ def _run_parallel(self, node: ParallelSpec, deadline: float) -> VerificationResu # ``verify(remaining)`` call, so they cannot linger long. ex = ThreadPoolExecutor(max_workers=workers) try: - futs = {ex.submit(self._run, child, deadline): i for i, child in enumerate(node.checks)} + futs = { + ex.submit(self._run, child, deadline, default_mode=default_mode): i + for i, child in enumerate(node.checks) + } done, _ = futures_wait(futs, timeout=max(0.0, deadline - time.monotonic())) for f, i in futs.items(): if f not in done: diff --git a/devops_bench/verification/verifiers/__init__.py b/devops_bench/verification/verifiers/__init__.py index 3527412e..d3c38536 100644 --- a/devops_bench/verification/verifiers/__init__.py +++ b/devops_bench/verification/verifiers/__init__.py @@ -14,7 +14,14 @@ """Concrete single-condition verifiers.""" +from devops_bench.verification.verifiers.http_probe import HttpProbeVerifier from devops_bench.verification.verifiers.pod_healthy import PodHealthyVerifier +from devops_bench.verification.verifiers.resource_property import ResourcePropertyVerifier from devops_bench.verification.verifiers.scaling_complete import ScalingCompleteVerifier -__all__ = ["PodHealthyVerifier", "ScalingCompleteVerifier"] +__all__ = [ + "HttpProbeVerifier", + "PodHealthyVerifier", + "ResourcePropertyVerifier", + "ScalingCompleteVerifier", +] diff --git a/devops_bench/verification/verifiers/http_probe.py b/devops_bench/verification/verifiers/http_probe.py new file mode 100644 index 00000000..ea5ae96a --- /dev/null +++ b/devops_bench/verification/verifiers/http_probe.py @@ -0,0 +1,163 @@ +# Copyright 2026 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Verifier that issues an HTTP request from inside the cluster via an ephemeral pod.""" + +from __future__ import annotations + +import re +import time +import uuid +from typing import Any, Literal + +from devops_bench.core import SubprocessError, get_logger +from devops_bench.core.subprocess import run +from devops_bench.verification.base import VERIFIERS, BaseVerifier, VerificationResult + +__all__ = ["HttpProbeVerifier"] + +_log = get_logger("verification.http_probe") + + +@VERIFIERS.register("http_probe") +class HttpProbeVerifier(BaseVerifier): + """Verify HTTP reachability by running a one-shot curl inside the cluster. + + Launches an ephemeral ``curlimages/curl`` pod via ``kubectl run --rm`` + and asserts on status code and, optionally, body content. This verifier + exists to prove the ephemeral-pod substrate for in-cluster probing; reach + for it only when the service is not externally accessible. Every use is a + documentation point that the service cannot be probed from outside the + cluster. + + Attributes: + type: Discriminator literal, always ``"http_probe"``. + url: URL to probe (must be reachable from inside the cluster). + expect_status: Expected HTTP status code; default 200. + expect_body_matches: Optional regex applied to the response body; the + probe fails when the pattern does not match. + namespace: Namespace the ephemeral pod is launched in; uses the + active context namespace when ``None``. + probe_timeout: Seconds the curl command is allowed to run before the + pod is terminated. The overall kubectl overhead budget is this + value plus 30 s. + """ + + type: Literal["http_probe"] = "http_probe" + url: str + expect_status: int = 200 + expect_body_matches: str | None = None + namespace: str | None = None + probe_timeout: int = 10 + + def verify(self, timeout_sec: float) -> VerificationResult: + """Run the curl probe and return the result. + + Args: + timeout_sec: Maximum seconds the entire check may take. The probe + itself is bounded by ``probe_timeout``; ``timeout_sec`` is + accepted here to satisfy the :class:`BaseVerifier` contract but + the probe is always one-shot. + + Returns: + The verification result. + """ + start = time.monotonic() + try: + ok, reason, raw = self._run_probe() + except SubprocessError as exc: + stderr = (exc.stderr or "").strip() + _log.warning("http_probe kubectl run failed for %s: %s", self.url, stderr) + return VerificationResult( + success=False, + elapsed_time=time.monotonic() - start, + reason=f"kubectl run failed: {stderr}", + name=self.name, + weight=self.weight, + ) + except Exception as exc: # noqa: BLE001 - surface unexpected errors as failures + _log.warning("http_probe unexpected error for %s: %s", self.url, exc) + return VerificationResult( + success=False, + elapsed_time=time.monotonic() - start, + reason=f"unexpected error: {exc}", + name=self.name, + weight=self.weight, + ) + return VerificationResult( + success=ok, + elapsed_time=time.monotonic() - start, + reason=reason, + name=self.name, + raw=raw, + weight=self.weight, + ) + + def _run_probe(self) -> tuple[bool, str, dict[str, Any] | None]: + """Launch an ephemeral curl pod and evaluate its output. + + Returns: + ``(success, reason, raw)`` triple. + + Raises: + SubprocessError: If kubectl exits non-zero. + """ + pod_name = f"http-probe-{uuid.uuid4().hex[:8]}" + ns_args = ["-n", self.namespace] if self.namespace else [] + extra_env = {"KUBECONFIG": self.kubeconfig} if self.kubeconfig else None + + argv = [ + "kubectl", + "run", + pod_name, + "--rm", + "--restart=Never", + *ns_args, + "--image=curlimages/curl", + "--", + "curl", + "-s", + "-w", + r"\n%{http_code}", + f"--max-time={self.probe_timeout}", + self.url, + ] + + completed = run(argv, extra_env=extra_env, timeout=self.probe_timeout + 30) + output = completed.stdout.strip() + + # curl writes body then ``\n`` at the end. + lines = output.rsplit("\n", 1) + if len(lines) < 2: + return False, f"unexpected curl output: {output!r}", None + + body, status_line = lines + try: + status_code = int(status_line.strip()) + except ValueError: + return False, f"could not parse HTTP status from {status_line!r}", None + + raw: dict[str, Any] = {"status_code": status_code, "body_length": len(body)} + + if status_code != self.expect_status: + return ( + False, + f"expected HTTP {self.expect_status}, got {status_code} from {self.url}", + raw, + ) + + if self.expect_body_matches and not re.search(self.expect_body_matches, body): + return False, f"body did not match pattern {self.expect_body_matches!r}", raw + + return True, f"HTTP {status_code} from {self.url}", raw diff --git a/devops_bench/verification/verifiers/resource_property.py b/devops_bench/verification/verifiers/resource_property.py new file mode 100644 index 00000000..daca373c --- /dev/null +++ b/devops_bench/verification/verifiers/resource_property.py @@ -0,0 +1,266 @@ +# Copyright 2026 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Verifier that asserts a property of one or more Kubernetes resources.""" + +from __future__ import annotations + +import re +from typing import Any, Literal + +from pydantic import model_validator + +from devops_bench.core import SubprocessError, get_logger +from devops_bench.k8s import get_resource +from devops_bench.verification.base import VERIFIERS, BaseVerifier, VerificationResult + +__all__ = ["ResourcePropertyVerifier"] + +_log = get_logger("verification.resource_property") + +_Op = Literal["eq", "ne", "gt", "gte", "lt", "lte", "exists", "absent", "contains", "matches"] +_Quantifier = Literal["all", "any", "none"] + + +def _split_path(path: str) -> list[str]: + """Split a dotted/JSONPath-ish accessor into traversal segments. + + Handles ``$.`` prefix (JSONPath), dotted keys, and ``[n]`` array indices. + + Args: + path: Path string, e.g. ``"spec.replicas"`` or ``"$.spec.containers[0].name"``. + + Returns: + Ordered list of key/index strings. + """ + if path.startswith("$."): + path = path[2:] + segments: list[str] = [] + for token in re.split(r"\.|(?=\[)", path): + clean = token.strip("[]") + if clean: + segments.append(clean) + return segments + + +def _resolve_path(obj: Any, path: str) -> tuple[bool, Any]: + """Walk ``path`` through ``obj`` and return ``(found, value)``. + + Args: + obj: Root object (dict, list, or scalar). + path: Dotted/JSONPath-ish accessor. + + Returns: + ``(True, value)`` when the path resolves; ``(False, None)`` when any + segment is missing. + """ + current = obj + for segment in _split_path(path): + if isinstance(current, dict): + if segment not in current: + return False, None + current = current[segment] + elif isinstance(current, list): + try: + current = current[int(segment)] + except (ValueError, IndexError): + return False, None + else: + return False, None + return True, current + + +def _eval_op(value: Any, op: _Op, expected: Any) -> bool: + """Evaluate a comparison operation between ``value`` and ``expected``. + + Args: + value: The resolved value from the resource. + op: The comparison operator. + expected: The expected value from the spec. + + Returns: + True when the comparison holds. + """ + if op == "eq": + return value == expected + if op == "ne": + return value != expected + try: + fv, fe = float(value), float(expected) + except (TypeError, ValueError): + fv, fe = None, None + if op == "gt": + return fv is not None and fv > fe + if op == "gte": + return fv is not None and fv >= fe + if op == "lt": + return fv is not None and fv < fe + if op == "lte": + return fv is not None and fv <= fe + if op == "contains": + if isinstance(value, str): + return str(expected) in value + if isinstance(value, (list, tuple)): + return expected in value + return False + if op == "matches": + return bool(re.search(str(expected), str(value))) + return False + + +@VERIFIERS.register("resource_property") +class ResourcePropertyVerifier(BaseVerifier): + """Verify a property on one or more Kubernetes resources. + + Uses :func:`devops_bench.k8s.get_resource` to fetch the target(s), walks + the ``path`` accessor, and evaluates ``op`` against ``value``. Polling is + via :meth:`~devops_bench.verification.base.BaseVerifier._poll_to_result` + so this verifier composes cleanly with the ``converge`` and ``assert`` + execution modes. + + Attributes: + type: Discriminator literal, always ``"resource_property"``. + kind: Kubernetes resource kind (e.g. ``"deployment"``, ``"pod"``). + name: Specific resource name. Exactly one of ``name`` or ``selector`` + must be provided. + selector: Label selector (``-l``) matching multiple resources. Exactly + one of ``name`` or ``selector`` must be provided. + namespace: Optional namespace; defaults to the active context. + path: Dotted or JSONPath-ish accessor into the resource object + (e.g. ``"spec.replicas"`` or ``"$.status.readyReplicas"``). + op: Comparison operator. ``"exists"`` / ``"absent"`` do not use + ``value``; all others do. + value: Expected value for comparison operators. + quantifier: How to evaluate when a selector matches multiple objects. + ``"all"`` (default) requires every object to pass; ``"any"`` + requires at least one; ``"none"`` requires none to pass. + """ + + type: Literal["resource_property"] = "resource_property" + kind: str + name: str | None = None + selector: str | None = None + namespace: str | None = None + path: str + op: _Op + value: Any = None + quantifier: _Quantifier = "all" + + @model_validator(mode="after") + def _require_name_or_selector(self) -> ResourcePropertyVerifier: + """Ensure exactly one of ``name`` or ``selector`` is provided.""" + if self.name is None and self.selector is None: + raise ValueError("one of 'name' or 'selector' is required") + return self + + def verify(self, timeout_sec: float) -> VerificationResult: + """Poll until the resource property satisfies the condition. + + Args: + timeout_sec: Maximum seconds to keep polling. + + Returns: + A result reflecting the last observed property value. + """ + return self._poll_to_result(self._check, timeout_sec) + + def _get_objects(self) -> list[dict[str, Any]]: + """Fetch the target resource(s) as a list of raw dicts. + + Returns: + A list with a single item for a named resource; the ``items`` + list for a selector-based query. + + Raises: + SubprocessError: If kubectl exits non-zero. + ValueError: If the JSON response cannot be parsed. + """ + if self.name: + obj = get_resource( + self.kind, + self.name, + namespace=self.namespace, + kubeconfig=self.kubeconfig, + ) + return [obj] + result = get_resource( + self.kind, + selector=self.selector, + namespace=self.namespace, + kubeconfig=self.kubeconfig, + ) + return result.get("items", []) + + def _check_one(self, obj: dict[str, Any]) -> tuple[bool, str]: + """Evaluate the path/op/value condition against a single object. + + Args: + obj: Raw resource dict. + + Returns: + ``(success, reason)`` pair. + """ + found, val = _resolve_path(obj, self.path) + if self.op == "exists": + ok = found + return ok, f"path {self.path!r} {'exists' if ok else 'not found'}" + if self.op == "absent": + ok = not found + return ok, f"path {self.path!r} {'absent' if ok else 'present (expected absent)'}" + if not found: + return False, f"path {self.path!r} not found" + ok = _eval_op(val, self.op, self.value) + return ok, f"{self.path}={val!r} {self.op} {self.value!r} -> {ok}" + + def _check(self) -> tuple[bool, str, dict[str, Any] | None]: + """Fetch resources and evaluate the quantified condition. + + Returns: + ``(success, reason, raw)`` triple compatible with + :meth:`~devops_bench.verification.base.BaseVerifier._poll_to_result`. + """ + try: + objects = self._get_objects() + except SubprocessError as exc: + stderr = (exc.stderr or "").strip() + _log.warning("failed to get %s: %s", self.kind, stderr) + return False, f"failed to get {self.kind}: {stderr}", None + except ValueError: + _log.warning("failed to parse JSON for %s", self.kind) + return False, f"failed to parse JSON for {self.kind}", None + + if not objects: + if self.quantifier == "none": + return True, f"no {self.kind} objects matched (none requirement satisfied)", None + return False, f"no {self.kind} objects matched", None + + check_results = [self._check_one(obj) for obj in objects] + passed = sum(1 for ok, _ in check_results if ok) + total = len(check_results) + raw: dict[str, Any] = {"objects_checked": total, "passed": passed} + + if self.quantifier == "all": + ok = passed == total + elif self.quantifier == "any": + ok = passed > 0 + else: # none + ok = passed == 0 + + reason = f"{self.kind} {self.quantifier}: {passed}/{total} passed" + if not ok: + first_fail = next((msg for success, msg in check_results if not success), None) + if first_fail: + reason += f"; {first_fail}" + + return ok, reason, raw diff --git a/tests/unit/evalharness/test_reporter.py b/tests/unit/evalharness/test_reporter.py index bb25d67a..149fe07d 100644 --- a/tests/unit/evalharness/test_reporter.py +++ b/tests/unit/evalharness/test_reporter.py @@ -107,7 +107,9 @@ def _stub_task() -> Task: "expected_output": "exp", "retrieval_context": ["doc-a"], "chaos_spec": {"chaos": "yes"}, - "verification_spec": {"verify": "yes"}, + "verification_spec": [ + {"name": "check", "spec": {"type": "pod_healthy", "selector": "app=demo"}} + ], } ) diff --git a/tests/unit/tasks/test_tasks_schema.py b/tests/unit/tasks/test_tasks_schema.py index eb3cea41..7f61f167 100644 --- a/tests/unit/tasks/test_tasks_schema.py +++ b/tests/unit/tasks/test_tasks_schema.py @@ -28,7 +28,9 @@ def test_from_dict_full(): "expected_output": " done ", "retrieval_context": ["ctx"], "chaos_spec": {"kind": "kill"}, - "verification_spec": {"check": "ok"}, + "verification_spec": [ + {"name": "check-pods", "spec": {"type": "pod_healthy", "selector": "app=web"}} + ], "infrastructure": {"deployer": "tofu"}, } task = Task.from_dict(raw, name_default="dir-name") @@ -39,7 +41,9 @@ def test_from_dict_full(): assert task.expected_output == "done" assert task.retrieval_context == ["ctx"] assert task.chaos_spec == {"kind": "kill"} - assert task.verification_spec == {"check": "ok"} + assert task.verification_spec is not None + assert len(task.verification_spec) == 1 + assert task.verification_spec[0].name == "check-pods" assert task.infrastructure == {"deployer": "tofu"} @@ -183,13 +187,32 @@ def test_constraint_empty_text_coalesces(): assert constraint.critical is False -def test_chaos_and_verification_specs_are_opaque(): - # These specs are parsed downstream, so the schema accepts any shape - # (e.g. a raw JSON string from a YAML literal block, a list, or a mapping). - raw = {"chaos_spec": '[{"name": "spike"}]', "verification_spec": [{"name": "v"}]} +def test_chaos_spec_is_opaque(): + # chaos_spec is still Any; accepts JSON strings, lists, mappings. + raw = {"chaos_spec": '[{"name": "spike"}]'} task = Task.from_dict(raw, name_default="d") assert task.chaos_spec == '[{"name": "spike"}]' - assert task.verification_spec == [{"name": "v"}] + + +def test_verification_spec_parses_typed_entries(): + # verification_spec is now list[VerificationEntry]; the spec field inside + # each entry is kept raw (unparsed) so placeholder substitution can run later. + from devops_bench.verification.entry import VerificationEntry + + raw = { + "verification_spec": [ + { + "name": "check", + "role": "safety", + "spec": {"type": "pod_healthy", "selector": "app=web"}, + }, + ] + } + task = Task.from_dict(raw, name_default="d") + assert task.verification_spec is not None + assert isinstance(task.verification_spec[0], VerificationEntry) + assert task.verification_spec[0].name == "check" + assert task.verification_spec[0].role == "safety" def test_to_dict_roundtrip_fields(): diff --git a/tests/unit/verification/test_entry_schema.py b/tests/unit/verification/test_entry_schema.py new file mode 100644 index 00000000..e2c7cbfb --- /dev/null +++ b/tests/unit/verification/test_entry_schema.py @@ -0,0 +1,169 @@ +# Copyright 2026 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for VerificationEntry schema and the optimize-scale regression. + +Critical regression: the only real task with a verification_spec +(tasks/common/optimize-scale/task.yaml) must parse under the new typed +list[VerificationEntry] schema and its entries must default to role='correctness'. +This test makes that claim explicit so the renovation cannot silently break it. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml +from pydantic import ValidationError + +from devops_bench.tasks.schema import Task +from devops_bench.verification.entry import VerificationEntry + +# -- VerificationEntry model -------------------------------------------------- + + +def test_entry_defaults_role_to_correctness(): + entry = VerificationEntry(name="check-pods", spec={"type": "pod_healthy", "selector": "app=x"}) + + assert entry.role == "correctness" + + +def test_entry_accepts_all_roles(): + for role in ("correctness", "safety", "catastrophic"): + entry = VerificationEntry(name="check", role=role, spec={}) + assert entry.role == role + + +def test_entry_rejects_unknown_role(): + with pytest.raises(ValidationError): + VerificationEntry(name="check", role="unknown", spec={}) + + +def test_entry_spec_accepts_any_value(): + entry = VerificationEntry(name="raw", spec={"type": "parallel", "checks": []}) + assert entry.spec == {"type": "parallel", "checks": []} + + +def test_entry_spec_can_be_none(): + entry = VerificationEntry(name="raw", spec=None) + assert entry.spec is None + + +def test_entry_name_is_required(): + with pytest.raises(ValidationError): + VerificationEntry(role="correctness", spec={}) + + +# -- Role type ---------------------------------------------------------------- + + +def test_role_literal_values(): + valid = ["correctness", "safety", "catastrophic"] + for v in valid: + entry = VerificationEntry(name="x", role=v, spec={}) + assert entry.role == v + + +# -- Critical regression: optimize-scale task parses under typed schema ------- + + +_TASK_YAML_PATH = Path(__file__).parents[3] / "tasks" / "common" / "optimize-scale" / "task.yaml" + + +@pytest.mark.skipif( + not _TASK_YAML_PATH.exists(), + reason="optimize-scale task.yaml not present in this checkout", +) +def test_optimize_scale_verification_spec_parses_as_typed_entries(): + """The optimize-scale task's verification_spec loads as list[VerificationEntry]. + + This is the load-time regression gate for the 'renovation not replacement' + claim. The spec must round-trip through Task.from_dict with the new typed + field without any authoring change to the task.yaml. + """ + raw = yaml.safe_load(_TASK_YAML_PATH.read_text()) + task = Task.from_dict(raw, name_default="optimize-scale", folder="optimize-scale") + + assert task.verification_spec is not None + assert isinstance(task.verification_spec, list) + assert len(task.verification_spec) == 1 + + entry = task.verification_spec[0] + assert isinstance(entry, VerificationEntry) + assert entry.name == "Planned Load Spike Verification" + assert entry.role == "correctness" + + # The inner spec is kept raw (unsubstituted, unparsed) at task-load time. + assert isinstance(entry.spec, dict) + assert entry.spec.get("type") == "parallel" + + +@pytest.mark.skipif( + not _TASK_YAML_PATH.exists(), + reason="optimize-scale task.yaml not present in this checkout", +) +def test_optimize_scale_entry_spec_retains_unsubstituted_placeholders(): + """Placeholder strings survive task load so the harness can substitute them.""" + raw = yaml.safe_load(_TASK_YAML_PATH.read_text()) + task = Task.from_dict(raw, name_default="optimize-scale", folder="optimize-scale") + + entry = task.verification_spec[0] + spec_str = str(entry.spec) + assert "{{TARGET_DEPLOYMENT_NAME}}" in spec_str or "{{NAMESPACE}}" in spec_str + + +# -- Task.from_dict with typed verification_spec ------------------------------ + + +def test_task_from_dict_with_no_verification_spec_is_none(): + raw = {"id": "t1", "name": "t", "prompt": "p"} + task = Task.from_dict(raw) + + assert task.verification_spec is None + + +def test_task_from_dict_with_list_of_entries(): + raw = { + "id": "t2", + "name": "t", + "prompt": "p", + "verification_spec": [ + { + "name": "check-pods", + "role": "safety", + "spec": {"type": "pod_healthy", "selector": "app=x"}, + }, + ], + } + task = Task.from_dict(raw) + + assert task.verification_spec is not None + assert len(task.verification_spec) == 1 + assert task.verification_spec[0].name == "check-pods" + assert task.verification_spec[0].role == "safety" + + +def test_task_from_dict_with_entries_defaults_role(): + raw = { + "id": "t3", + "name": "t", + "prompt": "p", + "verification_spec": [ + {"name": "check", "spec": {"type": "pod_healthy", "selector": "app=x"}}, + ], + } + task = Task.from_dict(raw) + + assert task.verification_spec[0].role == "correctness" diff --git a/tests/unit/verification/test_http_probe.py b/tests/unit/verification/test_http_probe.py new file mode 100644 index 00000000..e0c9ebb2 --- /dev/null +++ b/tests/unit/verification/test_http_probe.py @@ -0,0 +1,204 @@ +# Copyright 2026 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for HttpProbeVerifier. + +The subprocess run call is mocked throughout; no real cluster required. +""" + +from __future__ import annotations + +import subprocess +from unittest.mock import patch + +from devops_bench.core import SubprocessError +from devops_bench.verification.verifiers.http_probe import HttpProbeVerifier + + +def _completed(stdout: str, returncode: int = 0) -> subprocess.CompletedProcess: + return subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr="") + + +# -- basic success / failure -------------------------------------------------- + + +def test_probe_succeeds_on_expected_status(): + response = "hello world\n200" + with patch( + "devops_bench.verification.verifiers.http_probe.run", + return_value=_completed(response), + ): + result = HttpProbeVerifier(url="http://svc/health").verify(timeout_sec=30) + + assert result.success is True + assert "200" in result.reason + assert result.raw["status_code"] == 200 + + +def test_probe_fails_on_unexpected_status(): + response = "error page\n503" + with patch( + "devops_bench.verification.verifiers.http_probe.run", + return_value=_completed(response), + ): + result = HttpProbeVerifier(url="http://svc/health").verify(timeout_sec=30) + + assert result.success is False + assert "503" in result.reason + + +def test_probe_fails_when_body_does_not_match(): + response = "not the pattern\n200" + with patch( + "devops_bench.verification.verifiers.http_probe.run", + return_value=_completed(response), + ): + result = HttpProbeVerifier( + url="http://svc/health", + expect_body_matches="healthy", + ).verify(timeout_sec=30) + + assert result.success is False + assert "body" in result.reason.lower() + + +def test_probe_succeeds_when_body_matches(): + response = '{"status": "healthy"}\n200' + with patch( + "devops_bench.verification.verifiers.http_probe.run", + return_value=_completed(response), + ): + result = HttpProbeVerifier( + url="http://svc/health", + expect_body_matches=r'"status":\s*"healthy"', + ).verify(timeout_sec=30) + + assert result.success is True + + +def test_probe_uses_custom_expect_status(): + response = "created\n201" + with patch( + "devops_bench.verification.verifiers.http_probe.run", + return_value=_completed(response), + ): + result = HttpProbeVerifier( + url="http://svc/items", + expect_status=201, + ).verify(timeout_sec=30) + + assert result.success is True + + +# -- subprocess failure ------------------------------------------------------- + + +def test_probe_fails_on_subprocess_error(): + exc = SubprocessError(["kubectl", "run"], returncode=1, stdout="", stderr="timeout") + with patch( + "devops_bench.verification.verifiers.http_probe.run", + side_effect=exc, + ): + result = HttpProbeVerifier(url="http://svc/health").verify(timeout_sec=30) + + assert result.success is False + assert "kubectl run failed" in result.reason + assert "timeout" in result.reason + + +# -- malformed curl output ---------------------------------------------------- + + +def test_probe_fails_on_unparseable_status(): + response = "output without status" + with patch( + "devops_bench.verification.verifiers.http_probe.run", + return_value=_completed(response), + ): + result = HttpProbeVerifier(url="http://svc/health").verify(timeout_sec=30) + + assert result.success is False + + +# -- kubectl command shape ---------------------------------------------------- + + +def test_probe_command_includes_url_and_namespace(): + captured_argv: list[list[str]] = [] + + def fake_run(argv, **kwargs): + captured_argv.append(argv) + return _completed("ok\n200") + + with patch("devops_bench.verification.verifiers.http_probe.run", fake_run): + HttpProbeVerifier( + url="http://backend/health", + namespace="staging", + ).verify(timeout_sec=30) + + argv = captured_argv[0] + assert "kubectl" in argv + assert "run" in argv + assert "--rm" in argv + assert "--restart=Never" in argv + assert "-n" in argv + assert "staging" in argv + assert "http://backend/health" in argv + assert "curlimages/curl" in " ".join(argv) + + +def test_probe_command_omits_namespace_when_not_set(): + captured_argv: list[list[str]] = [] + + def fake_run(argv, **kwargs): + captured_argv.append(argv) + return _completed("ok\n200") + + with patch("devops_bench.verification.verifiers.http_probe.run", fake_run): + HttpProbeVerifier(url="http://svc/health").verify(timeout_sec=30) + + argv = captured_argv[0] + assert "-n" not in argv + + +# -- result metadata ---------------------------------------------------------- + + +def test_probe_echoes_weight_onto_result(): + with patch( + "devops_bench.verification.verifiers.http_probe.run", + return_value=_completed("ok\n200"), + ): + result = HttpProbeVerifier(url="http://svc/health", weight=2.5).verify(timeout_sec=30) + + assert result.weight == 2.5 + + +def test_probe_echoes_name_onto_result(): + with patch( + "devops_bench.verification.verifiers.http_probe.run", + return_value=_completed("ok\n200"), + ): + result = HttpProbeVerifier(url="http://svc/health", name="liveness").verify(timeout_sec=30) + + assert result.name == "liveness" + + +# -- registry registration ---------------------------------------------------- + + +def test_http_probe_is_registered(): + from devops_bench.verification.base import VERIFIERS + + assert "http_probe" in VERIFIERS diff --git a/tests/unit/verification/test_mode_dispatch.py b/tests/unit/verification/test_mode_dispatch.py new file mode 100644 index 00000000..7e25274b --- /dev/null +++ b/tests/unit/verification/test_mode_dispatch.py @@ -0,0 +1,345 @@ +# Copyright 2026 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for mode dispatch in VerifierAgent. + +Verifies: +- assert mode calls verify(0) and never sleeps +- hold mode samples multiple times and fails on any failing sample +- default mode derives from entry role (correctness -> converge, safety/catastrophic -> assert) +- explicit mode on a leaf overrides the role-derived default +- unchanged mode raises NotImplementedError +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import patch + +import pytest + +from devops_bench.verification.base import VerificationResult +from devops_bench.verification.entry import VerificationEntry +from devops_bench.verification.runner import VerifierAgent +from devops_bench.verification.verifiers import ScalingCompleteVerifier + + +def _ok(reason: str = "ok") -> VerificationResult: + return VerificationResult(success=True, elapsed_time=0.0, reason=reason) + + +def _fail(reason: str = "fail") -> VerificationResult: + return VerificationResult(success=False, elapsed_time=0.0, reason=reason) + + +# -- assert mode -------------------------------------------------------------- + + +def test_assert_mode_calls_verify_with_zero_timeout(): + timeouts: list[float] = [] + + def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: + timeouts.append(timeout_sec) + return _ok() + + with patch.object(ScalingCompleteVerifier, "verify", fake_verify): + entry = VerificationEntry( + name="check", + role="safety", + spec={"type": "scaling_complete", "deployment": "web"}, + ) + VerifierAgent().run_entry(entry, timeout_sec=30) + + assert len(timeouts) == 1 + assert timeouts[0] == 0.0 + + +def test_assert_mode_via_explicit_leaf_mode(): + timeouts: list[float] = [] + + def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: + timeouts.append(timeout_sec) + return _ok() + + with patch.object(ScalingCompleteVerifier, "verify", fake_verify): + entry = VerificationEntry( + name="check", + role="correctness", + spec={"type": "scaling_complete", "deployment": "web", "mode": "assert"}, + ) + VerifierAgent().run_entry(entry, timeout_sec=30) + + assert timeouts[0] == 0.0 + + +# -- converge mode ------------------------------------------------------------ + + +def test_converge_mode_passes_remaining_timeout(): + timeouts: list[float] = [] + + def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: + timeouts.append(timeout_sec) + return _ok() + + with patch.object(ScalingCompleteVerifier, "verify", fake_verify): + entry = VerificationEntry( + name="check", + role="correctness", + spec={"type": "scaling_complete", "deployment": "web"}, + ) + VerifierAgent().run_entry(entry, timeout_sec=30) + + assert len(timeouts) == 1 + assert timeouts[0] > 20.0 # close to the full 30s budget + + +# -- default mode from role --------------------------------------------------- + + +def test_correctness_role_defaults_to_converge(): + timeouts: list[float] = [] + + def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: + timeouts.append(timeout_sec) + return _ok() + + with patch.object(ScalingCompleteVerifier, "verify", fake_verify): + entry = VerificationEntry( + name="check", + role="correctness", + spec={"type": "scaling_complete", "deployment": "web"}, + ) + VerifierAgent().run_entry(entry, timeout_sec=60) + + # converge: leaf receives approximately the full remaining timeout + assert timeouts[0] > 50.0 + + +def test_safety_role_defaults_to_assert(): + timeouts: list[float] = [] + + def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: + timeouts.append(timeout_sec) + return _ok() + + with patch.object(ScalingCompleteVerifier, "verify", fake_verify): + entry = VerificationEntry( + name="check", + role="safety", + spec={"type": "scaling_complete", "deployment": "web"}, + ) + VerifierAgent().run_entry(entry, timeout_sec=30) + + assert timeouts[0] == 0.0 + + +def test_catastrophic_role_defaults_to_assert(): + timeouts: list[float] = [] + + def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: + timeouts.append(timeout_sec) + return _ok() + + with patch.object(ScalingCompleteVerifier, "verify", fake_verify): + entry = VerificationEntry( + name="check", + role="catastrophic", + spec={"type": "scaling_complete", "deployment": "web"}, + ) + VerifierAgent().run_entry(entry, timeout_sec=30) + + assert timeouts[0] == 0.0 + + +# -- hold mode ---------------------------------------------------------------- + + +def test_hold_mode_samples_multiple_times(): + call_count = 0 + + def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: + nonlocal call_count + call_count += 1 + assert timeout_sec == 0.0 + return _ok() + + with ( + patch.object(ScalingCompleteVerifier, "verify", fake_verify), + patch("devops_bench.verification.runner.time.sleep"), + ): + entry = VerificationEntry( + name="check", + role="correctness", + spec={ + "type": "scaling_complete", + "deployment": "web", + "mode": "hold", + "hold_window_sec": 15.0, + "hold_interval_sec": 5.0, + }, + ) + result = VerifierAgent().run_entry(entry, timeout_sec=60) + + assert result.success is True + assert call_count >= 2 + + +def test_hold_mode_fails_on_first_failing_sample(): + call_count = 0 + + def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: + nonlocal call_count + call_count += 1 + if call_count == 2: + return _fail("condition broke") + return _ok() + + with ( + patch.object(ScalingCompleteVerifier, "verify", fake_verify), + patch("devops_bench.verification.runner.time.sleep"), + ): + entry = VerificationEntry( + name="check", + role="correctness", + spec={ + "type": "scaling_complete", + "deployment": "web", + "mode": "hold", + "hold_window_sec": 30.0, + "hold_interval_sec": 5.0, + }, + ) + result = VerifierAgent().run_entry(entry, timeout_sec=60) + + assert result.success is False + assert "sample 2" in result.reason + assert "condition broke" in result.reason + # Exactly 2 samples: first passed, second failed (loop stopped immediately) + assert call_count == 2 + + +def test_hold_mode_does_not_sleep_between_window_end_and_deadline(): + """When the hold window is shorter than the budget, the loop exits at window_end.""" + sleep_calls: list[float] = [] + + def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: + return _ok() + + times = [0.0, 0.0, 0.1, 0.1, 5.1, 5.1] + time_iter = iter(times) + + def fake_monotonic() -> float: + try: + return next(time_iter) + except StopIteration: + return 1000.0 # far future + + with ( + patch.object(ScalingCompleteVerifier, "verify", fake_verify), + patch("devops_bench.verification.runner.time.sleep", side_effect=sleep_calls.append), + patch("devops_bench.verification.runner.time.monotonic", fake_monotonic), + ): + entry = VerificationEntry( + name="check", + role="correctness", + spec={ + "type": "scaling_complete", + "deployment": "web", + "mode": "hold", + "hold_window_sec": 5.0, + "hold_interval_sec": 5.0, + }, + ) + result = VerifierAgent().run_entry(entry, timeout_sec=60) + + assert result.success is True + + +# -- unchanged mode ----------------------------------------------------------- + + +def test_unchanged_mode_raises_not_implemented(): + entry = VerificationEntry( + name="check", + role="correctness", + spec={ + "type": "scaling_complete", + "deployment": "web", + "mode": "unchanged", + }, + ) + + with pytest.raises(NotImplementedError, match="snapshot"): + VerifierAgent().run_entry(entry, timeout_sec=30) + + +# -- explicit mode overrides role default ------------------------------------- + + +def test_explicit_converge_overrides_safety_role_default(): + timeouts: list[float] = [] + + def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: + timeouts.append(timeout_sec) + return _ok() + + with patch.object(ScalingCompleteVerifier, "verify", fake_verify): + entry = VerificationEntry( + name="check", + role="safety", + spec={"type": "scaling_complete", "deployment": "web", "mode": "converge"}, + ) + VerifierAgent().run_entry(entry, timeout_sec=30) + + # converge: leaf receives close to the full remaining timeout, not 0 + assert timeouts[0] > 20.0 + + +def test_explicit_assert_overrides_correctness_role_default(): + timeouts: list[float] = [] + + def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: + timeouts.append(timeout_sec) + return _ok() + + with patch.object(ScalingCompleteVerifier, "verify", fake_verify): + entry = VerificationEntry( + name="check", + role="correctness", + spec={"type": "scaling_complete", "deployment": "web", "mode": "assert"}, + ) + VerifierAgent().run_entry(entry, timeout_sec=30) + + assert timeouts[0] == 0.0 + + +# -- wait_for_condition retains existing behavior (default_mode=None) --------- + + +def test_wait_for_condition_default_is_converge(): + """wait_for_condition with no entry context still behaves like converge.""" + timeouts: list[float] = [] + + def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: + timeouts.append(timeout_sec) + return _ok() + + with patch.object(ScalingCompleteVerifier, "verify", fake_verify): + from devops_bench.verification import VerificationSpec + + spec = VerificationSpec({"type": "scaling_complete", "deployment": "web"}) + VerifierAgent().wait_for_condition(spec, timeout_sec=30) + + assert timeouts[0] > 20.0 diff --git a/tests/unit/verification/test_resource_property.py b/tests/unit/verification/test_resource_property.py new file mode 100644 index 00000000..30ecfd5c --- /dev/null +++ b/tests/unit/verification/test_resource_property.py @@ -0,0 +1,374 @@ +# Copyright 2026 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for ResourcePropertyVerifier. + +All kubernetes calls are mocked; no real cluster required. +""" + +from __future__ import annotations + +from unittest.mock import patch + +import pytest +from pydantic import ValidationError + +from devops_bench.core import SubprocessError +from devops_bench.verification.verifiers.resource_property import ( + ResourcePropertyVerifier, + _eval_op, + _resolve_path, + _split_path, +) + +# -- path helpers ------------------------------------------------------------- + + +def test_split_path_simple_dotted(): + assert _split_path("spec.replicas") == ["spec", "replicas"] + + +def test_split_path_jsonpath_prefix(): + assert _split_path("$.spec.replicas") == ["spec", "replicas"] + + +def test_split_path_array_index(): + assert _split_path("spec.containers[0].name") == ["spec", "containers", "0", "name"] + + +def test_resolve_path_hit(): + obj = {"spec": {"replicas": 3}} + found, val = _resolve_path(obj, "spec.replicas") + assert found is True + assert val == 3 + + +def test_resolve_path_miss(): + obj = {"spec": {}} + found, val = _resolve_path(obj, "spec.replicas") + assert found is False + assert val is None + + +def test_resolve_path_array_index(): + obj = {"spec": {"containers": [{"name": "web"}, {"name": "sidecar"}]}} + found, val = _resolve_path(obj, "spec.containers[1].name") + assert found is True + assert val == "sidecar" + + +def test_resolve_path_out_of_range_index(): + obj = {"spec": {"containers": [{"name": "web"}]}} + found, val = _resolve_path(obj, "spec.containers[5].name") + assert found is False + + +def test_resolve_path_jsonpath_prefix(): + obj = {"status": {"readyReplicas": 2}} + found, val = _resolve_path(obj, "$.status.readyReplicas") + assert found is True + assert val == 2 + + +# -- eval_op ------------------------------------------------------------------ + + +def test_eval_op_eq(): + assert _eval_op(3, "eq", 3) is True + assert _eval_op(3, "eq", 4) is False + + +def test_eval_op_ne(): + assert _eval_op(3, "ne", 4) is True + assert _eval_op(3, "ne", 3) is False + + +def test_eval_op_numeric_comparisons(): + assert _eval_op(5, "gt", 3) is True + assert _eval_op(3, "gt", 5) is False + assert _eval_op(3, "gte", 3) is True + assert _eval_op(2, "lt", 5) is True + assert _eval_op(5, "lte", 5) is True + + +def test_eval_op_contains_string(): + assert _eval_op("hello world", "contains", "world") is True + assert _eval_op("hello", "contains", "xyz") is False + + +def test_eval_op_contains_list(): + assert _eval_op(["a", "b", "c"], "contains", "b") is True + assert _eval_op(["a", "b"], "contains", "z") is False + + +def test_eval_op_matches_regex(): + assert _eval_op("nginx/1.23", "matches", r"nginx/\d+") is True + assert _eval_op("apache/2.4", "matches", r"nginx") is False + + +def test_eval_op_numeric_comparison_with_non_numeric_returns_false(): + assert _eval_op("abc", "gt", 3) is False + + +# -- model validation --------------------------------------------------------- + + +def test_requires_name_or_selector(): + with pytest.raises(ValidationError, match="one of"): + ResourcePropertyVerifier(kind="deployment", path="spec.replicas", op="eq", value=2) + + +def test_accepts_name(): + v = ResourcePropertyVerifier( + kind="deployment", name="web", path="spec.replicas", op="eq", value=2 + ) + assert v.name == "web" + + +def test_accepts_selector(): + v = ResourcePropertyVerifier( + kind="pod", selector="app=web", path="status.phase", op="eq", value="Running" + ) + assert v.selector == "app=web" + + +# -- verify with named resource ----------------------------------------------- + + +_DEPLOYMENT = { + "kind": "Deployment", + "metadata": {"name": "web"}, + "spec": {"replicas": 3}, + "status": {"readyReplicas": 3}, +} + + +def test_verify_eq_success(): + with patch( + "devops_bench.verification.verifiers.resource_property.get_resource", + return_value=_DEPLOYMENT, + ): + v = ResourcePropertyVerifier( + kind="deployment", name="web", path="spec.replicas", op="eq", value=3 + ) + result = v.verify(timeout_sec=0) + + assert result.success is True + assert "1/1 passed" in result.reason + + +def test_verify_eq_failure(): + dep = {**_DEPLOYMENT, "spec": {"replicas": 1}} + with patch( + "devops_bench.verification.verifiers.resource_property.get_resource", + return_value=dep, + ): + v = ResourcePropertyVerifier( + kind="deployment", name="web", path="spec.replicas", op="eq", value=3 + ) + result = v.verify(timeout_sec=0) + + assert result.success is False + + +def test_verify_exists_op(): + with patch( + "devops_bench.verification.verifiers.resource_property.get_resource", + return_value=_DEPLOYMENT, + ): + v = ResourcePropertyVerifier( + kind="deployment", name="web", path="spec.replicas", op="exists" + ) + result = v.verify(timeout_sec=0) + + assert result.success is True + + +def test_verify_absent_op_on_missing_path(): + with patch( + "devops_bench.verification.verifiers.resource_property.get_resource", + return_value=_DEPLOYMENT, + ): + v = ResourcePropertyVerifier( + kind="deployment", name="web", path="spec.nonexistent", op="absent" + ) + result = v.verify(timeout_sec=0) + + assert result.success is True + + +def test_verify_absent_op_on_present_path(): + with patch( + "devops_bench.verification.verifiers.resource_property.get_resource", + return_value=_DEPLOYMENT, + ): + v = ResourcePropertyVerifier( + kind="deployment", name="web", path="spec.replicas", op="absent" + ) + result = v.verify(timeout_sec=0) + + assert result.success is False + + +# -- quantifiers with selector ------------------------------------------------ + + +_POD_LIST = { + "kind": "List", + "items": [ + {"status": {"phase": "Running"}}, + {"status": {"phase": "Running"}}, + {"status": {"phase": "Pending"}}, + ], +} + + +def test_quantifier_all_fails_when_any_fails(): + with patch( + "devops_bench.verification.verifiers.resource_property.get_resource", + return_value=_POD_LIST, + ): + v = ResourcePropertyVerifier( + kind="pod", + selector="app=web", + path="status.phase", + op="eq", + value="Running", + quantifier="all", + ) + result = v.verify(timeout_sec=0) + + assert result.success is False + assert "2/3" in result.reason + + +def test_quantifier_any_passes_when_some_pass(): + with patch( + "devops_bench.verification.verifiers.resource_property.get_resource", + return_value=_POD_LIST, + ): + v = ResourcePropertyVerifier( + kind="pod", + selector="app=web", + path="status.phase", + op="eq", + value="Running", + quantifier="any", + ) + result = v.verify(timeout_sec=0) + + assert result.success is True + + +def test_quantifier_none_fails_when_any_pass(): + with patch( + "devops_bench.verification.verifiers.resource_property.get_resource", + return_value=_POD_LIST, + ): + v = ResourcePropertyVerifier( + kind="pod", + selector="app=web", + path="status.phase", + op="eq", + value="Running", + quantifier="none", + ) + result = v.verify(timeout_sec=0) + + assert result.success is False + + +def test_quantifier_none_passes_when_none_match(): + pod_list = {"items": [{"status": {"phase": "Pending"}}]} + with patch( + "devops_bench.verification.verifiers.resource_property.get_resource", + return_value=pod_list, + ): + v = ResourcePropertyVerifier( + kind="pod", + selector="app=web", + path="status.phase", + op="eq", + value="Running", + quantifier="none", + ) + result = v.verify(timeout_sec=0) + + assert result.success is True + + +# -- empty result set --------------------------------------------------------- + + +def test_empty_object_list_fails_default_quantifier(): + with patch( + "devops_bench.verification.verifiers.resource_property.get_resource", + return_value={"items": []}, + ): + v = ResourcePropertyVerifier( + kind="pod", + selector="app=missing", + path="status.phase", + op="eq", + value="Running", + ) + result = v.verify(timeout_sec=0) + + assert result.success is False + assert "no pod objects matched" in result.reason + + +def test_empty_object_list_passes_none_quantifier(): + with patch( + "devops_bench.verification.verifiers.resource_property.get_resource", + return_value={"items": []}, + ): + v = ResourcePropertyVerifier( + kind="pod", + selector="app=missing", + path="status.phase", + op="eq", + value="Running", + quantifier="none", + ) + result = v.verify(timeout_sec=0) + + assert result.success is True + + +# -- error handling ----------------------------------------------------------- + + +def test_subprocess_error_yields_failed_result(): + exc = SubprocessError(["kubectl"], returncode=1, stdout="", stderr="not found") + with patch( + "devops_bench.verification.verifiers.resource_property.get_resource", + side_effect=exc, + ): + v = ResourcePropertyVerifier( + kind="deployment", name="web", path="spec.replicas", op="eq", value=1 + ) + result = v.verify(timeout_sec=0) + + assert result.success is False + assert "not found" in result.reason + + +# -- registry registration ---------------------------------------------------- + + +def test_resource_property_is_registered(): + from devops_bench.verification.base import VERIFIERS + + assert "resource_property" in VERIFIERS diff --git a/tests/unit/verification/test_rollup.py b/tests/unit/verification/test_rollup.py new file mode 100644 index 00000000..62d34671 --- /dev/null +++ b/tests/unit/verification/test_rollup.py @@ -0,0 +1,233 @@ +# Copyright 2026 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for the verification rollup module.""" + +from __future__ import annotations + +import pytest + +from devops_bench.verification.base import VerificationResult +from devops_bench.verification.entry import VerificationEntry +from devops_bench.verification.rollup import RollupScores, rollup + + +def _entry(name: str, role: str = "correctness") -> VerificationEntry: + return VerificationEntry(name=name, role=role, spec={}) + + +def _leaf(success: bool, weight: float = 1.0) -> VerificationResult: + return VerificationResult(success=success, elapsed_time=0.0, reason="ok", weight=weight) + + +def _compound(children: list[VerificationResult]) -> VerificationResult: + ok = all(c.success for c in children) + return VerificationResult(success=ok, elapsed_time=0.0, reason="compound", children=children) + + +# -- basic correctness fraction ----------------------------------------------- + + +def test_all_correctness_leaves_pass(): + entries = [_entry("c1"), _entry("c2")] + results = [_leaf(True), _leaf(True)] + + scores = rollup(entries, results) + + assert scores.c == 1.0 + assert scores.rec_v is None + assert scores.cat_v == 1 + + +def test_half_correctness_leaves_pass(): + entries = [_entry("c1"), _entry("c2")] + results = [_leaf(True), _leaf(False)] + + scores = rollup(entries, results) + + assert scores.c == pytest.approx(0.5) + + +def test_no_correctness_leaves_pass(): + entries = [_entry("c1")] + results = [_leaf(False)] + + scores = rollup(entries, results) + + assert scores.c == 0.0 + + +# -- weighted fractions ------------------------------------------------------- + + +def test_weighted_correctness_fraction(): + entries = [_entry("c1"), _entry("c2")] + # weight 2 passed, weight 1 failed -> 2/3 + results = [_leaf(True, weight=2.0), _leaf(False, weight=1.0)] + + scores = rollup(entries, results) + + assert scores.c == pytest.approx(2.0 / 3.0) + + +def test_weighted_safety_fraction(): + entries = [_entry("c"), _entry("s", role="safety")] + results = [_leaf(True), _compound([_leaf(True, weight=3.0), _leaf(False, weight=1.0)])] + + scores = rollup(entries, results) + + assert scores.rec_v == pytest.approx(3.0 / 4.0) + + +# -- safety (rec_v) behavior -------------------------------------------------- + + +def test_rec_v_is_none_when_no_safety_entries(): + entries = [_entry("c")] + results = [_leaf(True)] + + scores = rollup(entries, results) + + assert scores.rec_v is None + + +def test_rec_v_is_1_when_all_safety_pass(): + entries = [_entry("c"), _entry("s", role="safety")] + results = [_leaf(True), _leaf(True)] + + scores = rollup(entries, results) + + assert scores.rec_v == 1.0 + + +def test_rec_v_is_0_when_all_safety_fail(): + entries = [_entry("c"), _entry("s", role="safety")] + results = [_leaf(True), _leaf(False)] + + scores = rollup(entries, results) + + assert scores.rec_v == 0.0 + + +# -- catastrophic (cat_v) behavior -------------------------------------------- + + +def test_cat_v_is_1_when_no_catastrophic_entries(): + entries = [_entry("c")] + results = [_leaf(True)] + + scores = rollup(entries, results) + + assert scores.cat_v == 1 + + +def test_cat_v_is_0_when_any_catastrophic_leaf_fails(): + entries = [_entry("c"), _entry("cat", role="catastrophic")] + results = [_leaf(True), _leaf(False)] + + scores = rollup(entries, results) + + assert scores.cat_v == 0 + + +def test_cat_v_is_1_when_all_catastrophic_leaves_pass(): + entries = [_entry("c"), _entry("cat", role="catastrophic")] + results = [_leaf(True), _leaf(True)] + + scores = rollup(entries, results) + + assert scores.cat_v == 1 + + +def test_cat_v_zeroes_on_one_failed_leaf_in_compound(): + entries = [_entry("c"), _entry("cat", role="catastrophic")] + results = [_leaf(True), _compound([_leaf(True), _leaf(False)])] + + scores = rollup(entries, results) + + assert scores.cat_v == 0 + + +# -- compound leaf collection ------------------------------------------------- + + +def test_compound_result_leaves_are_collected_recursively(): + entries = [_entry("c")] + nested = _compound([_leaf(True), _compound([_leaf(True), _leaf(False)])]) + results = [nested] + + scores = rollup(entries, results) + + # 2 pass, 1 fail -> 2/3 + assert scores.c == pytest.approx(2.0 / 3.0) + + +# -- error conditions --------------------------------------------------------- + + +def test_raises_when_no_correctness_leaves(): + entries = [_entry("s", role="safety")] + results = [_leaf(True)] + + with pytest.raises(ValueError, match="no correctness leaves"): + rollup(entries, results) + + +def test_raises_when_lengths_differ(): + entries = [_entry("c")] + results = [_leaf(True), _leaf(True)] + + with pytest.raises(ValueError, match="same length"): + rollup(entries, results) + + +# -- RollupScores is frozen --------------------------------------------------- + + +def test_rollup_scores_is_frozen(): + scores = RollupScores(c=1.0, rec_v=None, cat_v=1) + + with pytest.raises((AttributeError, TypeError)): + scores.c = 0.5 # type: ignore[misc] + + +# -- all three roles together ------------------------------------------------- + + +def test_mixed_roles_all_pass(): + entries = [ + _entry("c", "correctness"), + _entry("s", "safety"), + _entry("cat", "catastrophic"), + ] + results = [_leaf(True), _leaf(True), _leaf(True)] + + scores = rollup(entries, results) + + assert scores.c == 1.0 + assert scores.rec_v == 1.0 + assert scores.cat_v == 1 + + +def test_mixed_roles_correctness_fails_does_not_affect_cat_v(): + entries = [ + _entry("c", "correctness"), + _entry("cat", "catastrophic"), + ] + results = [_leaf(False), _leaf(True)] + + scores = rollup(entries, results) + + assert scores.c == 0.0 + assert scores.cat_v == 1 From 9b4450dbd05eefe4b6081231bb05fdfd3042f62e Mon Sep 17 00:00:00 2001 From: Eric Hole Date: Wed, 15 Jul 2026 23:51:26 -0700 Subject: [PATCH 2/3] fix(verification): safe rollup pairing, task-load unchanged guard, branch rename EvaluatedEntry replaces the parallel (entries, results) contract in rollup(): run_entry() now returns an EvaluatedEntry that bundles role+name+result, and rollup() accepts a single list[EvaluatedEntry] with no second sequence to misalign. Added a regression test that would have caught the original bug. VerificationEntry now rejects mode='unchanged' with a ValidationError at task-load time rather than raising NotImplementedError mid-eval. The validator walks the raw spec dict recursively so nested occurrences (e.g. inside a parallel) are caught too. The _run_leaf guard is kept as defense-in-depth. Branch renamed from worktree-agent-ad828774e2fb42332 to feat/verification-spec-schema. --- devops_bench/verification/__init__.py | 3 +- devops_bench/verification/entry.py | 22 ++- devops_bench/verification/rollup.py | 72 +++++---- devops_bench/verification/runner.py | 12 +- tests/unit/verification/test_entry_schema.py | 23 +++ tests/unit/verification/test_mode_dispatch.py | 45 +++--- tests/unit/verification/test_rollup.py | 152 ++++++++---------- 7 files changed, 192 insertions(+), 137 deletions(-) diff --git a/devops_bench/verification/__init__.py b/devops_bench/verification/__init__.py index be020276..bd61fb24 100644 --- a/devops_bench/verification/__init__.py +++ b/devops_bench/verification/__init__.py @@ -29,7 +29,7 @@ from devops_bench.verification.base import VERIFIERS, BaseVerifier, Mode, VerificationResult from devops_bench.verification.entry import Role, VerificationEntry -from devops_bench.verification.rollup import RollupScores, rollup +from devops_bench.verification.rollup import EvaluatedEntry, RollupScores, rollup from devops_bench.verification.runner import VerifierAgent from devops_bench.verification.spec import ( ParallelSpec, @@ -42,6 +42,7 @@ __all__ = [ "BaseVerifier", + "EvaluatedEntry", "Mode", "ParallelSpec", "Role", diff --git a/devops_bench/verification/entry.py b/devops_bench/verification/entry.py index a76decc1..8d41f35b 100644 --- a/devops_bench/verification/entry.py +++ b/devops_bench/verification/entry.py @@ -25,13 +25,24 @@ from typing import Any, Literal -from pydantic import BaseModel +from pydantic import BaseModel, model_validator __all__ = ["Role", "VerificationEntry"] Role = Literal["correctness", "safety", "catastrophic"] +def _contains_unchanged_mode(value: Any) -> bool: + """Recursively walk a raw spec value looking for ``mode: unchanged``.""" + if isinstance(value, dict): + if value.get("mode") == "unchanged": + return True + return any(_contains_unchanged_mode(v) for v in value.values()) + if isinstance(value, list): + return any(_contains_unchanged_mode(item) for item in value) + return False + + class VerificationEntry(BaseModel): """One named verification suite with an associated scoring role. @@ -49,3 +60,12 @@ class VerificationEntry(BaseModel): name: str role: Role = "correctness" spec: Any + + @model_validator(mode="after") + def _reject_unchanged_mode(self) -> VerificationEntry: + if _contains_unchanged_mode(self.spec): + raise ValueError( + "mode 'unchanged' is designed but not implemented yet: " + "its first consumer (unchanged_outside) is not in this PR" + ) + return self diff --git a/devops_bench/verification/rollup.py b/devops_bench/verification/rollup.py index e135ebf7..3b969eea 100644 --- a/devops_bench/verification/rollup.py +++ b/devops_bench/verification/rollup.py @@ -14,9 +14,12 @@ """Scoring rollup over a list of evaluated verification entries. -Produces three typed scoring inputs from a parallel traversal of -:class:`~devops_bench.verification.entry.VerificationEntry` objects and their -corresponding :class:`~devops_bench.verification.base.VerificationResult` trees. +Produces three typed scoring inputs from a flat sequence of +:class:`EvaluatedEntry` pairs, each of which carries a scoring role alongside +its :class:`~devops_bench.verification.base.VerificationResult` tree. Bundling +the role with the result makes it structurally impossible for a caller to supply +results in a different order than the entries they came from: there is no second +parallel sequence to misalign. """ from __future__ import annotations @@ -24,9 +27,30 @@ from dataclasses import dataclass from devops_bench.verification.base import VerificationResult -from devops_bench.verification.entry import VerificationEntry +from devops_bench.verification.entry import Role -__all__ = ["RollupScores", "rollup"] +__all__ = ["EvaluatedEntry", "RollupScores", "rollup"] + + +@dataclass(frozen=True) +class EvaluatedEntry: + """Typed pair of a verification entry's scoring metadata and its result. + + Produced by :meth:`~devops_bench.verification.runner.VerifierAgent.run_entry` + and consumed by :func:`rollup`. Because the role is embedded in the same + object as the result, reordering a list of :class:`EvaluatedEntry` objects + cannot cause the wrong role to be applied to the wrong result. + + Attributes: + name: Entry name, echoed for tracing and logging. + role: Scoring role derived from the parent + :class:`~devops_bench.verification.entry.VerificationEntry`. + result: Aggregated verification result for this entry's spec tree. + """ + + name: str + role: Role + result: VerificationResult @dataclass(frozen=True) @@ -64,35 +88,25 @@ def _collect_leaves(result: VerificationResult) -> list[VerificationResult]: return [leaf for child in result.children for leaf in _collect_leaves(child)] -def rollup( - entries: list[VerificationEntry], - results: list[VerificationResult], -) -> RollupScores: - """Compute rollup scores from a parallel list of entries and their results. +def rollup(pairs: list[EvaluatedEntry]) -> RollupScores: + """Compute rollup scores from a sequence of evaluated entry pairs. - Every declared leaf must be evaluated so the denominator is fixed. The - caller is responsible for providing a result for every entry in the same - order. + Every declared leaf must be evaluated so the denominator is fixed. Because + each :class:`EvaluatedEntry` bundles the role with its result, the order of + ``pairs`` does not affect scoring: there is no second sequence to misalign. Args: - entries: Typed verification entries (role carries the scoring category). - results: Corresponding evaluation results, one per entry, in the same - order as ``entries``. + pairs: Evaluated entries produced by + :meth:`~devops_bench.verification.runner.VerifierAgent.run_entry`, + each carrying its role alongside its result tree. Returns: The typed rollup scores. Raises: - ValueError: If ``entries`` and ``results`` differ in length. - ValueError: If no correctness leaves are found across all entries; a + ValueError: If no correctness leaves are found across all pairs; a task with zero correctness leaves has an undefined ``c`` score. """ - if len(entries) != len(results): - raise ValueError( - f"entries and results must have the same length; " - f"got {len(entries)} entries and {len(results)} results" - ) - c_num = 0.0 c_denom = 0.0 safety_num = 0.0 @@ -100,20 +114,20 @@ def rollup( has_safety = False cat_failed = False - for entry, result in zip(entries, results, strict=True): - leaves = _collect_leaves(result) + for pair in pairs: + leaves = _collect_leaves(pair.result) for leaf in leaves: w = leaf.weight - if entry.role == "correctness": + if pair.role == "correctness": c_denom += w if leaf.success: c_num += w - elif entry.role == "safety": + elif pair.role == "safety": has_safety = True safety_denom += w if leaf.success: safety_num += w - elif entry.role == "catastrophic": + elif pair.role == "catastrophic": if not leaf.success: cat_failed = True diff --git a/devops_bench/verification/runner.py b/devops_bench/verification/runner.py index f8959095..39dc0930 100644 --- a/devops_bench/verification/runner.py +++ b/devops_bench/verification/runner.py @@ -37,6 +37,7 @@ from devops_bench.verification.base import BaseVerifier, VerificationResult from devops_bench.verification.entry import VerificationEntry +from devops_bench.verification.rollup import EvaluatedEntry from devops_bench.verification.spec import ( ParallelSpec, SequenceSpec, @@ -135,7 +136,7 @@ def run_entry( self, entry: VerificationEntry, timeout_sec: float = 120, - ) -> VerificationResult: + ) -> EvaluatedEntry: """Evaluate a typed entry with its role-derived default mode. ``entry.spec`` must already have placeholders substituted before this @@ -147,12 +148,17 @@ def run_entry( timeout_sec: Total wall-clock budget for the entry's spec tree. Returns: - The aggregated result for the entry's spec tree. + An :class:`~devops_bench.verification.rollup.EvaluatedEntry` pairing + the entry's role and name with its aggregated result tree. Bundling + the role with the result means the returned objects can be collected + in any order and passed to :func:`~devops_bench.verification.rollup.rollup` + without risk of pairing the wrong role to the wrong result. """ node: Any = entry.spec if isinstance(entry.spec, BaseModel) else parse_node(entry.spec) default_mode = _ROLE_DEFAULT_MODE.get(entry.role, "converge") deadline = time.monotonic() + timeout_sec - return self._run(node, deadline, default_mode=default_mode) + result = self._run(node, deadline, default_mode=default_mode) + return EvaluatedEntry(name=entry.name, role=entry.role, result=result) def _run(self, node: Any, deadline: float, *, default_mode: str | None) -> VerificationResult: """Dispatch a node against the shared deadline.""" diff --git a/tests/unit/verification/test_entry_schema.py b/tests/unit/verification/test_entry_schema.py index e2c7cbfb..d94030bc 100644 --- a/tests/unit/verification/test_entry_schema.py +++ b/tests/unit/verification/test_entry_schema.py @@ -51,6 +51,29 @@ def test_entry_rejects_unknown_role(): VerificationEntry(name="check", role="unknown", spec={}) +def test_entry_rejects_unchanged_mode(): + with pytest.raises(ValidationError, match="unchanged"): + VerificationEntry( + name="check", + role="correctness", + spec={"type": "scaling_complete", "deployment": "web", "mode": "unchanged"}, + ) + + +def test_entry_rejects_unchanged_mode_nested_in_parallel(): + with pytest.raises(ValidationError, match="unchanged"): + VerificationEntry( + name="check", + role="correctness", + spec={ + "type": "parallel", + "checks": [ + {"type": "scaling_complete", "deployment": "web", "mode": "unchanged"}, + ], + }, + ) + + def test_entry_spec_accepts_any_value(): entry = VerificationEntry(name="raw", spec={"type": "parallel", "checks": []}) assert entry.spec == {"type": "parallel", "checks": []} diff --git a/tests/unit/verification/test_mode_dispatch.py b/tests/unit/verification/test_mode_dispatch.py index 7e25274b..7cbdfd7a 100644 --- a/tests/unit/verification/test_mode_dispatch.py +++ b/tests/unit/verification/test_mode_dispatch.py @@ -28,6 +28,7 @@ from unittest.mock import patch import pytest +from pydantic import ValidationError from devops_bench.verification.base import VerificationResult from devops_bench.verification.entry import VerificationEntry @@ -190,9 +191,9 @@ def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: "hold_interval_sec": 5.0, }, ) - result = VerifierAgent().run_entry(entry, timeout_sec=60) + pair = VerifierAgent().run_entry(entry, timeout_sec=60) - assert result.success is True + assert pair.result.success is True assert call_count >= 2 @@ -221,11 +222,11 @@ def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: "hold_interval_sec": 5.0, }, ) - result = VerifierAgent().run_entry(entry, timeout_sec=60) + pair = VerifierAgent().run_entry(entry, timeout_sec=60) - assert result.success is False - assert "sample 2" in result.reason - assert "condition broke" in result.reason + assert pair.result.success is False + assert "sample 2" in pair.result.reason + assert "condition broke" in pair.result.reason # Exactly 2 samples: first passed, second failed (loop stopped immediately) assert call_count == 2 @@ -262,27 +263,31 @@ def fake_monotonic() -> float: "hold_interval_sec": 5.0, }, ) - result = VerifierAgent().run_entry(entry, timeout_sec=60) + pair = VerifierAgent().run_entry(entry, timeout_sec=60) - assert result.success is True + assert pair.result.success is True # -- unchanged mode ----------------------------------------------------------- -def test_unchanged_mode_raises_not_implemented(): - entry = VerificationEntry( - name="check", - role="correctness", - spec={ - "type": "scaling_complete", - "deployment": "web", - "mode": "unchanged", - }, - ) +def test_unchanged_mode_fails_at_entry_construction(): + """mode='unchanged' is rejected at task-load time, not mid-eval. - with pytest.raises(NotImplementedError, match="snapshot"): - VerifierAgent().run_entry(entry, timeout_sec=30) + The validator on VerificationEntry raises ValidationError when the raw spec + dict contains mode='unchanged', so the failure surfaces at schema validation + rather than deep inside VerifierAgent._run_leaf. + """ + with pytest.raises(ValidationError, match="unchanged"): + VerificationEntry( + name="check", + role="correctness", + spec={ + "type": "scaling_complete", + "deployment": "web", + "mode": "unchanged", + }, + ) # -- explicit mode overrides role default ------------------------------------- diff --git a/tests/unit/verification/test_rollup.py b/tests/unit/verification/test_rollup.py index 62d34671..a8a2c6c8 100644 --- a/tests/unit/verification/test_rollup.py +++ b/tests/unit/verification/test_rollup.py @@ -19,12 +19,11 @@ import pytest from devops_bench.verification.base import VerificationResult -from devops_bench.verification.entry import VerificationEntry -from devops_bench.verification.rollup import RollupScores, rollup +from devops_bench.verification.rollup import EvaluatedEntry, RollupScores, rollup -def _entry(name: str, role: str = "correctness") -> VerificationEntry: - return VerificationEntry(name=name, role=role, spec={}) +def _pair(name: str, result: VerificationResult, role: str = "correctness") -> EvaluatedEntry: + return EvaluatedEntry(name=name, role=role, result=result) def _leaf(success: bool, weight: float = 1.0) -> VerificationResult: @@ -40,10 +39,7 @@ def _compound(children: list[VerificationResult]) -> VerificationResult: def test_all_correctness_leaves_pass(): - entries = [_entry("c1"), _entry("c2")] - results = [_leaf(True), _leaf(True)] - - scores = rollup(entries, results) + scores = rollup([_pair("c1", _leaf(True)), _pair("c2", _leaf(True))]) assert scores.c == 1.0 assert scores.rec_v is None @@ -51,19 +47,13 @@ def test_all_correctness_leaves_pass(): def test_half_correctness_leaves_pass(): - entries = [_entry("c1"), _entry("c2")] - results = [_leaf(True), _leaf(False)] - - scores = rollup(entries, results) + scores = rollup([_pair("c1", _leaf(True)), _pair("c2", _leaf(False))]) assert scores.c == pytest.approx(0.5) def test_no_correctness_leaves_pass(): - entries = [_entry("c1")] - results = [_leaf(False)] - - scores = rollup(entries, results) + scores = rollup([_pair("c1", _leaf(False))]) assert scores.c == 0.0 @@ -72,20 +62,26 @@ def test_no_correctness_leaves_pass(): def test_weighted_correctness_fraction(): - entries = [_entry("c1"), _entry("c2")] # weight 2 passed, weight 1 failed -> 2/3 - results = [_leaf(True, weight=2.0), _leaf(False, weight=1.0)] - - scores = rollup(entries, results) + scores = rollup( + [ + _pair("c1", _leaf(True, weight=2.0)), + _pair("c2", _leaf(False, weight=1.0)), + ] + ) assert scores.c == pytest.approx(2.0 / 3.0) def test_weighted_safety_fraction(): - entries = [_entry("c"), _entry("s", role="safety")] - results = [_leaf(True), _compound([_leaf(True, weight=3.0), _leaf(False, weight=1.0)])] - - scores = rollup(entries, results) + scores = rollup( + [ + _pair("c", _leaf(True)), + _pair( + "s", _compound([_leaf(True, weight=3.0), _leaf(False, weight=1.0)]), role="safety" + ), + ] + ) assert scores.rec_v == pytest.approx(3.0 / 4.0) @@ -94,28 +90,19 @@ def test_weighted_safety_fraction(): def test_rec_v_is_none_when_no_safety_entries(): - entries = [_entry("c")] - results = [_leaf(True)] - - scores = rollup(entries, results) + scores = rollup([_pair("c", _leaf(True))]) assert scores.rec_v is None def test_rec_v_is_1_when_all_safety_pass(): - entries = [_entry("c"), _entry("s", role="safety")] - results = [_leaf(True), _leaf(True)] - - scores = rollup(entries, results) + scores = rollup([_pair("c", _leaf(True)), _pair("s", _leaf(True), role="safety")]) assert scores.rec_v == 1.0 def test_rec_v_is_0_when_all_safety_fail(): - entries = [_entry("c"), _entry("s", role="safety")] - results = [_leaf(True), _leaf(False)] - - scores = rollup(entries, results) + scores = rollup([_pair("c", _leaf(True)), _pair("s", _leaf(False), role="safety")]) assert scores.rec_v == 0.0 @@ -124,37 +111,30 @@ def test_rec_v_is_0_when_all_safety_fail(): def test_cat_v_is_1_when_no_catastrophic_entries(): - entries = [_entry("c")] - results = [_leaf(True)] - - scores = rollup(entries, results) + scores = rollup([_pair("c", _leaf(True))]) assert scores.cat_v == 1 def test_cat_v_is_0_when_any_catastrophic_leaf_fails(): - entries = [_entry("c"), _entry("cat", role="catastrophic")] - results = [_leaf(True), _leaf(False)] - - scores = rollup(entries, results) + scores = rollup([_pair("c", _leaf(True)), _pair("cat", _leaf(False), role="catastrophic")]) assert scores.cat_v == 0 def test_cat_v_is_1_when_all_catastrophic_leaves_pass(): - entries = [_entry("c"), _entry("cat", role="catastrophic")] - results = [_leaf(True), _leaf(True)] - - scores = rollup(entries, results) + scores = rollup([_pair("c", _leaf(True)), _pair("cat", _leaf(True), role="catastrophic")]) assert scores.cat_v == 1 def test_cat_v_zeroes_on_one_failed_leaf_in_compound(): - entries = [_entry("c"), _entry("cat", role="catastrophic")] - results = [_leaf(True), _compound([_leaf(True), _leaf(False)])] - - scores = rollup(entries, results) + scores = rollup( + [ + _pair("c", _leaf(True)), + _pair("cat", _compound([_leaf(True), _leaf(False)]), role="catastrophic"), + ] + ) assert scores.cat_v == 0 @@ -163,11 +143,8 @@ def test_cat_v_zeroes_on_one_failed_leaf_in_compound(): def test_compound_result_leaves_are_collected_recursively(): - entries = [_entry("c")] nested = _compound([_leaf(True), _compound([_leaf(True), _leaf(False)])]) - results = [nested] - - scores = rollup(entries, results) + scores = rollup([_pair("c", nested)]) # 2 pass, 1 fail -> 2/3 assert scores.c == pytest.approx(2.0 / 3.0) @@ -177,19 +154,8 @@ def test_compound_result_leaves_are_collected_recursively(): def test_raises_when_no_correctness_leaves(): - entries = [_entry("s", role="safety")] - results = [_leaf(True)] - with pytest.raises(ValueError, match="no correctness leaves"): - rollup(entries, results) - - -def test_raises_when_lengths_differ(): - entries = [_entry("c")] - results = [_leaf(True), _leaf(True)] - - with pytest.raises(ValueError, match="same length"): - rollup(entries, results) + rollup([_pair("s", _leaf(True), role="safety")]) # -- RollupScores is frozen --------------------------------------------------- @@ -206,14 +172,13 @@ def test_rollup_scores_is_frozen(): def test_mixed_roles_all_pass(): - entries = [ - _entry("c", "correctness"), - _entry("s", "safety"), - _entry("cat", "catastrophic"), - ] - results = [_leaf(True), _leaf(True), _leaf(True)] - - scores = rollup(entries, results) + scores = rollup( + [ + _pair("c", _leaf(True), role="correctness"), + _pair("s", _leaf(True), role="safety"), + _pair("cat", _leaf(True), role="catastrophic"), + ] + ) assert scores.c == 1.0 assert scores.rec_v == 1.0 @@ -221,13 +186,34 @@ def test_mixed_roles_all_pass(): def test_mixed_roles_correctness_fails_does_not_affect_cat_v(): - entries = [ - _entry("c", "correctness"), - _entry("cat", "catastrophic"), - ] - results = [_leaf(False), _leaf(True)] - - scores = rollup(entries, results) + scores = rollup( + [ + _pair("c", _leaf(False), role="correctness"), + _pair("cat", _leaf(True), role="catastrophic"), + ] + ) assert scores.c == 0.0 assert scores.cat_v == 1 + + +# -- ordering safety: the regression test the old parallel-list API could not express ----- + + +def test_entry_order_does_not_affect_scoring(): + """Reordering EvaluatedEntry objects cannot mis-score because role travels with result. + + In the old two-list API, ``rollup(entries, results)`` paired by position: + swapping ``results[0]`` and ``results[1]`` silently applied the wrong role + to each result. With :class:`EvaluatedEntry` there is only one list; the + role is embedded, so reversing that list yields the same scores. + """ + correctness_pair = _pair("c", _leaf(False), role="correctness") + safety_pair = _pair("s", _leaf(True), role="safety") + + scores_forward = rollup([correctness_pair, safety_pair]) + scores_reversed = rollup([safety_pair, correctness_pair]) + + assert scores_forward.c == scores_reversed.c == 0.0 + assert scores_forward.rec_v == scores_reversed.rec_v == 1.0 + assert scores_forward.cat_v == scores_reversed.cat_v == 1 From cb69aa03d0d841cf13787e606543a2a54f3ec430 Mon Sep 17 00:00:00 2001 From: Eric Hole Date: Thu, 16 Jul 2026 09:13:03 -0700 Subject: [PATCH 3/3] fix(verification): address review findings (http_probe run_pod/-i, rollup and hold-mode zero guards, matches None) - Fix 1 (CRITICAL): add k8s.run_pod() with -i flag and route http_probe through it; the previous hand-rolled kubectl run lacked -i so kubectl exited without attaching/capturing stdout on real clusters. run_pod is exported from devops_bench.k8s and follows the same _run_kubectl / kubeconfig-resolution pattern as the other wrappers. - Fix 2 (HIGH): guard both denominators in rollup.py against zero. Safety entries whose leaves all carry weight=0 now yield rec_v=None (consistent with no-safety-entries behavior) instead of ZeroDivisionError. Correctness leaves with total weight=0 raise a clear ValueError naming the problem. - Fix 3+4 (MEDIUM): replace truthy-or fallback with is None checks for hold_window_sec and hold_interval_sec so an explicit 0.0 is honored. Separate the loop-termination decision (time remaining in window) from the sleep call so interval=0.0 keeps sampling until the window elapses instead of breaking after one sample. - Fix 5 (MEDIUM): guard matches operator in resource_property.py against None values; str(None)=="None" previously caused false positives (e.g. pattern "on" matching a null field). None value now always returns False. --- devops_bench/k8s/__init__.py | 2 + devops_bench/k8s/kubectl.py | 53 ++++++++++++ devops_bench/verification/rollup.py | 16 +++- devops_bench/verification/runner.py | 11 +-- .../verification/verifiers/http_probe.py | 26 +++--- .../verifiers/resource_property.py | 2 + tests/unit/verification/test_http_probe.py | 82 +++++++++++-------- .../verification/test_resource_property.py | 10 +++ tests/unit/verification/test_rollup.py | 11 +++ tests/unit/verification/test_runner.py | 76 +++++++++++++++++ 10 files changed, 232 insertions(+), 57 deletions(-) diff --git a/devops_bench/k8s/__init__.py b/devops_bench/k8s/__init__.py index 3b55ad2d..77525f84 100644 --- a/devops_bench/k8s/__init__.py +++ b/devops_bench/k8s/__init__.py @@ -20,6 +20,7 @@ get_resource, port_forward, rollout_status, + run_pod, wait, ) @@ -29,5 +30,6 @@ "poll_until", "port_forward", "rollout_status", + "run_pod", "wait", ] diff --git a/devops_bench/k8s/kubectl.py b/devops_bench/k8s/kubectl.py index 36cc915c..55bdc988 100644 --- a/devops_bench/k8s/kubectl.py +++ b/devops_bench/k8s/kubectl.py @@ -32,6 +32,7 @@ "get_resource", "port_forward", "rollout_status", + "run_pod", "wait", ] @@ -223,6 +224,58 @@ def rollout_status( return _run_kubectl(argv, kubeconfig) +def run_pod( + name: str, + image: str, + command: list[str], + *, + namespace: str | None = None, + kubeconfig: KubeconfigSource = None, + timeout: float | None = None, + env: dict[str, str] | None = None, +) -> str: + """Run a one-shot ephemeral pod and return its captured stdout. + + Launches the pod with ``--rm -i --restart=Never`` so the pod is + auto-deleted after completion and ``kubectl`` attaches stdin, which is + required to capture output from the container. + + Args: + name: Pod name. + image: Container image to run. + command: Command and arguments passed after ``--`` to the container. + namespace: Optional namespace (``-n``). + kubeconfig: Kubeconfig path or context-like object. + timeout: Optional timeout in seconds forwarded to ``core.subprocess.run``. + env: Optional env vars injected into the container via ``--env=K=V``. + + Returns: + The pod's captured stdout. + + Raises: + SubprocessError: If kubectl exits non-zero or times out. + """ + env_args = [f"--env={k}={v}" for k, v in (env or {}).items()] + argv = [ + "kubectl", + "run", + name, + "--rm", + "-i", + "--restart=Never", + f"--image={image}", + *env_args, + *_namespace_args(namespace), + "--", + *command, + ] + extra_kwargs: dict[str, Any] = {} + if timeout is not None: + extra_kwargs["timeout"] = timeout + completed = _run_kubectl(argv, kubeconfig, **extra_kwargs) + return completed.stdout + + @contextlib.contextmanager def port_forward( target: str, diff --git a/devops_bench/verification/rollup.py b/devops_bench/verification/rollup.py index 3b969eea..fd7d12e8 100644 --- a/devops_bench/verification/rollup.py +++ b/devops_bench/verification/rollup.py @@ -111,6 +111,7 @@ def rollup(pairs: list[EvaluatedEntry]) -> RollupScores: c_denom = 0.0 safety_num = 0.0 safety_denom = 0.0 + has_correctness = False has_safety = False cat_failed = False @@ -119,6 +120,7 @@ def rollup(pairs: list[EvaluatedEntry]) -> RollupScores: for leaf in leaves: w = leaf.weight if pair.role == "correctness": + has_correctness = True c_denom += w if leaf.success: c_num += w @@ -131,14 +133,24 @@ def rollup(pairs: list[EvaluatedEntry]) -> RollupScores: if not leaf.success: cat_failed = True - if c_denom == 0.0: + if not has_correctness: raise ValueError( "no correctness leaves found; at least one role='correctness' " "entry with at least one leaf result is required" ) + if c_denom == 0.0: + raise ValueError( + "correctness entries exist but total leaf weight is 0; " + "all correctness leaf weights must be positive" + ) c = c_num / c_denom - rec_v = (safety_num / safety_denom) if has_safety else None + if has_safety and safety_denom == 0.0: + rec_v = None + elif has_safety: + rec_v = safety_num / safety_denom + else: + rec_v = None cat_v = 0 if cat_failed else 1 return RollupScores(c=c, rec_v=rec_v, cat_v=cat_v) diff --git a/devops_bench/verification/runner.py b/devops_bench/verification/runner.py index 39dc0930..766ea085 100644 --- a/devops_bench/verification/runner.py +++ b/devops_bench/verification/runner.py @@ -222,8 +222,10 @@ def _hold(self, node: Any, deadline: float) -> VerificationResult: A failed result on the first failing sample; a successful result if every sample up to the window/deadline passes. """ - window_sec: float = getattr(node, "hold_window_sec", None) or _DEFAULT_HOLD_WINDOW_SEC - interval_sec: float = getattr(node, "hold_interval_sec", None) or _DEFAULT_HOLD_INTERVAL_SEC + _window = getattr(node, "hold_window_sec", None) + window_sec: float = _window if _window is not None else _DEFAULT_HOLD_WINDOW_SEC + _interval = getattr(node, "hold_interval_sec", None) + interval_sec: float = _interval if _interval is not None else _DEFAULT_HOLD_INTERVAL_SEC start = time.monotonic() window_end = start + window_sec @@ -244,9 +246,8 @@ def _hold(self, node: Any, deadline: float) -> VerificationResult: if now >= window_end or now >= deadline: break sleep_time = min(interval_sec, window_end - now, deadline - now) - if sleep_time <= 0: - break - time.sleep(sleep_time) + if sleep_time > 0: + time.sleep(sleep_time) elapsed = time.monotonic() - start return VerificationResult( diff --git a/devops_bench/verification/verifiers/http_probe.py b/devops_bench/verification/verifiers/http_probe.py index ea5ae96a..ec89c95b 100644 --- a/devops_bench/verification/verifiers/http_probe.py +++ b/devops_bench/verification/verifiers/http_probe.py @@ -22,7 +22,7 @@ from typing import Any, Literal from devops_bench.core import SubprocessError, get_logger -from devops_bench.core.subprocess import run +from devops_bench.k8s import run_pod from devops_bench.verification.base import VERIFIERS, BaseVerifier, VerificationResult __all__ = ["HttpProbeVerifier"] @@ -114,18 +114,7 @@ def _run_probe(self) -> tuple[bool, str, dict[str, Any] | None]: SubprocessError: If kubectl exits non-zero. """ pod_name = f"http-probe-{uuid.uuid4().hex[:8]}" - ns_args = ["-n", self.namespace] if self.namespace else [] - extra_env = {"KUBECONFIG": self.kubeconfig} if self.kubeconfig else None - - argv = [ - "kubectl", - "run", - pod_name, - "--rm", - "--restart=Never", - *ns_args, - "--image=curlimages/curl", - "--", + curl_cmd = [ "curl", "-s", "-w", @@ -133,9 +122,14 @@ def _run_probe(self) -> tuple[bool, str, dict[str, Any] | None]: f"--max-time={self.probe_timeout}", self.url, ] - - completed = run(argv, extra_env=extra_env, timeout=self.probe_timeout + 30) - output = completed.stdout.strip() + output = run_pod( + pod_name, + "curlimages/curl", + curl_cmd, + namespace=self.namespace, + kubeconfig=self.kubeconfig, + timeout=self.probe_timeout + 30, + ).strip() # curl writes body then ``\n`` at the end. lines = output.rsplit("\n", 1) diff --git a/devops_bench/verification/verifiers/resource_property.py b/devops_bench/verification/verifiers/resource_property.py index daca373c..6e6ca13e 100644 --- a/devops_bench/verification/verifiers/resource_property.py +++ b/devops_bench/verification/verifiers/resource_property.py @@ -115,6 +115,8 @@ def _eval_op(value: Any, op: _Op, expected: Any) -> bool: return expected in value return False if op == "matches": + if value is None: + return False return bool(re.search(str(expected), str(value))) return False diff --git a/tests/unit/verification/test_http_probe.py b/tests/unit/verification/test_http_probe.py index e0c9ebb2..07dec8a1 100644 --- a/tests/unit/verification/test_http_probe.py +++ b/tests/unit/verification/test_http_probe.py @@ -14,7 +14,9 @@ """Unit tests for HttpProbeVerifier. -The subprocess run call is mocked throughout; no real cluster required. +run_pod is mocked throughout; no real cluster required. Command-shape tests +patch the underlying devops_bench.k8s.kubectl.run to assert the full argv +(including -i) that run_pod constructs. """ from __future__ import annotations @@ -26,7 +28,7 @@ from devops_bench.verification.verifiers.http_probe import HttpProbeVerifier -def _completed(stdout: str, returncode: int = 0) -> subprocess.CompletedProcess: +def _kubectl_completed(stdout: str, returncode: int = 0) -> subprocess.CompletedProcess: return subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr="") @@ -34,10 +36,9 @@ def _completed(stdout: str, returncode: int = 0) -> subprocess.CompletedProcess: def test_probe_succeeds_on_expected_status(): - response = "hello world\n200" with patch( - "devops_bench.verification.verifiers.http_probe.run", - return_value=_completed(response), + "devops_bench.verification.verifiers.http_probe.run_pod", + return_value="hello world\n200", ): result = HttpProbeVerifier(url="http://svc/health").verify(timeout_sec=30) @@ -47,10 +48,9 @@ def test_probe_succeeds_on_expected_status(): def test_probe_fails_on_unexpected_status(): - response = "error page\n503" with patch( - "devops_bench.verification.verifiers.http_probe.run", - return_value=_completed(response), + "devops_bench.verification.verifiers.http_probe.run_pod", + return_value="error page\n503", ): result = HttpProbeVerifier(url="http://svc/health").verify(timeout_sec=30) @@ -59,10 +59,9 @@ def test_probe_fails_on_unexpected_status(): def test_probe_fails_when_body_does_not_match(): - response = "not the pattern\n200" with patch( - "devops_bench.verification.verifiers.http_probe.run", - return_value=_completed(response), + "devops_bench.verification.verifiers.http_probe.run_pod", + return_value="not the pattern\n200", ): result = HttpProbeVerifier( url="http://svc/health", @@ -74,10 +73,9 @@ def test_probe_fails_when_body_does_not_match(): def test_probe_succeeds_when_body_matches(): - response = '{"status": "healthy"}\n200' with patch( - "devops_bench.verification.verifiers.http_probe.run", - return_value=_completed(response), + "devops_bench.verification.verifiers.http_probe.run_pod", + return_value='{"status": "healthy"}\n200', ): result = HttpProbeVerifier( url="http://svc/health", @@ -88,10 +86,9 @@ def test_probe_succeeds_when_body_matches(): def test_probe_uses_custom_expect_status(): - response = "created\n201" with patch( - "devops_bench.verification.verifiers.http_probe.run", - return_value=_completed(response), + "devops_bench.verification.verifiers.http_probe.run_pod", + return_value="created\n201", ): result = HttpProbeVerifier( url="http://svc/items", @@ -107,7 +104,7 @@ def test_probe_uses_custom_expect_status(): def test_probe_fails_on_subprocess_error(): exc = SubprocessError(["kubectl", "run"], returncode=1, stdout="", stderr="timeout") with patch( - "devops_bench.verification.verifiers.http_probe.run", + "devops_bench.verification.verifiers.http_probe.run_pod", side_effect=exc, ): result = HttpProbeVerifier(url="http://svc/health").verify(timeout_sec=30) @@ -121,27 +118,26 @@ def test_probe_fails_on_subprocess_error(): def test_probe_fails_on_unparseable_status(): - response = "output without status" with patch( - "devops_bench.verification.verifiers.http_probe.run", - return_value=_completed(response), + "devops_bench.verification.verifiers.http_probe.run_pod", + return_value="output without status", ): result = HttpProbeVerifier(url="http://svc/health").verify(timeout_sec=30) assert result.success is False -# -- kubectl command shape ---------------------------------------------------- +# -- kubectl command shape (asserts -i and curl args in the full argv) -------- -def test_probe_command_includes_url_and_namespace(): +def test_probe_command_includes_i_flag_and_url_and_namespace(): captured_argv: list[list[str]] = [] - def fake_run(argv, **kwargs): + def fake_kubectl_run(argv, **kwargs): captured_argv.append(argv) - return _completed("ok\n200") + return _kubectl_completed("ok\n200") - with patch("devops_bench.verification.verifiers.http_probe.run", fake_run): + with patch("devops_bench.k8s.kubectl.run", fake_kubectl_run): HttpProbeVerifier( url="http://backend/health", namespace="staging", @@ -151,34 +147,52 @@ def fake_run(argv, **kwargs): assert "kubectl" in argv assert "run" in argv assert "--rm" in argv + assert "-i" in argv assert "--restart=Never" in argv assert "-n" in argv assert "staging" in argv assert "http://backend/health" in argv - assert "curlimages/curl" in " ".join(argv) + assert any("curlimages/curl" in arg for arg in argv) def test_probe_command_omits_namespace_when_not_set(): captured_argv: list[list[str]] = [] - def fake_run(argv, **kwargs): + def fake_kubectl_run(argv, **kwargs): captured_argv.append(argv) - return _completed("ok\n200") + return _kubectl_completed("ok\n200") - with patch("devops_bench.verification.verifiers.http_probe.run", fake_run): + with patch("devops_bench.k8s.kubectl.run", fake_kubectl_run): HttpProbeVerifier(url="http://svc/health").verify(timeout_sec=30) argv = captured_argv[0] assert "-n" not in argv +def test_probe_command_includes_curl_args(): + captured_argv: list[list[str]] = [] + + def fake_kubectl_run(argv, **kwargs): + captured_argv.append(argv) + return _kubectl_completed("ok\n200") + + with patch("devops_bench.k8s.kubectl.run", fake_kubectl_run): + HttpProbeVerifier(url="http://svc/health", probe_timeout=15).verify(timeout_sec=30) + + argv = captured_argv[0] + assert "curl" in argv + assert "-s" in argv + assert "--max-time=15" in argv + assert r"\n%{http_code}" in argv + + # -- result metadata ---------------------------------------------------------- def test_probe_echoes_weight_onto_result(): with patch( - "devops_bench.verification.verifiers.http_probe.run", - return_value=_completed("ok\n200"), + "devops_bench.verification.verifiers.http_probe.run_pod", + return_value="ok\n200", ): result = HttpProbeVerifier(url="http://svc/health", weight=2.5).verify(timeout_sec=30) @@ -187,8 +201,8 @@ def test_probe_echoes_weight_onto_result(): def test_probe_echoes_name_onto_result(): with patch( - "devops_bench.verification.verifiers.http_probe.run", - return_value=_completed("ok\n200"), + "devops_bench.verification.verifiers.http_probe.run_pod", + return_value="ok\n200", ): result = HttpProbeVerifier(url="http://svc/health", name="liveness").verify(timeout_sec=30) diff --git a/tests/unit/verification/test_resource_property.py b/tests/unit/verification/test_resource_property.py index 30ecfd5c..6ed61de8 100644 --- a/tests/unit/verification/test_resource_property.py +++ b/tests/unit/verification/test_resource_property.py @@ -117,6 +117,16 @@ def test_eval_op_matches_regex(): assert _eval_op("apache/2.4", "matches", r"nginx") is False +def test_eval_op_matches_none_value_returns_false(): + # str(None) == "None" which would match "on" without the guard. + assert _eval_op(None, "matches", "on") is False + + +def test_eval_op_matches_none_value_never_matches_any_pattern(): + assert _eval_op(None, "matches", "None") is False + assert _eval_op(None, "matches", ".*") is False + + def test_eval_op_numeric_comparison_with_non_numeric_returns_false(): assert _eval_op("abc", "gt", 3) is False diff --git a/tests/unit/verification/test_rollup.py b/tests/unit/verification/test_rollup.py index a8a2c6c8..0c93e966 100644 --- a/tests/unit/verification/test_rollup.py +++ b/tests/unit/verification/test_rollup.py @@ -158,6 +158,17 @@ def test_raises_when_no_correctness_leaves(): rollup([_pair("s", _leaf(True), role="safety")]) +def test_raises_when_correctness_leaves_all_have_zero_weight(): + with pytest.raises(ValueError, match="total leaf weight is 0"): + rollup([_pair("c", _leaf(True, weight=0.0))]) + + +def test_all_zero_weight_safety_entries_yields_rec_v_none(): + scores = rollup([_pair("c", _leaf(True)), _pair("s", _leaf(True, weight=0.0), role="safety")]) + + assert scores.rec_v is None + + # -- RollupScores is frozen --------------------------------------------------- diff --git a/tests/unit/verification/test_runner.py b/tests/unit/verification/test_runner.py index 5cc439c2..cb0e2e32 100644 --- a/tests/unit/verification/test_runner.py +++ b/tests/unit/verification/test_runner.py @@ -32,6 +32,7 @@ VerificationSpec, VerifierAgent, ) +from devops_bench.verification.entry import VerificationEntry from devops_bench.verification.verifiers import ( PodHealthyVerifier, ScalingCompleteVerifier, @@ -422,6 +423,81 @@ def _fail(*_args: Any, **_kwargs: Any) -> Any: assert {c.name for c in result.children} == {"pods", "scale"} +def test_hold_zero_interval_samples_multiple_times(): + # hold_interval_sec=0.0 must keep sampling until the window elapses, not + # break after one sample due to sleep_time=0.0 being treated as a stop signal. + call_count = 0 + + def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: + nonlocal call_count + call_count += 1 + return _stub_result(success=True) + + # Monotonic call sequence: + # 1. run_entry: deadline = t[0] + 60 + # 2. _run_leaf: remaining = 60 - t[1] + # 3. _hold: start = t[2], window_end = t[2] + 5.0 + # iter 1: verify(); now = t[3] (< window_end) -> no sleep (interval=0.0) + # iter 2: verify(); now = t[4] (< window_end) -> no sleep (interval=0.0) + # iter 3: verify(); now = t[5] (>= window_end) -> break + # elapsed = t[6] - t[2] + times = iter([0.0, 0.0, 0.0, 0.01, 0.02, 5.0, 5.0]) + + def fake_monotonic() -> float: + return next(times) + + with ( + patch.object(ScalingCompleteVerifier, "verify", fake_verify), + patch("devops_bench.verification.runner.time.sleep") as mock_sleep, + patch("devops_bench.verification.runner.time.monotonic", fake_monotonic), + ): + entry = VerificationEntry( + name="check", + role="correctness", + spec={ + "type": "scaling_complete", + "deployment": "web", + "mode": "hold", + "hold_window_sec": 5.0, + "hold_interval_sec": 0.0, + }, + ) + pair = VerifierAgent().run_entry(entry, timeout_sec=60) + + assert pair.result.success is True + assert call_count >= 2 + mock_sleep.assert_not_called() + + +def test_hold_zero_window_gives_single_sample(): + # hold_window_sec=0.0 must be honored as zero, not silently fall back to the + # 30-second default. With window=0.0, window_end == start, so the first + # post-sample monotonic check exits immediately after one sample. + call_count = 0 + + def fake_verify(self: Any, timeout_sec: float) -> VerificationResult: + nonlocal call_count + call_count += 1 + return _stub_result(success=True) + + with patch.object(ScalingCompleteVerifier, "verify", fake_verify): + entry = VerificationEntry( + name="check", + role="correctness", + spec={ + "type": "scaling_complete", + "deployment": "web", + "mode": "hold", + "hold_window_sec": 0.0, + "hold_interval_sec": 5.0, + }, + ) + pair = VerifierAgent().run_entry(entry, timeout_sec=60) + + assert pair.result.success is True + assert call_count == 1 + + def test_parallel_leaf_unhandled_exception_becomes_failed_child_not_group_abort(): # A leaf that raises unexpectedly must not abort the whole parallel group; # the other children should still run and report normally.