380 lines
15 KiB
Python
380 lines
15 KiB
Python
"""Tests for TgBackendInterface base class."""
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
import httpx
|
|
from pydantic import BaseModel, ValidationError
|
|
from app.interfaces.base import TgBackendInterface
|
|
|
|
|
|
class TestModel(BaseModel):
|
|
"""Test Pydantic model for testing."""
|
|
name: str
|
|
value: int
|
|
|
|
|
|
class TestTgBackendInterface:
|
|
"""Tests for TgBackendInterface base class."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init(self):
|
|
"""Test initialization with default parameters."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient') as MockClient:
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com/v1")
|
|
|
|
assert interface.api_prefix == "http://api.example.com/v1"
|
|
MockClient.assert_called_once()
|
|
|
|
# Verify timeout and retries configured
|
|
call_kwargs = MockClient.call_args[1]
|
|
assert call_kwargs['follow_redirects'] is True
|
|
assert isinstance(call_kwargs['timeout'], httpx.Timeout)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_strips_trailing_slash(self):
|
|
"""Test that trailing slash is stripped from api_prefix."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient'):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com/v1/")
|
|
assert interface.api_prefix == "http://api.example.com/v1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_custom_params(self):
|
|
"""Test initialization with custom timeout and retries."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient') as MockClient:
|
|
interface = TgBackendInterface(
|
|
api_prefix="http://api.example.com",
|
|
timeout=60.0,
|
|
max_retries=5
|
|
)
|
|
|
|
call_kwargs = MockClient.call_args[1]
|
|
# Timeout object is created, just verify it exists
|
|
assert isinstance(call_kwargs['timeout'], httpx.Timeout)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close(self):
|
|
"""Test closing the HTTP client."""
|
|
mock_client = AsyncMock()
|
|
with patch('app.interfaces.base.httpx.AsyncClient', return_value=mock_client):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
await interface.close()
|
|
|
|
mock_client.aclose.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_context_manager(self):
|
|
"""Test using interface as async context manager."""
|
|
mock_client = AsyncMock()
|
|
with patch('app.interfaces.base.httpx.AsyncClient', return_value=mock_client):
|
|
async with TgBackendInterface(api_prefix="http://api.example.com") as interface:
|
|
assert interface is not None
|
|
|
|
# Should close on exit
|
|
mock_client.aclose.assert_called_once()
|
|
|
|
def test_build_url_with_leading_slash(self):
|
|
"""Test building URL with path that has leading slash."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient'):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com/v1")
|
|
url = interface._build_url("/users/123")
|
|
|
|
assert url == "http://api.example.com/v1/users/123"
|
|
|
|
def test_build_url_without_leading_slash(self):
|
|
"""Test building URL with path without leading slash."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient'):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com/v1")
|
|
url = interface._build_url("users/123")
|
|
|
|
assert url == "http://api.example.com/v1/users/123"
|
|
|
|
def test_serialize_body_with_model(self):
|
|
"""Test serializing Pydantic model to dict."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient'):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
model = TestModel(name="test", value=42)
|
|
|
|
result = interface._serialize_body(model)
|
|
|
|
assert result == {"name": "test", "value": 42}
|
|
|
|
def test_serialize_body_with_none(self):
|
|
"""Test serializing None body."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient'):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
|
|
result = interface._serialize_body(None)
|
|
|
|
assert result is None
|
|
|
|
def test_deserialize_response_with_dict(self):
|
|
"""Test deserializing dict response to Pydantic model."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient'):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
data = {"name": "test", "value": 42}
|
|
|
|
result = interface._deserialize_response(data, TestModel)
|
|
|
|
assert isinstance(result, TestModel)
|
|
assert result.name == "test"
|
|
assert result.value == 42
|
|
|
|
def test_deserialize_response_no_model(self):
|
|
"""Test deserializing response without model returns raw data."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient'):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
data = {"name": "test", "value": 42}
|
|
|
|
result = interface._deserialize_response(data, None)
|
|
|
|
assert result == data
|
|
|
|
def test_deserialize_response_validation_error(self):
|
|
"""Test deserialization with validation error."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient'):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
# Invalid data: missing 'value' field
|
|
data = {"name": "test"}
|
|
|
|
with pytest.raises(ValidationError):
|
|
interface._deserialize_response(data, TestModel)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_response_success(self):
|
|
"""Test handling successful HTTP response."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient'):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.content = b'{"name": "test", "value": 42}' # Non-empty content
|
|
mock_response.json.return_value = {"name": "test", "value": 42}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
result = await interface._handle_response(mock_response, TestModel)
|
|
|
|
assert isinstance(result, TestModel)
|
|
assert result.name == "test"
|
|
mock_response.raise_for_status.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_response_204_no_content(self):
|
|
"""Test handling 204 No Content response."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient'):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 204
|
|
mock_response.content = b''
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
result = await interface._handle_response(mock_response)
|
|
|
|
assert result == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_response_empty_content(self):
|
|
"""Test handling response with empty content."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient'):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.content = b''
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
result = await interface._handle_response(mock_response)
|
|
|
|
assert result == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_response_http_error(self):
|
|
"""Test handling HTTP error response."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient'):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 404
|
|
mock_response.text = "Not Found"
|
|
|
|
error = httpx.HTTPStatusError(
|
|
"Not Found",
|
|
request=MagicMock(),
|
|
response=mock_response
|
|
)
|
|
mock_response.raise_for_status = MagicMock(side_effect=error)
|
|
|
|
with pytest.raises(httpx.HTTPStatusError):
|
|
await interface._handle_response(mock_response)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_response_invalid_json(self):
|
|
"""Test handling response with invalid JSON."""
|
|
with patch('app.interfaces.base.httpx.AsyncClient'):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.content = b'not empty'
|
|
mock_response.text = "Invalid JSON"
|
|
mock_response.json.side_effect = ValueError("Invalid JSON")
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with pytest.raises(ValueError):
|
|
await interface._handle_response(mock_response)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_success(self):
|
|
"""Test successful GET request."""
|
|
mock_client = AsyncMock()
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.content = b'{"name": "test", "value": 42}' # Non-empty content
|
|
mock_response.json.return_value = {"name": "test", "value": 42}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_client.get.return_value = mock_response
|
|
|
|
with patch('app.interfaces.base.httpx.AsyncClient', return_value=mock_client):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
|
|
result = await interface.get("/users", params={"id": 123}, response_model=TestModel)
|
|
|
|
assert isinstance(result, TestModel)
|
|
assert result.name == "test"
|
|
mock_client.get.assert_called_once()
|
|
call_args = mock_client.get.call_args
|
|
assert call_args[0][0] == "http://api.example.com/users"
|
|
assert call_args[1]['params'] == {"id": 123}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_success(self):
|
|
"""Test successful POST request."""
|
|
mock_client = AsyncMock()
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 201
|
|
mock_response.content = b'{"name": "created", "value": 100}' # Non-empty content
|
|
mock_response.json.return_value = {"name": "created", "value": 100}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_client.post.return_value = mock_response
|
|
|
|
with patch('app.interfaces.base.httpx.AsyncClient', return_value=mock_client):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
body = TestModel(name="new", value=50)
|
|
|
|
result = await interface.post("/users", body=body, response_model=TestModel)
|
|
|
|
assert isinstance(result, TestModel)
|
|
assert result.name == "created"
|
|
mock_client.post.assert_called_once()
|
|
call_args = mock_client.post.call_args
|
|
assert call_args[0][0] == "http://api.example.com/users"
|
|
assert call_args[1]['json'] == {"name": "new", "value": 50}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_without_body(self):
|
|
"""Test POST request without body."""
|
|
mock_client = AsyncMock()
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.content = b'{"result": "ok"}' # Non-empty content
|
|
mock_response.json.return_value = {"result": "ok"}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_client.post.return_value = mock_response
|
|
|
|
with patch('app.interfaces.base.httpx.AsyncClient', return_value=mock_client):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
|
|
result = await interface.post("/action")
|
|
|
|
assert result == {"result": "ok"}
|
|
call_args = mock_client.post.call_args
|
|
assert call_args[1]['json'] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_put_success(self):
|
|
"""Test successful PUT request."""
|
|
mock_client = AsyncMock()
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.content = b'{"name": "updated", "value": 75}' # Non-empty content
|
|
mock_response.json.return_value = {"name": "updated", "value": 75}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_client.put.return_value = mock_response
|
|
|
|
with patch('app.interfaces.base.httpx.AsyncClient', return_value=mock_client):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
body = TestModel(name="updated", value=75)
|
|
|
|
result = await interface.put("/users/1", body=body, response_model=TestModel)
|
|
|
|
assert isinstance(result, TestModel)
|
|
assert result.name == "updated"
|
|
mock_client.put.assert_called_once()
|
|
call_args = mock_client.put.call_args
|
|
assert call_args[0][0] == "http://api.example.com/users/1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_success(self):
|
|
"""Test successful DELETE request."""
|
|
mock_client = AsyncMock()
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 204
|
|
mock_response.content = b''
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_client.delete.return_value = mock_response
|
|
|
|
with patch('app.interfaces.base.httpx.AsyncClient', return_value=mock_client):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
|
|
result = await interface.delete("/users/1")
|
|
|
|
assert result == {}
|
|
mock_client.delete.assert_called_once()
|
|
call_args = mock_client.delete.call_args
|
|
assert call_args[0][0] == "http://api.example.com/users/1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_http_error(self):
|
|
"""Test GET request with HTTP error."""
|
|
mock_client = AsyncMock()
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 500
|
|
mock_response.text = "Internal Server Error"
|
|
|
|
error = httpx.HTTPStatusError(
|
|
"Internal Server Error",
|
|
request=MagicMock(),
|
|
response=mock_response
|
|
)
|
|
mock_response.raise_for_status = MagicMock(side_effect=error)
|
|
mock_client.get.return_value = mock_response
|
|
|
|
with patch('app.interfaces.base.httpx.AsyncClient', return_value=mock_client):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
|
|
with pytest.raises(httpx.HTTPStatusError):
|
|
await interface.get("/users")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_http_error(self):
|
|
"""Test POST request with HTTP error."""
|
|
mock_client = AsyncMock()
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 400
|
|
mock_response.text = "Bad Request"
|
|
|
|
error = httpx.HTTPStatusError(
|
|
"Bad Request",
|
|
request=MagicMock(),
|
|
response=mock_response
|
|
)
|
|
mock_response.raise_for_status = MagicMock(side_effect=error)
|
|
mock_client.post.return_value = mock_response
|
|
|
|
with patch('app.interfaces.base.httpx.AsyncClient', return_value=mock_client):
|
|
interface = TgBackendInterface(api_prefix="http://api.example.com")
|
|
body = TestModel(name="test", value=1)
|
|
|
|
with pytest.raises(httpx.HTTPStatusError):
|
|
await interface.post("/users", body=body)
|