CoreRasurae
470 lines
16 KiB
Python
470 lines
16 KiB
Python
"""Step definitions for security exception handling tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.core.error_handling import (
|
|
ErrorCategory,
|
|
classify_error,
|
|
format_error_for_cli,
|
|
redact_error_details,
|
|
redact_value,
|
|
wrap_unexpected,
|
|
)
|
|
from cleveragents.core.exceptions import (
|
|
AuthenticationError,
|
|
AuthorizationError,
|
|
BusinessRuleViolation,
|
|
CleverAgentsError,
|
|
ConfigurationError,
|
|
DatabaseError,
|
|
ExecutionError,
|
|
ExternalServiceError,
|
|
FileSystemError,
|
|
MissingConfigurationError,
|
|
ModelNotAvailableError,
|
|
NetworkError,
|
|
PlanError,
|
|
ProviderError,
|
|
RateLimitError,
|
|
ResourceConflictError,
|
|
ResourceNotFoundError,
|
|
StreamRoutingError,
|
|
TokenLimitExceededError,
|
|
ValidationError,
|
|
)
|
|
|
|
__all__: list[str] = []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a security exceptions test environment")
|
|
def step_given_sec_exc_env(context: Context) -> None:
|
|
context.sec_exc = None # Exception
|
|
context.sec_error_info = None # ErrorInfo | None
|
|
context.sec_details = {} # dict
|
|
context.sec_redacted = {} # dict
|
|
context.sec_wrapped = None # CleverAgentsError | None
|
|
context.sec_cli_output = ""
|
|
context.sec_wrap_context = {} # dict
|
|
context.sec_string_value = ""
|
|
context.sec_redacted_string = ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given — exception types
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a ValidationError with message "{msg}"')
|
|
def step_given_validation_error(context: Context, msg: str) -> None:
|
|
context.sec_exc = ValidationError(msg)
|
|
|
|
|
|
@given('a ResourceNotFoundError for resource "{rtype}" with id "{rid}"')
|
|
def step_given_not_found(context: Context, rtype: str, rid: str) -> None:
|
|
context.sec_exc = ResourceNotFoundError(resource_type=rtype, resource_id=rid)
|
|
|
|
|
|
@given('an AuthenticationError with message "{msg}"')
|
|
def step_given_auth_error(context: Context, msg: str) -> None:
|
|
context.sec_exc = AuthenticationError(msg)
|
|
|
|
|
|
@given('an AuthorizationError with message "{msg}"')
|
|
def step_given_authz_error(context: Context, msg: str) -> None:
|
|
context.sec_exc = AuthorizationError(msg)
|
|
|
|
|
|
@given('a RateLimitError with message "{msg}"')
|
|
def step_given_rate_limit(context: Context, msg: str) -> None:
|
|
context.sec_exc = RateLimitError(msg)
|
|
|
|
|
|
@given('a DatabaseError with message "{msg}"')
|
|
def step_given_db_error(context: Context, msg: str) -> None:
|
|
context.sec_exc = DatabaseError(msg)
|
|
|
|
|
|
@given('a ConfigurationError with message "{msg}"')
|
|
def step_given_config_error(context: Context, msg: str) -> None:
|
|
context.sec_exc = ConfigurationError(msg)
|
|
|
|
|
|
@given('a NetworkError with message "{msg}"')
|
|
def step_given_network_error(context: Context, msg: str) -> None:
|
|
context.sec_exc = NetworkError(msg)
|
|
|
|
|
|
@given('a FileSystemError with message "{msg}"')
|
|
def step_given_fs_error(context: Context, msg: str) -> None:
|
|
context.sec_exc = FileSystemError(msg)
|
|
|
|
|
|
@given('a ProviderError with message "{msg}"')
|
|
def step_given_provider_error(context: Context, msg: str) -> None:
|
|
context.sec_exc = ProviderError(msg)
|
|
|
|
|
|
@given('a TokenLimitExceededError with message "{msg}"')
|
|
def step_given_token_limit(context: Context, msg: str) -> None:
|
|
context.sec_exc = TokenLimitExceededError(msg)
|
|
|
|
|
|
@given('a BusinessRuleViolation with message "{msg}"')
|
|
def step_given_biz_rule(context: Context, msg: str) -> None:
|
|
context.sec_exc = BusinessRuleViolation(msg)
|
|
|
|
|
|
@given('a PlanError with message "{msg}"')
|
|
def step_given_plan_error(context: Context, msg: str) -> None:
|
|
context.sec_exc = PlanError(msg)
|
|
|
|
|
|
@given('a bare Exception with message "{msg}"')
|
|
def step_given_bare_exception(context: Context, msg: str) -> None:
|
|
context.sec_exc = Exception(msg)
|
|
|
|
|
|
@given('an ExecutionError with message "{msg}"')
|
|
def step_given_execution_error(context: Context, msg: str) -> None:
|
|
context.sec_exc = ExecutionError(msg)
|
|
|
|
|
|
@given('a StreamRoutingError with message "{msg}"')
|
|
def step_given_stream_error(context: Context, msg: str) -> None:
|
|
context.sec_exc = StreamRoutingError(msg)
|
|
|
|
|
|
@given('a ModelNotAvailableError with message "{msg}"')
|
|
def step_given_model_unavail(context: Context, msg: str) -> None:
|
|
context.sec_exc = ModelNotAvailableError(msg)
|
|
|
|
|
|
@given('a ResourceConflictError with message "{msg}"')
|
|
def step_given_conflict(context: Context, msg: str) -> None:
|
|
context.sec_exc = ResourceConflictError(msg)
|
|
|
|
|
|
@given('a MissingConfigurationError with message "{msg}"')
|
|
def step_given_missing_config(context: Context, msg: str) -> None:
|
|
context.sec_exc = MissingConfigurationError(msg)
|
|
|
|
|
|
@given('an ExternalServiceError with message "{msg}"')
|
|
def step_given_ext_svc_error(context: Context, msg: str) -> None:
|
|
context.sec_exc = ExternalServiceError(msg)
|
|
|
|
|
|
@given('a CleverAgentsError with message "{msg}"')
|
|
def step_given_ca_error(context: Context, msg: str) -> None:
|
|
context.sec_exc = CleverAgentsError(msg)
|
|
|
|
|
|
@given('a detailed CleverAgentsError "{msg}" with detail "{k}" as "{v}"')
|
|
def step_given_ca_error_details(context: Context, msg: str, k: str, v: str) -> None:
|
|
context.sec_exc = CleverAgentsError(msg, details={k: v})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given — error details for redaction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('redaction test details with key "{key}" and value "{value}"')
|
|
def step_given_error_details(context: Context, key: str, value: str) -> None:
|
|
context.sec_details[key] = value
|
|
|
|
|
|
@given(
|
|
'redaction test details with nested key "{outer}" containing secret_key "{value}"'
|
|
)
|
|
def step_given_nested_details(context: Context, outer: str, value: str) -> None:
|
|
context.sec_details[outer] = {"secret_key": value}
|
|
|
|
|
|
@given('a redaction test string value "{value}"')
|
|
def step_given_string_value(context: Context, value: str) -> None:
|
|
context.sec_string_value = value
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given — wrapping context
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a wrapping context with key "{key}" and value "{value}"')
|
|
def step_given_wrap_context(context: Context, key: str, value: str) -> None:
|
|
context.sec_wrap_context[key] = value
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I classify the security exception")
|
|
def step_when_classify(context: Context) -> None:
|
|
context.sec_error_info = classify_error(context.sec_exc)
|
|
|
|
|
|
@when("I redact the error details")
|
|
def step_when_redact_details(context: Context) -> None:
|
|
context.sec_redacted = redact_error_details(context.sec_details)
|
|
|
|
|
|
@when("I redact the string value")
|
|
def step_when_redact_string(context: Context) -> None:
|
|
context.sec_redacted_string = redact_value(context.sec_string_value)
|
|
|
|
|
|
@when("I wrap the unexpected exception")
|
|
def step_when_wrap(context: Context) -> None:
|
|
ctx = context.sec_wrap_context or None
|
|
context.sec_wrapped = wrap_unexpected(context.sec_exc, context=ctx)
|
|
|
|
|
|
@when('I wrap the unexpected exception with message "{msg}"')
|
|
def step_when_wrap_msg(context: Context, msg: str) -> None:
|
|
context.sec_wrapped = wrap_unexpected(context.sec_exc, safe_message=msg)
|
|
|
|
|
|
@when("I format the error for CLI")
|
|
def step_when_format_cli(context: Context) -> None:
|
|
assert context.sec_error_info is not None
|
|
context.sec_cli_output = format_error_for_cli(context.sec_error_info)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps — classification
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the error code should be {code}")
|
|
def step_then_error_code(context: Context, code: str) -> None:
|
|
assert context.sec_error_info is not None
|
|
assert context.sec_error_info.code.value == int(code), (
|
|
f"Expected {code}, got {context.sec_error_info.code.value}"
|
|
)
|
|
|
|
|
|
@then('the error code name should be "{name}"')
|
|
def step_then_error_code_name(context: Context, name: str) -> None:
|
|
assert context.sec_error_info is not None
|
|
assert context.sec_error_info.code.name == name, (
|
|
f"Expected {name}, got {context.sec_error_info.code.name}"
|
|
)
|
|
|
|
|
|
@then('the error handling category should be "{cat}"')
|
|
def step_then_error_category(context: Context, cat: str) -> None:
|
|
assert context.sec_error_info is not None
|
|
expected = ErrorCategory.CLIENT if cat == "CLIENT" else ErrorCategory.SERVER
|
|
assert context.sec_error_info.category == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps — redaction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the redacted detail "{key}" should be "{expected}"')
|
|
def step_then_redacted_detail(context: Context, key: str, expected: str) -> None:
|
|
assert context.sec_redacted[key] == expected, (
|
|
f"Expected {expected!r}, got {context.sec_redacted[key]!r}"
|
|
)
|
|
|
|
|
|
@then('the redacted detail "{key}" should contain "{text}"')
|
|
def step_then_redacted_contains(context: Context, key: str, text: str) -> None:
|
|
val = str(context.sec_redacted[key])
|
|
assert text in val, f"Expected '{text}' in {val!r}"
|
|
|
|
|
|
@then('the redacted detail "{key}" should not contain "{text}"')
|
|
def step_then_redacted_not_contains(context: Context, key: str, text: str) -> None:
|
|
val = str(context.sec_redacted[key])
|
|
assert text not in val, f"Did not expect '{text}' in {val!r}"
|
|
|
|
|
|
@then('the redacted nested detail "{outer}" key "{inner}" should be "{expected}"')
|
|
def step_then_nested_redacted(
|
|
context: Context, outer: str, inner: str, expected: str
|
|
) -> None:
|
|
nested = context.sec_redacted[outer]
|
|
assert nested[inner] == expected
|
|
|
|
|
|
@then('the redacted string should contain "{text}"')
|
|
def step_then_redacted_string_contains(context: Context, text: str) -> None:
|
|
assert text in context.sec_redacted_string
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps — wrapping
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the wrapped error should be a CleverAgentsError")
|
|
def step_then_wrapped_is_ca(context: Context) -> None:
|
|
assert isinstance(context.sec_wrapped, CleverAgentsError)
|
|
|
|
|
|
@then('the wrapped error message should be "{msg}"')
|
|
def step_then_wrapped_msg(context: Context, msg: str) -> None:
|
|
assert context.sec_wrapped is not None
|
|
assert context.sec_wrapped.message == msg, (
|
|
f"Expected {msg!r}, got {context.sec_wrapped.message!r}"
|
|
)
|
|
|
|
|
|
@then('the wrapped error details should have key "{key}"')
|
|
def step_then_wrapped_details_key(context: Context, key: str) -> None:
|
|
assert context.sec_wrapped is not None
|
|
assert key in context.sec_wrapped.details, (
|
|
f"Key '{key}' not in {list(context.sec_wrapped.details.keys())}"
|
|
)
|
|
|
|
|
|
@then('the wrapped error details key "{key}" should be "{expected}"')
|
|
def step_then_wrapped_detail_value(context: Context, key: str, expected: str) -> None:
|
|
assert context.sec_wrapped is not None
|
|
assert context.sec_wrapped.details[key] == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps — CLI formatting
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the error CLI output should contain "{text}"')
|
|
def step_then_cli_contains(context: Context, text: str) -> None:
|
|
assert text in context.sec_cli_output, (
|
|
f"Expected '{text}' in:\n{context.sec_cli_output}"
|
|
)
|
|
|
|
|
|
@then('the error CLI output should not contain "{text}"')
|
|
def step_then_cli_not_contains(context: Context, text: str) -> None:
|
|
assert text not in context.sec_cli_output, (
|
|
f"Did not expect '{text}' in:\n{context.sec_cli_output}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given steps — list / integer / boolean details
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('redaction test details with key "{key}" and list values "{v1}" and "{v2}"')
|
|
def step_given_list_details(context: Context, key: str, v1: str, v2: str) -> None:
|
|
context.sec_details[key] = [v1, v2]
|
|
|
|
|
|
@given('redaction test details with key "{key}" and integer value {val:d}')
|
|
def step_given_int_details(context: Context, key: str, val: int) -> None:
|
|
context.sec_details[key] = val
|
|
|
|
|
|
@given('redaction test details with key "{key}" and boolean value {val}')
|
|
def step_given_bool_details(context: Context, key: str, val: str) -> None:
|
|
context.sec_details[key] = val.lower() == "true"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given steps — tracked details for non-mutation test (B1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a tracked CleverAgentsError "{msg}" with details key "{k}" value "{v}"')
|
|
def step_given_ca_tracked(context: Context, msg: str, k: str, v: str) -> None:
|
|
context.sec_exc = CleverAgentsError(msg, details={k: v})
|
|
# Capture a snapshot of the original details *before* wrapping.
|
|
context.sec_original_details_ref = dict(context.sec_exc.details)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps — error info message assertions (S1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the error info message should contain "{text}"')
|
|
def step_then_info_msg_contains(context: Context, text: str) -> None:
|
|
assert context.sec_error_info is not None
|
|
assert text in context.sec_error_info.message, (
|
|
f"Expected '{text}' in message: {context.sec_error_info.message!r}"
|
|
)
|
|
|
|
|
|
@then('the error info message should not contain "{text}"')
|
|
def step_then_info_msg_not_contains(context: Context, text: str) -> None:
|
|
assert context.sec_error_info is not None
|
|
assert text not in context.sec_error_info.message, (
|
|
f"Did not expect '{text}' in message: {context.sec_error_info.message!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps — list redaction (B2/T1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the redacted detail "{key}" should be a list')
|
|
def step_then_is_list(context: Context, key: str) -> None:
|
|
assert isinstance(context.sec_redacted[key], list), (
|
|
f"Expected list, got {type(context.sec_redacted[key]).__name__}"
|
|
)
|
|
|
|
|
|
@then('the redacted list "{key}" should contain "{text}"')
|
|
def step_then_list_contains(context: Context, key: str, text: str) -> None:
|
|
values = [str(v) for v in context.sec_redacted[key]]
|
|
assert any(text in v for v in values), f"Expected '{text}' in one of: {values}"
|
|
|
|
|
|
@then('the redacted list "{key}" should not contain "{text}"')
|
|
def step_then_list_not_contains(context: Context, key: str, text: str) -> None:
|
|
values = [str(v) for v in context.sec_redacted[key]]
|
|
assert all(text not in v for v in values), (
|
|
f"Did not expect '{text}' in any of: {values}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps — non-string passthrough (T4)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the redacted detail "{key}" should be integer {val:d}')
|
|
def step_then_int_detail(context: Context, key: str, val: int) -> None:
|
|
assert context.sec_redacted[key] == val, (
|
|
f"Expected {val}, got {context.sec_redacted[key]!r}"
|
|
)
|
|
|
|
|
|
@then('the redacted detail "{key}" should be boolean {val}')
|
|
def step_then_bool_detail(context: Context, key: str, val: str) -> None:
|
|
expected = val.lower() == "true"
|
|
assert context.sec_redacted[key] is expected, (
|
|
f"Expected {expected}, got {context.sec_redacted[key]!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps — non-mutation (B1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the original exception details reference should not have key "{key}"')
|
|
def step_then_orig_ref_no_key(context: Context, key: str) -> None:
|
|
assert key not in context.sec_original_details_ref, (
|
|
f"Key '{key}' unexpectedly found in original snapshot: "
|
|
f"{list(context.sec_original_details_ref.keys())}"
|
|
)
|