File size: 11,648 Bytes
0ac42c5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 | # -*- coding: utf-8 -*-
"""
Unit tests for exception handlers.
Tests validation error handling and debug logging integration.
"""
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from fastapi import Request
from fastapi.exceptions import RequestValidationError
class TestSanitizeValidationErrors:
"""Tests for sanitize_validation_errors function."""
def test_sanitizes_bytes_in_input_field(self):
"""
What it does: Verifies that bytes in 'input' field are converted to strings.
Purpose: Ensure JSON serialization works for bytes objects.
"""
print("Setup: Creating error with bytes in input field...")
from kiro.exceptions import sanitize_validation_errors
errors = [
{
"type": "json_invalid",
"loc": ["body", 0],
"msg": "Invalid JSON",
"input": b'{"invalid": json}'
}
]
print("Action: Calling sanitize_validation_errors...")
result = sanitize_validation_errors(errors)
print(f"Comparing input type: Expected str, Got {type(result[0]['input'])}")
assert isinstance(result[0]["input"], str)
assert result[0]["input"] == '{"invalid": json}'
def test_sanitizes_bytes_in_list_values(self):
"""
What it does: Verifies that bytes in list values are converted to strings.
Purpose: Ensure nested bytes are handled.
"""
print("Setup: Creating error with bytes in list...")
from kiro.exceptions import sanitize_validation_errors
errors = [
{
"type": "value_error",
"loc": ["body", "messages"],
"msg": "Invalid value",
"input": [b'bytes1', "string", b'bytes2']
}
]
print("Action: Calling sanitize_validation_errors...")
result = sanitize_validation_errors(errors)
print(f"Checking list values are converted...")
assert result[0]["input"] == ["bytes1", "string", "bytes2"]
def test_preserves_non_bytes_values(self):
"""
What it does: Verifies that non-bytes values are preserved.
Purpose: Ensure normal values are not modified.
"""
print("Setup: Creating error with normal values...")
from kiro.exceptions import sanitize_validation_errors
errors = [
{
"type": "missing",
"loc": ["body", "model"],
"msg": "Field required",
"input": {"messages": []}
}
]
print("Action: Calling sanitize_validation_errors...")
result = sanitize_validation_errors(errors)
print(f"Checking values are preserved...")
assert result[0]["input"] == {"messages": []}
assert result[0]["type"] == "missing"
class TestValidationExceptionHandler:
"""Tests for validation_exception_handler function."""
@pytest.mark.asyncio
async def test_returns_422_status_code(self):
"""
What it does: Verifies that handler returns 422 status code.
Purpose: Ensure proper HTTP status for validation errors.
"""
print("Setup: Creating mock request and exception...")
from kiro.exceptions import validation_exception_handler
mock_request = MagicMock(spec=Request)
mock_request.body = AsyncMock(return_value=b'{"invalid": json}')
mock_exc = MagicMock(spec=RequestValidationError)
mock_exc.errors.return_value = [
{"type": "json_invalid", "loc": ["body"], "msg": "Invalid JSON", "input": {}}
]
# Patch debug_logger at the source module
with patch('kiro.debug_logger.debug_logger') as mock_logger:
print("Action: Calling validation_exception_handler...")
response = await validation_exception_handler(mock_request, mock_exc)
print(f"Comparing status_code: Expected 422, Got {response.status_code}")
assert response.status_code == 422
@pytest.mark.asyncio
async def test_calls_flush_on_error_with_422(self):
"""
What it does: Verifies that handler calls flush_on_error(422).
Purpose: Ensure debug logs are flushed for validation errors.
"""
print("Setup: Creating mock request and exception...")
from kiro.exceptions import validation_exception_handler
mock_request = MagicMock(spec=Request)
mock_request.body = AsyncMock(return_value=b'{"test": "data"}')
mock_exc = MagicMock(spec=RequestValidationError)
mock_exc.errors.return_value = [
{"type": "missing", "loc": ["body", "model"], "msg": "Field required", "input": {}}
]
# Patch debug_logger at the source module
with patch('kiro.debug_logger.debug_logger') as mock_logger:
print("Action: Calling validation_exception_handler...")
await validation_exception_handler(mock_request, mock_exc)
print("Verifying flush_on_error was called with 422...")
mock_logger.flush_on_error.assert_called_once()
call_args = mock_logger.flush_on_error.call_args
assert call_args[0][0] == 422 # First positional argument is status_code
@pytest.mark.asyncio
async def test_includes_sanitized_errors_in_response(self):
"""
What it does: Verifies that response includes sanitized errors.
Purpose: Ensure error details are returned to client.
"""
print("Setup: Creating mock request and exception...")
from kiro.exceptions import validation_exception_handler
import json
mock_request = MagicMock(spec=Request)
mock_request.body = AsyncMock(return_value=b'{"test": "data"}')
mock_exc = MagicMock(spec=RequestValidationError)
mock_exc.errors.return_value = [
{"type": "missing", "loc": ["body", "model"], "msg": "Field required", "input": {}}
]
with patch('kiro.debug_logger.debug_logger'):
print("Action: Calling validation_exception_handler...")
response = await validation_exception_handler(mock_request, mock_exc)
print("Parsing response body...")
body = json.loads(response.body.decode())
print(f"Verifying 'detail' is in response...")
assert "detail" in body
assert len(body["detail"]) == 1
assert body["detail"][0]["type"] == "missing"
@pytest.mark.asyncio
async def test_truncates_body_in_response(self):
"""
What it does: Verifies that body is truncated to 500 chars in response.
Purpose: Ensure large bodies don't bloat error responses.
"""
print("Setup: Creating mock request with large body...")
from kiro.exceptions import validation_exception_handler
import json
large_body = b'{"data": "' + b'x' * 1000 + b'"}'
mock_request = MagicMock(spec=Request)
mock_request.body = AsyncMock(return_value=large_body)
mock_exc = MagicMock(spec=RequestValidationError)
mock_exc.errors.return_value = [
{"type": "json_invalid", "loc": ["body"], "msg": "Invalid", "input": {}}
]
with patch('kiro.debug_logger.debug_logger'):
print("Action: Calling validation_exception_handler...")
response = await validation_exception_handler(mock_request, mock_exc)
print("Parsing response body...")
body = json.loads(response.body.decode())
print(f"Verifying body is truncated to 500 chars...")
assert len(body["body"]) <= 500
class TestValidationExceptionHandlerLogging:
"""Tests for logging behavior in validation_exception_handler."""
@pytest.mark.asyncio
async def test_logs_error_at_error_level(self):
"""
What it does: Verifies that validation error is logged at ERROR level.
Purpose: Ensure errors are visible in logs.
"""
print("Setup: Creating mock request and exception...")
from kiro.exceptions import validation_exception_handler
mock_request = MagicMock(spec=Request)
mock_request.body = AsyncMock(return_value=b'{"test": "data"}')
mock_exc = MagicMock(spec=RequestValidationError)
mock_exc.errors.return_value = [
{"type": "missing", "loc": ["body", "model"], "msg": "Field required", "input": {}}
]
with patch('kiro.debug_logger.debug_logger'):
with patch('kiro.exceptions.logger') as mock_logger:
print("Action: Calling validation_exception_handler...")
await validation_exception_handler(mock_request, mock_exc)
print("Verifying logger.error was called...")
mock_logger.error.assert_called()
class TestValidationExceptionHandlerEdgeCases:
"""Tests for edge cases in validation_exception_handler."""
@pytest.mark.asyncio
async def test_handles_empty_errors_list(self):
"""
What it does: Verifies that handler works with empty errors list.
Purpose: Ensure edge case doesn't cause crash.
"""
print("Setup: Creating mock request with empty errors...")
from kiro.exceptions import validation_exception_handler
import json
mock_request = MagicMock(spec=Request)
mock_request.body = AsyncMock(return_value=b'{}')
mock_exc = MagicMock(spec=RequestValidationError)
mock_exc.errors.return_value = []
with patch('kiro.debug_logger.debug_logger'):
print("Action: Calling validation_exception_handler...")
response = await validation_exception_handler(mock_request, mock_exc)
print(f"Verifying response is valid...")
assert response.status_code == 422
body = json.loads(response.body.decode())
assert body["detail"] == []
@pytest.mark.asyncio
async def test_handles_unicode_in_body(self):
"""
What it does: Verifies that handler works with unicode in body.
Purpose: Ensure international characters are handled.
"""
print("Setup: Creating mock request with unicode body...")
from kiro.exceptions import validation_exception_handler
import json
unicode_body = '{"message": "Привет мир 🌍"}'.encode('utf-8')
mock_request = MagicMock(spec=Request)
mock_request.body = AsyncMock(return_value=unicode_body)
mock_exc = MagicMock(spec=RequestValidationError)
mock_exc.errors.return_value = [
{"type": "missing", "loc": ["body", "model"], "msg": "Field required", "input": {}}
]
with patch('kiro.debug_logger.debug_logger'):
print("Action: Calling validation_exception_handler...")
response = await validation_exception_handler(mock_request, mock_exc)
print(f"Verifying response is valid...")
assert response.status_code == 422
body = json.loads(response.body.decode())
assert "Привет мир" in body["body"]
|