Travis Vasceannie
1ce24cdf7b
- Moved all hookify configuration files from `.claude/` to `.claude/hooks/` subdirectory for better organization - Added four new blocking hooks to prevent common error handling anti-patterns: - `block-broad-exception-handler`: Prevents catching generic `Exception` with only logging - `block-datetime-now-fallback`: Blocks returning `datetime.now()` as fallback on parse failures to prevent data corruption - `block-default
216 lines
7.8 KiB
Python
216 lines
7.8 KiB
Python
"""Tests for ClientResult type and error handling utilities."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import grpc
|
|
import pytest
|
|
|
|
from noteflow.grpc.client_mixins._error_handling import (
|
|
grpc_error_to_result,
|
|
not_connected_error,
|
|
)
|
|
from noteflow.grpc.types import ClientErrorCode, ClientResult, Err, Ok
|
|
|
|
|
|
class TestOkResult:
|
|
"""Test successful result creation."""
|
|
|
|
def test_ok_creates_successful_result(self) -> None:
|
|
"""Ok() creates result with success=True."""
|
|
result = Ok("test_value")
|
|
|
|
assert (
|
|
result.success
|
|
and not result.failed
|
|
and result.value == "test_value"
|
|
and result.error is None
|
|
), f"Ok() should create successful result: success={result.success}, value={result.value}"
|
|
|
|
def test_ok_unwrap_returns_value(self) -> None:
|
|
"""unwrap() returns value for successful result."""
|
|
result: ClientResult[str] = Ok("test_value")
|
|
|
|
assert result.unwrap() == "test_value"
|
|
|
|
def test_ok_unwrap_or_returns_value(self) -> None:
|
|
"""unwrap_or() returns value for successful result."""
|
|
result: ClientResult[str] = Ok("test_value")
|
|
|
|
assert result.unwrap_or("default") == "test_value"
|
|
|
|
|
|
class TestErrResult:
|
|
"""Test failed result creation."""
|
|
|
|
def test_err_creates_failed_result(self) -> None:
|
|
"""Err() creates result with failed=True."""
|
|
result: ClientResult[str] = Err(
|
|
code=ClientErrorCode.NOT_FOUND, message="Not found"
|
|
)
|
|
|
|
assert (
|
|
result.failed
|
|
and not result.success
|
|
and result.value is None
|
|
and result.error is not None
|
|
and result.error.code == ClientErrorCode.NOT_FOUND
|
|
and result.error.message == "Not found"
|
|
), f"Err() should create failed result: {result}"
|
|
|
|
def test_err_unwrap_raises(self) -> None:
|
|
"""unwrap() raises for failed result."""
|
|
result: ClientResult[str] = Err(
|
|
code=ClientErrorCode.INTERNAL, message="Internal error"
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="Client operation failed: Internal error"):
|
|
result.unwrap()
|
|
|
|
def test_err_unwrap_or_returns_default(self) -> None:
|
|
"""unwrap_or() returns default for failed result."""
|
|
result: ClientResult[str] = Err(
|
|
code=ClientErrorCode.UNAVAILABLE, message="Service unavailable"
|
|
)
|
|
|
|
assert result.unwrap_or("default") == "default"
|
|
|
|
|
|
class TestClientError:
|
|
"""Test ClientError properties."""
|
|
|
|
@pytest.mark.parametrize(
|
|
"code",
|
|
[
|
|
ClientErrorCode.UNAVAILABLE,
|
|
ClientErrorCode.DEADLINE_EXCEEDED,
|
|
ClientErrorCode.INTERNAL,
|
|
],
|
|
)
|
|
def test_retryable_errors(self, code: ClientErrorCode) -> None:
|
|
"""is_retryable identifies transient errors."""
|
|
result: ClientResult[str] = Err(code=code, message="Error")
|
|
|
|
assert (
|
|
result.error is not None and result.error.is_retryable
|
|
), f"{code} should be retryable"
|
|
|
|
@pytest.mark.parametrize(
|
|
"code",
|
|
[
|
|
ClientErrorCode.NOT_FOUND,
|
|
ClientErrorCode.INVALID_ARGUMENT,
|
|
ClientErrorCode.PERMISSION_DENIED,
|
|
],
|
|
)
|
|
def test_non_retryable_errors(self, code: ClientErrorCode) -> None:
|
|
"""is_retryable correctly identifies non-retryable errors."""
|
|
result: ClientResult[str] = Err(code=code, message="Error")
|
|
|
|
assert (
|
|
result.error is not None and not result.error.is_retryable
|
|
), f"{code} should not be retryable"
|
|
|
|
def test_not_found_property(self) -> None:
|
|
"""is_not_found correctly identifies NOT_FOUND errors."""
|
|
result: ClientResult[str] = Err(
|
|
code=ClientErrorCode.NOT_FOUND, message="Not found"
|
|
)
|
|
|
|
assert (
|
|
result.error is not None and result.error.is_not_found
|
|
), "NOT_FOUND error should have is_not_found=True"
|
|
|
|
def test_not_found_property_false_for_other_codes(self) -> None:
|
|
"""is_not_found returns False for non-NOT_FOUND errors."""
|
|
result: ClientResult[str] = Err(
|
|
code=ClientErrorCode.INTERNAL, message="Internal error"
|
|
)
|
|
|
|
assert (
|
|
result.error is not None and not result.error.is_not_found
|
|
), "INTERNAL error should have is_not_found=False"
|
|
|
|
|
|
class TestGrpcErrorConversion:
|
|
"""Test conversion of gRPC errors to ClientResult."""
|
|
|
|
@pytest.mark.parametrize(
|
|
("grpc_code", "expected_client_code"),
|
|
[
|
|
(grpc.StatusCode.NOT_FOUND, ClientErrorCode.NOT_FOUND),
|
|
(grpc.StatusCode.INVALID_ARGUMENT, ClientErrorCode.INVALID_ARGUMENT),
|
|
(grpc.StatusCode.DEADLINE_EXCEEDED, ClientErrorCode.DEADLINE_EXCEEDED),
|
|
(grpc.StatusCode.ALREADY_EXISTS, ClientErrorCode.ALREADY_EXISTS),
|
|
(grpc.StatusCode.PERMISSION_DENIED, ClientErrorCode.PERMISSION_DENIED),
|
|
(grpc.StatusCode.FAILED_PRECONDITION, ClientErrorCode.FAILED_PRECONDITION),
|
|
(grpc.StatusCode.UNIMPLEMENTED, ClientErrorCode.UNIMPLEMENTED),
|
|
(grpc.StatusCode.INTERNAL, ClientErrorCode.INTERNAL),
|
|
(grpc.StatusCode.UNAVAILABLE, ClientErrorCode.UNAVAILABLE),
|
|
(grpc.StatusCode.RESOURCE_EXHAUSTED, ClientErrorCode.UNAVAILABLE),
|
|
],
|
|
)
|
|
def test_grpc_status_code_mapping(
|
|
self, grpc_code: grpc.StatusCode, expected_client_code: ClientErrorCode
|
|
) -> None:
|
|
"""grpc_error_to_result correctly maps gRPC status codes."""
|
|
mock_error = MagicMock()
|
|
mock_error.code = MagicMock(return_value=grpc_code)
|
|
mock_error.details = MagicMock(return_value="Test error")
|
|
|
|
result: ClientResult[str] = grpc_error_to_result(mock_error, "test_operation")
|
|
|
|
assert (
|
|
result.failed
|
|
and result.error is not None
|
|
and result.error.code == expected_client_code
|
|
and result.error.message == "Test error"
|
|
and result.error.grpc_status == grpc_code
|
|
), f"gRPC {grpc_code} should map to {expected_client_code}"
|
|
|
|
def test_grpc_error_without_details(self) -> None:
|
|
"""grpc_error_to_result handles errors without details."""
|
|
mock_error = MagicMock()
|
|
mock_error.code = MagicMock(return_value=grpc.StatusCode.INTERNAL)
|
|
mock_error.details = MagicMock(return_value=None)
|
|
mock_error.__str__ = MagicMock(return_value="Generic error")
|
|
|
|
result: ClientResult[str] = grpc_error_to_result(mock_error, "test_operation")
|
|
|
|
assert (
|
|
result.failed
|
|
and result.error is not None
|
|
and result.error.message == "Generic error"
|
|
), "Error without details should use str(error) as message"
|
|
|
|
def test_grpc_error_unknown_status(self) -> None:
|
|
"""grpc_error_to_result maps unknown status to UNKNOWN."""
|
|
mock_error = MagicMock()
|
|
mock_error.code = MagicMock(return_value=grpc.StatusCode.CANCELLED)
|
|
mock_error.details = MagicMock(return_value="Operation cancelled")
|
|
|
|
result: ClientResult[str] = grpc_error_to_result(mock_error, "test_operation")
|
|
|
|
assert (
|
|
result.failed
|
|
and result.error is not None
|
|
and result.error.code == ClientErrorCode.UNKNOWN
|
|
), "Unmapped gRPC status should become UNKNOWN"
|
|
|
|
|
|
class TestNotConnectedError:
|
|
"""Test not_connected_error utility."""
|
|
|
|
def test_creates_not_connected_error(self) -> None:
|
|
"""not_connected_error creates NOT_CONNECTED error."""
|
|
result: ClientResult[str] = not_connected_error("test_operation")
|
|
|
|
assert (
|
|
result.failed
|
|
and result.error is not None
|
|
and result.error.code == ClientErrorCode.NOT_CONNECTED
|
|
and result.error.message == "Not connected to server"
|
|
and result.error.grpc_status is None
|
|
), "not_connected_error should create NOT_CONNECTED error"
|