forked from HAL9000/cleveragents-core
freemo
051ee7c290
Added 52 new .feature files and corresponding _steps.py files targeting previously uncovered code paths in the following areas: - TUI layer: app, commands, persona (state/schema/registry), widgets, input (shell_exec, reference_parser) - Application services: plan lifecycle/service/executor, session, project, repo indexing, correction, checkpoint, actor, llm_actors, strategy coordinator, resource file watcher, service retry wiring - CLI commands: session, resource, repl, plan, db, automation_profile - Domain models: retry_policy, resource_type, cost_budget, docker_compose_analyzer, detail_level, _sql_string_aware, _postgresql_helpers - Core: circuit_breaker, retry_service_patterns - Infrastructure: repositories, transaction_sandbox, strategy_registry, plugins/loader, container - Config: settings - Agents: plan_generation, context_analysis, auto_debug - A2A: facade All new tests follow the Behave/Gherkin BDD standard. Resolved step definition collisions with unique prefixes. Fixed Alembic fileConfig logger disabling issue (disable_existing_loggers=False). ISSUES CLOSED: #1068
447 lines
18 KiB
Python
447 lines
18 KiB
Python
"""Step definitions for retry_policy_model_coverage.feature.
|
|
|
|
These steps target specific uncovered lines in retry_policy.py:
|
|
- Lines 228-232: CircuitBreakerConfig._check_cooldown_le_recovery validator
|
|
- Lines 268-269: ServiceRetryPolicy._reject_unicode_control_chars validator
|
|
- Lines 556-560: apply_overrides unknown keys warning
|
|
- Lines 566-572: apply_overrides invalid service name skip
|
|
- Lines 579-581: apply_overrides non-dict sub-key warning
|
|
- Lines 600-607: apply_overrides final ValidationError skip
|
|
- Lines 546-552: apply_overrides non-dict override data
|
|
"""
|
|
|
|
from behave import given, then, when
|
|
from pydantic import ValidationError
|
|
|
|
from cleveragents.domain.models.core.retry_policy import (
|
|
CircuitBreakerConfig,
|
|
RetryCategory,
|
|
RetryPolicyConfig,
|
|
ServiceRetryPolicy,
|
|
ServiceRetryPolicyRegistry,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
@given("the retry policy model module is imported")
|
|
def step_module_imported(context):
|
|
"""Ensure the retry_policy module is importable."""
|
|
assert CircuitBreakerConfig is not None
|
|
assert ServiceRetryPolicy is not None
|
|
assert ServiceRetryPolicyRegistry is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CircuitBreakerConfig: cooldown > recovery (lines 228-232)
|
|
# ---------------------------------------------------------------------------
|
|
@when(
|
|
"I create a CircuitBreakerConfig with cooldown {cooldown:d} and recovery timeout {recovery:d}"
|
|
)
|
|
def step_create_cb_config(context, cooldown, recovery):
|
|
"""Attempt to create a CircuitBreakerConfig with given cooldown and recovery."""
|
|
context.cb_error = None
|
|
context.cb_config = None
|
|
try:
|
|
context.cb_config = CircuitBreakerConfig(
|
|
cooldown_seconds=float(cooldown),
|
|
recovery_timeout=float(recovery),
|
|
)
|
|
except (ValidationError, ValueError) as exc:
|
|
context.cb_error = exc
|
|
|
|
|
|
@then("a ValueError should be raised mentioning cooldown and recovery_timeout")
|
|
def step_verify_cooldown_error(context):
|
|
"""Verify the validator rejects cooldown > recovery."""
|
|
assert context.cb_error is not None, "Expected an error but none was raised"
|
|
error_text = str(context.cb_error)
|
|
assert "cooldown_seconds" in error_text or "cooldown" in error_text, (
|
|
f"Error should mention cooldown: {error_text}"
|
|
)
|
|
|
|
|
|
@then("the CircuitBreakerConfig should be created successfully")
|
|
def step_verify_cb_success(context):
|
|
"""Verify the config was created without error."""
|
|
assert context.cb_error is None, f"Unexpected error: {context.cb_error}"
|
|
assert context.cb_config is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ServiceRetryPolicy: Unicode control chars (lines 268-269)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I create a ServiceRetryPolicy with a zero-width-space in the name")
|
|
def step_create_policy_zwsp(context):
|
|
"""Attempt to create a policy with a zero-width space (U+200B)."""
|
|
context.unicode_error = None
|
|
try:
|
|
ServiceRetryPolicy(service_name="my\u200bservice")
|
|
except (ValidationError, ValueError) as exc:
|
|
context.unicode_error = exc
|
|
|
|
|
|
@when("I create a ServiceRetryPolicy with an RTL override character in the name")
|
|
def step_create_policy_rtl(context):
|
|
"""Attempt to create a policy with an RTL override (U+202E)."""
|
|
context.unicode_error = None
|
|
try:
|
|
ServiceRetryPolicy(service_name="my\u202eservice")
|
|
except (ValidationError, ValueError) as exc:
|
|
context.unicode_error = exc
|
|
|
|
|
|
@then("a ValidationError should be raised mentioning invisible Unicode")
|
|
def step_verify_unicode_error(context):
|
|
"""Verify the validator rejects invisible Unicode characters."""
|
|
assert context.unicode_error is not None, (
|
|
"Expected a ValidationError for invisible Unicode chars"
|
|
)
|
|
error_text = str(context.unicode_error)
|
|
assert (
|
|
"invisible" in error_text.lower()
|
|
or "unicode" in error_text.lower()
|
|
or "control" in error_text.lower()
|
|
), f"Error should mention invisible/unicode/control: {error_text}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fresh registry helper
|
|
# ---------------------------------------------------------------------------
|
|
@given("a fresh ServiceRetryPolicyRegistry")
|
|
def step_fresh_registry(context):
|
|
"""Create a new registry for the scenario."""
|
|
context.registry = ServiceRetryPolicyRegistry()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# apply_overrides: unknown keys (lines 556-560)
|
|
# ---------------------------------------------------------------------------
|
|
@when('I apply overrides with unknown keys for service "plan_service"')
|
|
def step_apply_unknown_keys(context):
|
|
"""Apply overrides containing unrecognised top-level keys."""
|
|
context.registry.apply_overrides(
|
|
{
|
|
"plan_service": {
|
|
"bogus_key": True,
|
|
"another_unknown": 42,
|
|
}
|
|
}
|
|
)
|
|
|
|
|
|
@then("the unknown keys should be silently ignored")
|
|
def step_verify_unknown_keys_ignored(context):
|
|
"""Verify the override did not crash despite unknown keys."""
|
|
# The registry should still function
|
|
policy = context.registry.get("plan_service")
|
|
assert policy is not None
|
|
|
|
|
|
@then('the policy for "plan_service" should still be valid')
|
|
def step_verify_plan_service_valid(context):
|
|
"""Verify the plan_service policy is intact."""
|
|
policy = context.registry.get("plan_service")
|
|
assert policy.service_name == "plan_service"
|
|
assert policy.retry is not None
|
|
assert policy.circuit_breaker is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# apply_overrides: invalid service name — empty (lines 566-572)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I apply overrides with an empty-string service name")
|
|
def step_apply_empty_service_name(context):
|
|
"""Apply overrides with an empty string as the service name key."""
|
|
context.apply_error = None
|
|
try:
|
|
context.registry.apply_overrides({"": {"retry": {"max_attempts": 5}}})
|
|
except Exception as exc:
|
|
context.apply_error = exc
|
|
|
|
|
|
@when("I apply overrides with a whitespace-only service name")
|
|
def step_apply_whitespace_service_name(context):
|
|
"""Apply overrides with a whitespace-only service name key.
|
|
|
|
After stripping, the name is empty, triggering a validation error
|
|
which is caught and skipped by the except block (lines 566-572).
|
|
"""
|
|
context.apply_error = None
|
|
try:
|
|
context.registry.apply_overrides({" ": {"retry": {"max_attempts": 5}}})
|
|
except Exception as exc:
|
|
context.apply_error = exc
|
|
|
|
|
|
@then("the override should be skipped without crashing")
|
|
def step_verify_no_crash(context):
|
|
"""Verify no exception propagated from apply_overrides."""
|
|
assert context.apply_error is None, f"Expected no error, got: {context.apply_error}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# apply_overrides: non-dict sub-key (lines 579-581)
|
|
# ---------------------------------------------------------------------------
|
|
@when('I apply overrides where "retry" value is a string instead of dict')
|
|
def step_apply_non_dict_retry(context):
|
|
"""Apply overrides where the 'retry' value is not a dict."""
|
|
context.registry.apply_overrides({"plan_service": {"retry": "not_a_dict"}})
|
|
|
|
|
|
@then("the non-dict retry sub-key should be ignored")
|
|
def step_verify_non_dict_retry_ignored(context):
|
|
"""Verify the non-dict retry value was ignored."""
|
|
policy = context.registry.get("plan_service")
|
|
# The retry config should be unchanged from defaults
|
|
assert policy.retry is not None
|
|
assert isinstance(policy.retry, RetryPolicyConfig)
|
|
|
|
|
|
@when('I apply overrides where "circuit_breaker" value is an integer')
|
|
def step_apply_non_dict_cb(context):
|
|
"""Apply overrides where the 'circuit_breaker' value is an int."""
|
|
context.registry.apply_overrides({"plan_service": {"circuit_breaker": 999}})
|
|
|
|
|
|
@then("the non-dict circuit_breaker sub-key should be ignored")
|
|
def step_verify_non_dict_cb_ignored(context):
|
|
"""Verify the non-dict circuit_breaker value was ignored."""
|
|
policy = context.registry.get("plan_service")
|
|
assert policy.circuit_breaker is not None
|
|
assert isinstance(policy.circuit_breaker, CircuitBreakerConfig)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# apply_overrides: final ValidationError from invalid merged policy
|
|
# (lines 600-607)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I apply overrides with an invalid retry max_attempts of -5")
|
|
def step_apply_invalid_max_attempts(context):
|
|
"""Apply overrides that produce an invalid merged policy.
|
|
|
|
max_attempts has ge=1 constraint so -5 will fail validation on the
|
|
final ServiceRetryPolicy.model_validate() call (lines 595-607).
|
|
"""
|
|
context.original_policy = context.registry.get("plan_service")
|
|
context.apply_error = None
|
|
try:
|
|
context.registry.apply_overrides(
|
|
{"plan_service": {"retry": {"max_attempts": -5}}}
|
|
)
|
|
except Exception as exc:
|
|
context.apply_error = exc
|
|
|
|
|
|
@then("the invalid merged override should be skipped")
|
|
def step_verify_invalid_merged_skipped(context):
|
|
"""Verify the override did not crash and was silently skipped."""
|
|
assert context.apply_error is None, (
|
|
f"Expected no exception, got: {context.apply_error}"
|
|
)
|
|
|
|
|
|
@then('the policy for "plan_service" should retain its original max_attempts')
|
|
def step_verify_original_max_attempts(context):
|
|
"""Verify the policy was not corrupted by the invalid override."""
|
|
policy = context.registry.get("plan_service")
|
|
assert policy.retry.max_attempts == context.original_policy.retry.max_attempts
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# apply_overrides: non-dict override_data (lines 546-552)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I apply overrides where the override data is a string")
|
|
def step_apply_non_dict_override_data(context):
|
|
"""Apply overrides where the value for a service is not a dict."""
|
|
context.apply_error = None
|
|
try:
|
|
context.registry.apply_overrides({"plan_service": "this_is_not_a_dict"})
|
|
except Exception as exc:
|
|
context.apply_error = exc
|
|
|
|
|
|
@then("the non-dict override data should be ignored")
|
|
def step_verify_non_dict_data_ignored(context):
|
|
"""Verify the non-dict data was ignored without crashing."""
|
|
assert context.apply_error is None, f"Expected no error, got: {context.apply_error}"
|
|
# The policy should still exist and be valid
|
|
policy = context.registry.get("plan_service")
|
|
assert policy.service_name == "plan_service"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Combined: unknown keys + valid description change
|
|
# ---------------------------------------------------------------------------
|
|
@when(
|
|
'I apply overrides with unknown keys and a valid description change for "plan_service"'
|
|
)
|
|
def step_apply_mixed_override(context):
|
|
"""Apply overrides with both unknown keys and a valid description."""
|
|
context.registry.apply_overrides(
|
|
{
|
|
"plan_service": {
|
|
"unknown_field": True,
|
|
"description": "Updated description via override",
|
|
}
|
|
}
|
|
)
|
|
|
|
|
|
@then('the description for "plan_service" should be updated')
|
|
def step_verify_description_updated(context):
|
|
"""Verify the valid description field was applied."""
|
|
policy = context.registry.get("plan_service")
|
|
assert policy.description == "Updated description via override"
|
|
|
|
|
|
@then("the unknown keys should not appear in the policy")
|
|
def step_verify_no_unknown_fields(context):
|
|
"""Verify unknown keys did not leak into the policy."""
|
|
policy = context.registry.get("plan_service")
|
|
dumped = policy.model_dump()
|
|
assert "unknown_field" not in dumped
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# apply_overrides: retry_category scalar merge (lines 593-594)
|
|
# ---------------------------------------------------------------------------
|
|
@when('I apply overrides changing retry_category to "network" for "plan_service"')
|
|
def step_apply_retry_category_override(context):
|
|
"""Apply overrides that change the retry_category scalar field."""
|
|
context.registry.apply_overrides({"plan_service": {"retry_category": "network"}})
|
|
|
|
|
|
@then('the retry_category for "plan_service" should be "network"')
|
|
def step_verify_retry_category(context):
|
|
"""Verify the retry_category was updated."""
|
|
policy = context.registry.get("plan_service")
|
|
assert policy.retry_category == RetryCategory.NETWORK
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RetryPolicyConfig: max_delay < base_delay (lines 159-163)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I create a RetryPolicyConfig with base_delay {base:d} and max_delay {max_d:d}")
|
|
def step_create_retry_config_bad_delays(context, base, max_d):
|
|
"""Attempt to create a RetryPolicyConfig with max_delay < base_delay."""
|
|
context.retry_config_error = None
|
|
context.retry_config = None
|
|
try:
|
|
context.retry_config = RetryPolicyConfig(
|
|
base_delay=float(base),
|
|
max_delay=float(max_d),
|
|
)
|
|
except (ValidationError, ValueError) as exc:
|
|
context.retry_config_error = exc
|
|
|
|
|
|
@then("a ValueError should be raised mentioning max_delay and base_delay")
|
|
def step_verify_delay_error(context):
|
|
"""Verify the validator rejects max_delay < base_delay."""
|
|
assert context.retry_config_error is not None, (
|
|
"Expected an error but none was raised"
|
|
)
|
|
error_text = str(context.retry_config_error)
|
|
assert "max_delay" in error_text or "base_delay" in error_text, (
|
|
f"Error should mention max_delay/base_delay: {error_text}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry.get: auto-generate default for unknown service (lines 496-497)
|
|
# ---------------------------------------------------------------------------
|
|
@when('I get the policy for unknown service "{service_name}"')
|
|
def step_get_unknown_service(context, service_name):
|
|
"""Request a policy for a service not in the default registry."""
|
|
context.auto_policy = context.registry.get(service_name)
|
|
|
|
|
|
@then('a default policy should be returned for "{service_name}"')
|
|
def step_verify_auto_policy(context, service_name):
|
|
"""Verify a policy was returned with the correct service_name."""
|
|
assert context.auto_policy is not None
|
|
assert context.auto_policy.service_name == service_name
|
|
|
|
|
|
@then("the auto-generated policy should have the database retry category")
|
|
def step_verify_auto_category(context):
|
|
"""Verify the auto-generated policy defaults to DATABASE category."""
|
|
assert context.auto_policy.retry_category == RetryCategory.DATABASE
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry.register: store and retrieve a custom policy (lines 510-511)
|
|
# ---------------------------------------------------------------------------
|
|
@when('I register a custom policy for service "{svc}" with max_attempts {attempts:d}')
|
|
def step_register_custom_policy(context, svc, attempts):
|
|
"""Create and register a custom policy."""
|
|
policy = ServiceRetryPolicy(
|
|
service_name=svc,
|
|
retry_category=RetryCategory.NETWORK,
|
|
retry=RetryPolicyConfig(max_attempts=attempts),
|
|
description=f"Custom policy for {svc}",
|
|
)
|
|
context.registry.register(policy)
|
|
|
|
|
|
@then('the registry should return a policy for "{svc}" with max_attempts {attempts:d}')
|
|
def step_verify_registered_policy(context, svc, attempts):
|
|
"""Verify the registered policy can be retrieved with correct values."""
|
|
policy = context.registry.get(svc)
|
|
assert policy.service_name == svc
|
|
assert policy.retry.max_attempts == attempts
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry.all_policies: snapshot of all policies (lines 618-619)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I request all policies from the registry")
|
|
def step_request_all_policies(context):
|
|
"""Get a snapshot of all policies."""
|
|
context.all_policies = context.registry.all_policies()
|
|
|
|
|
|
@then("the snapshot should contain all default service policies")
|
|
def step_verify_all_policies_count(context):
|
|
"""Verify the snapshot contains the expected number of policies."""
|
|
assert len(context.all_policies) >= 11, (
|
|
f"Expected at least 11 default policies, got {len(context.all_policies)}"
|
|
)
|
|
assert "plan_service" in context.all_policies
|
|
assert "session_service" in context.all_policies
|
|
|
|
|
|
@then("the snapshot should be a deep copy not sharing identity")
|
|
def step_verify_deep_copy(context):
|
|
"""Verify the returned dict is a deep copy."""
|
|
snapshot1 = context.all_policies
|
|
snapshot2 = context.registry.all_policies()
|
|
# Same keys but different object identity
|
|
assert set(snapshot1.keys()) == set(snapshot2.keys())
|
|
assert snapshot1["plan_service"] is not snapshot2["plan_service"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry.registered_services: sorted list (lines 628-629)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I request the registered services list")
|
|
def step_request_services_list(context):
|
|
"""Get the sorted list of registered service names."""
|
|
context.services_list = context.registry.registered_services()
|
|
|
|
|
|
@then("rpmcov the list should be sorted alphabetically")
|
|
def step_verify_sorted(context):
|
|
"""Verify the list is sorted."""
|
|
assert context.services_list == sorted(context.services_list)
|
|
|
|
|
|
@then("the list should contain the default service names")
|
|
def step_verify_default_names(context):
|
|
"""Verify expected default service names are present."""
|
|
assert "plan_service" in context.services_list
|
|
assert "session_service" in context.services_list
|
|
assert "context_service" in context.services_list
|