"""Step definitions for validation pipeline scenarios. All step names are prefixed with 'validation pipeline' context to avoid AmbiguousStep collisions with existing step definitions. Uses the ``re`` step matcher for patterns that have optional trailing parts. """ from __future__ import annotations import sys import time from typing import Any from behave import given, then, use_step_matcher, when from cleveragents.application.services.validation_pipeline import ( ValidationCommand, ValidationPipeline, ValidationResult, ValidationSummary, ) from cleveragents.domain.models.core.tool import ValidationMode # --------------------------------------------------------------------------- # Mock executor # --------------------------------------------------------------------------- class MockValidationExecutor: """Configurable mock executor for testing the validation pipeline.""" def __init__(self) -> None: self._results: dict[str, dict[str, Any]] = {} self._timeout_names: set[str] = set() self._exception_names: dict[str, Exception] = {} self._non_dict_names: set[str] = set() self._missing_passed_names: set[str] = set() self._default_passed: bool = False self._stdout_names: set[str] = set() self._stderr_names: set[str] = set() def set_passed(self, name: str) -> None: self._results[name] = {"passed": True, "message": f"{name} passed"} def set_passed_with_data(self, name: str) -> None: self._results[name] = { "passed": True, "message": f"{name} passed with data", "data": {"detail": "some-value", "count": 42}, } def set_failed(self, name: str, message: str) -> None: self._results[name] = {"passed": False, "message": message} def set_timeout(self, name: str) -> None: self._timeout_names.add(name) def set_exception(self, name: str, exc: Exception) -> None: self._exception_names[name] = exc def set_non_dict(self, name: str) -> None: self._non_dict_names.add(name) def set_missing_passed(self, name: str) -> None: self._missing_passed_names.add(name) def set_default_passed(self) -> None: self._default_passed = True def set_non_bool_passed(self, name: str) -> None: self._results[name] = {"passed": "yes", "message": f"{name} passed"} def set_non_string_message(self, name: str) -> None: self._results[name] = {"passed": True, "message": 12345} def set_non_dict_data(self, name: str) -> None: self._results[name] = {"passed": True, "message": f"{name} ok", "data": [1, 2]} def set_prints_stdout(self, name: str) -> None: self._results[name] = {"passed": True, "message": f"{name} ok"} self._stdout_names.add(name) def set_prints_stderr(self, name: str) -> None: self._results[name] = {"passed": True, "message": f"{name} ok"} self._stderr_names.add(name) def __call__( self, validation_name: str, arguments: dict[str, Any] ) -> dict[str, Any]: if validation_name in self._timeout_names: # Sleep longer than the test timeout (0.2 s) but not excessively. # Use _original_sleep to bypass the fast-sleep test patch. _real_sleep = getattr(time, "_original_sleep", time.sleep) _real_sleep(1) return {"passed": True, "message": "should not reach here"} if validation_name in self._exception_names: raise self._exception_names[validation_name] if validation_name in self._non_dict_names: return "not-a-dict" # type: ignore[return-value] if validation_name in self._missing_passed_names: return {"message": "no passed key here"} if validation_name in self._stdout_names: print(f"stdout from {validation_name}") if validation_name in self._stderr_names: print(f"stderr from {validation_name}", file=sys.stderr) if validation_name in self._results: return self._results[validation_name] if self._default_passed: return {"passed": True, "message": f"{validation_name} passed (default)"} return {"passed": False, "message": f"{validation_name}: no result configured"} # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _RESOURCE_ID_COUNTER: int = 0 def _next_resource_id() -> str: global _RESOURCE_ID_COUNTER _RESOURCE_ID_COUNTER += 1 return f"R{_RESOURCE_ID_COUNTER:05d}" # --------------------------------------------------------------------------- # Given steps — use "re" matcher for the required/informational patterns # --------------------------------------------------------------------------- use_step_matcher("re") @given(r"a validation pipeline test environment") def step_given_pipeline_env(context: Any) -> None: context.vp_commands: list[ValidationCommand] = [] context.vp_executor = MockValidationExecutor() context.vp_max_workers: int = 4 context.vp_read_only_resources: set[str] = set() context.vp_summary: ValidationSummary | None = None context.vp_plan_metadata: dict[str, Any] | None = None context.vp_resource_id_map: dict[str, str] = {} global _RESOURCE_ID_COUNTER _RESOURCE_ID_COUNTER = 0 @given(r"an empty list of validation commands") def step_given_empty_commands(context: Any) -> None: context.vp_commands = [] @given( r'a required vp command "(?P[^"]+)" on resource "(?P[^"]+)"' r" with timeout (?P[\d.]+)" ) def step_given_required_command_timeout( context: Any, name: str, resource: str, timeout: str ) -> None: rid = context.vp_resource_id_map.get(resource, _next_resource_id()) context.vp_resource_id_map[resource] = rid context.vp_commands.append( ValidationCommand( validation_name=name, resource_id=rid, resource_name=resource, mode=ValidationMode.REQUIRED, arguments={}, timeout_seconds=float(timeout), ) ) @given( r'a required vp command "(?P[^"]+)" on resource "(?P[^"]+)"' r' having resource_id "(?P[^"]+)"' ) def step_given_required_command_with_rid( context: Any, name: str, resource: str, rid: str ) -> None: context.vp_resource_id_map[resource] = rid context.vp_commands.append( ValidationCommand( validation_name=name, resource_id=rid, resource_name=resource, mode=ValidationMode.REQUIRED, arguments={}, ) ) @given(r'a required vp command "(?P[^"]+)" on resource "(?P[^"]+)"') def step_given_required_command(context: Any, name: str, resource: str) -> None: rid = context.vp_resource_id_map.get(resource, _next_resource_id()) context.vp_resource_id_map[resource] = rid context.vp_commands.append( ValidationCommand( validation_name=name, resource_id=rid, resource_name=resource, mode=ValidationMode.REQUIRED, arguments={}, ) ) @given( r'an informational vp command "(?P[^"]+)"' r' on resource "(?P[^"]+)"' ) def step_given_informational_command(context: Any, name: str, resource: str) -> None: rid = context.vp_resource_id_map.get(resource, _next_resource_id()) context.vp_resource_id_map[resource] = rid context.vp_commands.append( ValidationCommand( validation_name=name, resource_id=rid, resource_name=resource, mode=ValidationMode.INFORMATIONAL, arguments={}, ) ) @given(r'the vp executor returns passed for "(?P[^"]+)"') def step_given_executor_passed(context: Any, name: str) -> None: context.vp_executor.set_passed(name) @given( r'the vp executor returns failed for "(?P[^"]+)"' r' with message "(?P[^"]+)"' ) def step_given_executor_failed(context: Any, name: str, message: str) -> None: context.vp_executor.set_failed(name, message) @given(r"the vp executor returns passed for all validations") def step_given_executor_all_passed(context: Any) -> None: context.vp_executor.set_default_passed() @given(r'the vp executor will timeout for "(?P[^"]+)"') def step_given_executor_timeout(context: Any, name: str) -> None: context.vp_executor.set_timeout(name) @given(r'the vp executor returns non-dict output for "(?P[^"]+)"') def step_given_executor_non_dict(context: Any, name: str) -> None: context.vp_executor.set_non_dict(name) @given(r'the vp executor returns dict without passed key for "(?P[^"]+)"') def step_given_executor_missing_passed(context: Any, name: str) -> None: context.vp_executor.set_missing_passed(name) @given(r'the vp executor raises an exception for "(?P[^"]+)"') def step_given_executor_exception(context: Any, name: str) -> None: context.vp_executor.set_exception(name, RuntimeError("Test exception")) @given(r'the vp executor returns passed with data for "(?P[^"]+)"') def step_given_executor_passed_with_data(context: Any, name: str) -> None: context.vp_executor.set_passed_with_data(name) @given(r'the vp resource "(?P[^"]+)" is marked as read-only') def step_given_read_only_resource(context: Any, rid: str) -> None: context.vp_read_only_resources.add(rid) @given(r"a validation summary with (?P\d+) required failures") def step_given_summary_with_failures(context: Any, count: str) -> None: cnt = int(count) context.vp_summary = ValidationSummary( total=cnt + 1, required_passed=1, required_failed=cnt, informational_passed=0, informational_failed=0, results=[], ) @given(r"the pipeline max_workers is set to (?P\d+)") def step_given_max_workers(context: Any, workers: str) -> None: context.vp_max_workers = int(workers) # --------------------------------------------------------------------------- # When steps # --------------------------------------------------------------------------- @when(r"the validation pipeline runs for plan with metadata") def step_when_pipeline_runs_for_plan(context: Any) -> None: context.vp_plan_metadata = {} pipeline = ValidationPipeline( commands=context.vp_commands, executor=context.vp_executor, max_workers=context.vp_max_workers, read_only_resources=context.vp_read_only_resources, ) context.vp_summary = pipeline.run_for_plan(plan_metadata=context.vp_plan_metadata) @when(r"the validation pipeline runs") def step_when_pipeline_runs(context: Any) -> None: pipeline = ValidationPipeline( commands=context.vp_commands, executor=context.vp_executor, max_workers=context.vp_max_workers, read_only_resources=context.vp_read_only_resources, ) context.vp_summary = pipeline.run() # --------------------------------------------------------------------------- # Then steps # --------------------------------------------------------------------------- @then(r"the validation summary total is (?P\d+)") def step_then_summary_total(context: Any, count: str) -> None: cnt = int(count) assert context.vp_summary is not None, "No summary available" assert context.vp_summary.total == cnt, ( f"Expected total={cnt}, got {context.vp_summary.total}" ) @then(r"the validation summary required_passed is (?P\d+)") def step_then_summary_required_passed(context: Any, count: str) -> None: cnt = int(count) assert context.vp_summary is not None assert context.vp_summary.required_passed == cnt, ( f"Expected required_passed={cnt}, got {context.vp_summary.required_passed}" ) @then(r"the validation summary required_failed is (?P\d+)") def step_then_summary_required_failed(context: Any, count: str) -> None: cnt = int(count) assert context.vp_summary is not None assert context.vp_summary.required_failed == cnt, ( f"Expected required_failed={cnt}, got {context.vp_summary.required_failed}" ) @then(r"the validation summary informational_passed is (?P\d+)") def step_then_summary_informational_passed(context: Any, count: str) -> None: cnt = int(count) assert context.vp_summary is not None assert context.vp_summary.informational_passed == cnt, ( f"Expected informational_passed={cnt}, " f"got {context.vp_summary.informational_passed}" ) @then(r"the validation summary informational_failed is (?P\d+)") def step_then_summary_informational_failed(context: Any, count: str) -> None: cnt = int(count) assert context.vp_summary is not None assert context.vp_summary.informational_failed == cnt, ( f"Expected informational_failed={cnt}, " f"got {context.vp_summary.informational_failed}" ) @then(r"all required validations passed") def step_then_all_required_passed(context: Any) -> None: assert context.vp_summary is not None assert context.vp_summary.all_required_passed, ( f"Expected all_required_passed=True, got False " f"(required_failed={context.vp_summary.required_failed})" ) @then(r"not all required validations passed") def step_then_not_all_required_passed(context: Any) -> None: assert context.vp_summary is not None assert not context.vp_summary.all_required_passed, ( "Expected all_required_passed=False, got True" ) @then(r"the validation results are sorted by resource then mode then name") def step_then_results_sorted(context: Any) -> None: assert context.vp_summary is not None results = context.vp_summary.results keys = [(r.resource_name, r.mode.value, r.validation_name) for r in results] assert keys == sorted(keys), f"Results not sorted: {keys}" @then(r'the vp result for "(?P[^"]+)" has timed_out true') def step_then_result_timed_out(context: Any, name: str) -> None: assert context.vp_summary is not None result = _find_result(context.vp_summary, name) assert result.timed_out, f"Expected timed_out=True for '{name}'" @then(r'the vp result for "(?P[^"]+)" has passed false') def step_then_result_passed_false(context: Any, name: str) -> None: assert context.vp_summary is not None result = _find_result(context.vp_summary, name) assert not result.passed, f"Expected passed=False for '{name}'" @then(r'the vp result for "(?P[^"]+)" has passed true') def step_then_result_passed_true(context: Any, name: str) -> None: assert context.vp_summary is not None result = _find_result(context.vp_summary, name) assert result.passed, f"Expected passed=True for '{name}'" @then(r'the vp result for "(?P[^"]+)" has error containing "(?P[^"]+)"') def step_then_result_error_contains(context: Any, name: str, text: str) -> None: assert context.vp_summary is not None result = _find_result(context.vp_summary, name) assert result.error is not None, f"Expected error for '{name}', got None" assert text in result.error, ( f"Expected error containing '{text}', got: {result.error}" ) @then(r'the vp result for "(?P[^"]+)" has message containing "(?P[^"]+)"') def step_then_result_message_contains(context: Any, name: str, text: str) -> None: assert context.vp_summary is not None result = _find_result(context.vp_summary, name) assert text in result.message, ( f"Expected message containing '{text}', got: {result.message}" ) @then(r'the vp result for "(?P[^"]+)" has data key "(?P[^"]+)"') def step_then_result_has_data_key(context: Any, name: str, key: str) -> None: assert context.vp_summary is not None result = _find_result(context.vp_summary, name) assert result.data is not None, f"Expected data for '{name}', got None" assert key in result.data, ( f"Expected data key '{key}' in {list(result.data.keys())}" ) @then(r"the results grouped by resource have (?P\d+) groups") def step_then_groups_count(context: Any, count: str) -> None: cnt = int(count) assert context.vp_summary is not None groups = ValidationPipeline.group_by_resource(context.vp_summary.results) assert len(groups) == cnt, f"Expected {cnt} groups, got {len(groups)}" @then(r'the vp resource group "(?P[^"]+)" has (?P\d+) result') def step_then_group_result_count(context: Any, resource: str, count: str) -> None: cnt = int(count) assert context.vp_summary is not None groups = ValidationPipeline.group_by_resource(context.vp_summary.results) assert resource in groups, f"Resource '{resource}' not in groups" actual = len(groups[resource]) 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") # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- 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)