forked from cleveragents/cleveragents-core
freemo
4ad51561fc
Cap time.sleep and asyncio.sleep globally at 10ms in before_all to eliminate retry/backoff waits (tenacity wait_fixed, retry_auto_debug exponential backoff). Save originals as time._original_sleep and asyncio._original_sleep for tests that need real wall-clock delays (CircuitBreaker recovery, debounce timers, validation timeouts). Enhance template-DB patch: add "db." prefix to _SCENARIO_DB_PREFIXES and check db_path.stat().st_size > 0 so 0-byte auto-created SQLite files receive the template copy instead of falling through to real migrations. Replace subprocess.run CLI invocations with typer.testing.CliRunner in module_coverage, main_coverage_complete, and coverage_extras step files, eliminating ~6s Python cold-start overhead per call. Switch plan_persistence and action_persistence to in-memory SQLite by default; only cross-restart scenarios use file-based via an explicit Given step. Replace MigrationRunner.run_migrations with Base.metadata.create_all in plan_service_steps for freshly created in-memory engines. Removed leftover debug comment in environment.py. Total tier runtime: 565s -> 21s (96% reduction), 20 features all under 5s behave-internal time. ISSUES CLOSED: #480
548 lines
20 KiB
Python
548 lines
20 KiB
Python
"""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<name>[^"]+)" on resource "(?P<resource>[^"]+)"'
|
|
r" with timeout (?P<timeout>[\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<name>[^"]+)" on resource "(?P<resource>[^"]+)"'
|
|
r' having resource_id "(?P<rid>[^"]+)"'
|
|
)
|
|
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<name>[^"]+)" on resource "(?P<resource>[^"]+)"')
|
|
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<name>[^"]+)"'
|
|
r' on resource "(?P<resource>[^"]+)"'
|
|
)
|
|
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<name>[^"]+)"')
|
|
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<name>[^"]+)"'
|
|
r' with message "(?P<message>[^"]+)"'
|
|
)
|
|
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<name>[^"]+)"')
|
|
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<name>[^"]+)"')
|
|
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<name>[^"]+)"')
|
|
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<name>[^"]+)"')
|
|
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<name>[^"]+)"')
|
|
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<rid>[^"]+)" 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<count>\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<workers>\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<count>\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<count>\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<count>\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<count>\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<count>\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<name>[^"]+)" 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<name>[^"]+)" 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<name>[^"]+)" 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<name>[^"]+)" has error containing "(?P<text>[^"]+)"')
|
|
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<name>[^"]+)" has message containing "(?P<text>[^"]+)"')
|
|
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<name>[^"]+)" has data key "(?P<key>[^"]+)"')
|
|
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<count>\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<resource>[^"]+)" has (?P<count>\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<count>\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<name>[^"]+)"')
|
|
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<name>[^"]+)"')
|
|
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<name>[^"]+)"')
|
|
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<name>[^"]+)"')
|
|
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<name>[^"]+)"')
|
|
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<name>[^"]+)" 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)
|