diff --git a/features/steps/validation_pipeline_output_steps.py b/features/steps/validation_pipeline_output_steps.py new file mode 100644 index 000000000..8da81729f --- /dev/null +++ b/features/steps/validation_pipeline_output_steps.py @@ -0,0 +1,88 @@ +"""Step definitions for validation pipeline output and edge-case scenarios. + +Covers executor output normalisation edge cases (non-bool passed, non-string +message, non-dict data) and stdout/stderr capture steps. +""" + +from __future__ import annotations + +from typing import Any + +from behave import given, then, use_step_matcher + +from cleveragents.application.services.validation_pipeline import ( + ValidationResult, + ValidationSummary, +) + + +def _find_result(summary: ValidationSummary, name: str) -> ValidationResult: + """Find a validation result by name.""" + for r in summary.results: + if r.validation_name == name: + return r + available = [r.validation_name for r in summary.results] + msg = f"No result for '{name}', available: {available}" + raise AssertionError(msg) + + +use_step_matcher("re") + + +@given(r'the vp executor returns non-bool passed for "(?P[^"]+)"') +def step_given_executor_non_bool_passed(context: Any, name: str) -> None: + context.vp_executor.set_non_bool_passed(name) + + +@given(r'the vp executor returns non-string message for "(?P[^"]+)"') +def step_given_executor_non_string_message(context: Any, name: str) -> None: + context.vp_executor.set_non_string_message(name) + + +@given(r'the vp executor returns non-dict data for "(?P[^"]+)"') +def step_given_executor_non_dict_data(context: Any, name: str) -> None: + context.vp_executor.set_non_dict_data(name) + + +@given(r'the vp executor prints to stdout for "(?P[^"]+)"') +def step_given_executor_prints_stdout(context: Any, name: str) -> None: + context.vp_executor.set_prints_stdout(name) + + +@given(r'the vp executor prints to stderr for "(?P[^"]+)"') +def step_given_executor_prints_stderr(context: Any, name: str) -> None: + context.vp_executor.set_prints_stderr(name) + + +@then(r'the vp result for "(?P[^"]+)" has positive duration_ms') +def step_then_result_positive_duration(context: Any, name: str) -> None: + assert context.vp_summary is not None + result = _find_result(context.vp_summary, name) + assert result.duration_ms > 0, ( + f"Expected duration_ms > 0 for '{name}', got {result.duration_ms}" + ) + +@then(r"the plan metadata contains validation_summary") +def step_then_plan_metadata_has_summary(context: Any) -> None: + assert context.vp_plan_metadata is not None + assert "validation_summary" in context.vp_plan_metadata + + +@then(r"the plan metadata validation_summary total is (?P\d+)") +def step_then_plan_metadata_total(context: Any, count: str) -> None: + cnt = int(count) + assert context.vp_plan_metadata is not None + vs = context.vp_plan_metadata["validation_summary"] + assert vs["total"] == cnt, f"Expected total={cnt}, got {vs['total']}" + + +@then(r"the summary all_required_passed property is true") +def step_then_summary_property_true(context: Any) -> None: + assert context.vp_summary is not None + assert context.vp_summary.all_required_passed + + +@then(r"the summary all_required_passed property is false") +def step_then_summary_property_false(context: Any) -> None: + assert context.vp_summary is not None + assert not context.vp_summary.all_required_passed diff --git a/features/steps/validation_pipeline_steps.py b/features/steps/validation_pipeline_steps.py index 5423605aa..17162df26 100644 --- a/features/steps/validation_pipeline_steps.py +++ b/features/steps/validation_pipeline_steps.py @@ -469,65 +469,8 @@ def step_then_group_result_count(context: Any, resource: str, count: str) -> Non assert actual == cnt, f"Expected {cnt} results for '{resource}', got {actual}" -@then(r"the summary all_required_passed property is true") -def step_then_summary_property_true(context: Any) -> None: - assert context.vp_summary is not None - assert context.vp_summary.all_required_passed -@then(r"the summary all_required_passed property is false") -def step_then_summary_property_false(context: Any) -> None: - assert context.vp_summary is not None - assert not context.vp_summary.all_required_passed - - -@then(r"the plan metadata contains validation_summary") -def step_then_plan_metadata_has_summary(context: Any) -> None: - assert context.vp_plan_metadata is not None - assert "validation_summary" in context.vp_plan_metadata - - -@then(r"the plan metadata validation_summary total is (?P\d+)") -def step_then_plan_metadata_total(context: Any, count: str) -> None: - cnt = int(count) - assert context.vp_plan_metadata is not None - vs = context.vp_plan_metadata["validation_summary"] - assert vs["total"] == cnt, f"Expected total={cnt}, got {vs['total']}" - - -@given(r'the vp executor returns non-bool passed for "(?P[^"]+)"') -def step_given_executor_non_bool_passed(context: Any, name: str) -> None: - context.vp_executor.set_non_bool_passed(name) - - -@given(r'the vp executor returns non-string message for "(?P[^"]+)"') -def step_given_executor_non_string_message(context: Any, name: str) -> None: - context.vp_executor.set_non_string_message(name) - - -@given(r'the vp executor returns non-dict data for "(?P[^"]+)"') -def step_given_executor_non_dict_data(context: Any, name: str) -> None: - context.vp_executor.set_non_dict_data(name) - - -@given(r'the vp executor prints to stdout for "(?P[^"]+)"') -def step_given_executor_prints_stdout(context: Any, name: str) -> None: - context.vp_executor.set_prints_stdout(name) - - -@given(r'the vp executor prints to stderr for "(?P[^"]+)"') -def step_given_executor_prints_stderr(context: Any, name: str) -> None: - context.vp_executor.set_prints_stderr(name) - - -@then(r'the vp result for "(?P[^"]+)" has positive duration_ms') -def step_then_result_positive_duration(context: Any, name: str) -> None: - assert context.vp_summary is not None - result = _find_result(context.vp_summary, name) - assert result.duration_ms > 0, ( - f"Expected duration_ms > 0 for '{name}', got {result.duration_ms}" - ) - # Reset to default parse matcher use_step_matcher("parse") @@ -538,7 +481,6 @@ use_step_matcher("parse") # --------------------------------------------------------------------------- - def _find_result(summary: ValidationSummary, name: str) -> ValidationResult: """Find a validation result by name.""" for r in summary.results: diff --git a/src/cleveragents/application/services/validation_pipeline.py b/src/cleveragents/application/services/validation_pipeline.py index cf0f59e7e..4019e3f53 100644 --- a/src/cleveragents/application/services/validation_pipeline.py +++ b/src/cleveragents/application/services/validation_pipeline.py @@ -26,7 +26,7 @@ import time from collections import defaultdict from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor -from typing import TYPE_CHECKING, Any, Protocol +from typing import TYPE_CHECKING, Any, Protocol, TextIO, cast from pydantic import BaseModel, ConfigDict, Field @@ -52,15 +52,16 @@ class _ThreadLocalStream: Threads that call ``start_capture()`` collect writes in a per-thread ``StringIO`` buffer until ``stop_capture()`` returns the captured text. + + Assigned to ``sys.stdout``/``sys.stderr`` via ``typing.cast`` so that + Pyright accepts the assignment without requiring ``io.TextIOBase`` + inheritance (which conflicts with the ``encoding`` slot on Python 3.13). """ def __init__(self, original: Any) -> None: self._original = original self._local = threading.local() - - @property - def encoding(self) -> str: # matches io.TextIOBase.encoding signature - return getattr(self._original, "encoding", "utf-8") + self.encoding: str = getattr(original, "encoding", "utf-8") def writable(self) -> bool: return True @@ -143,8 +144,8 @@ def _install_thread_local_streams() -> None: "initialisation" ) - sys.stdout = _STREAM_STDOUT_WRAPPER - sys.stderr = _STREAM_STDERR_WRAPPER + sys.stdout = cast(TextIO, _STREAM_STDOUT_WRAPPER) + sys.stderr = cast(TextIO, _STREAM_STDERR_WRAPPER) _STREAM_PATCH_COUNT += 1