19fc9f3289
Backend (516 tests, 94% coverage): - Add azure_openai endpoint/deployment validation tests (config.py -> 100%) - Add _total_conversations and _avg_turns direct tests (queries.py -> 100%) - Add transformer edge cases: list content, string checkpoint, invalid JSON, malformed message graceful skip (transformer.py -> 93%) - Add safety combined status_code+error_message interaction tests - Fix ambiguous 200/422 assertion to strict 422 - Add E2E pagination shape assertions (total, page, per_page, row count) - Fix ReplayPool mock to respect LIMIT/OFFSET params Frontend (23 tests, vitest + happy-dom + @testing-library/react): - Add vitest infrastructure with happy-dom environment - Add api.ts tests: success, HTTP error, success=false, URL encoding - Add DashboardPage tests: loading, data, error, empty states - Add ReplayListPage tests: loading, empty, data, error, status badge classes - Add ReplayPage tests: loading, steps, empty, error states
97 lines
3.4 KiB
Python
97 lines
3.4 KiB
Python
"""Tests for app.safety module -- confirmation rules and MCP error taxonomy."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.safety import (
|
|
classify_mcp_error,
|
|
is_retryable,
|
|
max_retries,
|
|
requires_confirmation,
|
|
)
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
class TestRequiresConfirmation:
|
|
def test_read_agent_no_override(self) -> None:
|
|
result = requires_confirmation(agent_permission="read")
|
|
assert result.requires_confirmation is False
|
|
|
|
def test_write_agent_no_override(self) -> None:
|
|
result = requires_confirmation(agent_permission="write")
|
|
assert result.requires_confirmation is True
|
|
|
|
def test_interrupt_override_true(self) -> None:
|
|
result = requires_confirmation(
|
|
agent_permission="read", needs_interrupt=True,
|
|
)
|
|
assert result.requires_confirmation is True
|
|
|
|
def test_interrupt_override_false(self) -> None:
|
|
result = requires_confirmation(
|
|
agent_permission="write", needs_interrupt=False,
|
|
)
|
|
assert result.requires_confirmation is False
|
|
|
|
|
|
class TestClassifyMcpError:
|
|
@pytest.mark.parametrize("code", [408, 429, 500, 502, 503, 504])
|
|
def test_transient_status_codes(self, code: int) -> None:
|
|
assert classify_mcp_error(status_code=code) == "transient"
|
|
|
|
@pytest.mark.parametrize("code", [401, 403])
|
|
def test_auth_status_codes(self, code: int) -> None:
|
|
assert classify_mcp_error(status_code=code) == "auth"
|
|
|
|
@pytest.mark.parametrize("code", [400, 404, 422])
|
|
def test_validation_status_codes(self, code: int) -> None:
|
|
assert classify_mcp_error(status_code=code) == "validation"
|
|
|
|
def test_unknown_status_code(self) -> None:
|
|
assert classify_mcp_error(status_code=200) == "unknown"
|
|
|
|
def test_timeout_message(self) -> None:
|
|
assert classify_mcp_error(error_message="Connection timed out") == "transient"
|
|
|
|
def test_rate_limit_message(self) -> None:
|
|
assert classify_mcp_error(error_message="Rate limit exceeded") == "transient"
|
|
|
|
def test_unauthorized_message(self) -> None:
|
|
assert classify_mcp_error(error_message="Unauthorized access") == "auth"
|
|
|
|
def test_invalid_message(self) -> None:
|
|
assert classify_mcp_error(error_message="Invalid parameter") == "validation"
|
|
|
|
def test_unknown_message(self) -> None:
|
|
assert classify_mcp_error(error_message="Something happened") == "unknown"
|
|
|
|
def test_status_code_takes_precedence_over_message(self) -> None:
|
|
# 429 is transient by code; message would classify as validation
|
|
assert classify_mcp_error(status_code=429, error_message="invalid param") == "transient"
|
|
|
|
def test_non_classified_status_falls_through_to_message(self) -> None:
|
|
# 200 is not in any status set, so message classification takes over
|
|
assert classify_mcp_error(status_code=200, error_message="timed out") == "transient"
|
|
|
|
def test_no_args_returns_unknown(self) -> None:
|
|
assert classify_mcp_error() == "unknown"
|
|
|
|
|
|
class TestRetryPolicy:
|
|
def test_transient_is_retryable(self) -> None:
|
|
assert is_retryable("transient") is True
|
|
|
|
def test_validation_not_retryable(self) -> None:
|
|
assert is_retryable("validation") is False
|
|
|
|
def test_auth_not_retryable(self) -> None:
|
|
assert is_retryable("auth") is False
|
|
|
|
def test_unknown_not_retryable(self) -> None:
|
|
assert is_retryable("unknown") is False
|
|
|
|
def test_max_retries_value(self) -> None:
|
|
assert max_retries() == 3
|