mirror of
https://github.com/bohd4nx/FragmentAPI.git
synced 2026-07-25 06:14:29 +00:00
bohd4nx
78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""Check raw Fragment API calls and transport error handling."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from curl_cffi.requests import AsyncSession, Response
|
|
|
|
from pyfragment import FragmentClient, FragmentPageError
|
|
from pyfragment.core.transport import fragment_request
|
|
from tests.shared import FAKE_HASH, FAKE_RESPONSE
|
|
|
|
# client.call() mocked tests
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_call_returns_api_response(client: FragmentClient) -> None:
|
|
with (
|
|
patch("pyfragment.domains.base.get_fragment_hash", AsyncMock(return_value=FAKE_HASH)),
|
|
patch("pyfragment.domains.base.fragment_request", AsyncMock(return_value=FAKE_RESPONSE)),
|
|
):
|
|
result = await client.call("anyMethod", {"key": "value"})
|
|
|
|
assert result == FAKE_RESPONSE
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_call_default_page_url(client: FragmentClient) -> None:
|
|
with (
|
|
patch("pyfragment.domains.base.get_fragment_hash", AsyncMock(return_value=FAKE_HASH)),
|
|
patch("pyfragment.domains.base.fragment_request", AsyncMock(return_value=FAKE_RESPONSE)),
|
|
):
|
|
result = await client.call("anyMethod")
|
|
|
|
assert result == FAKE_RESPONSE
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_call_no_data(client: FragmentClient) -> None:
|
|
mock_request = AsyncMock(return_value={})
|
|
|
|
with (
|
|
patch("pyfragment.domains.base.get_fragment_hash", AsyncMock(return_value=FAKE_HASH)),
|
|
patch("pyfragment.domains.base.fragment_request", mock_request),
|
|
):
|
|
await client.call("anyMethod")
|
|
|
|
_, _, _, sent_data = mock_request.call_args.args
|
|
assert sent_data == {"method": "anyMethod"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_call_merges_extra_data(client: FragmentClient) -> None:
|
|
mock_request = AsyncMock(return_value={})
|
|
|
|
with (
|
|
patch("pyfragment.domains.base.get_fragment_hash", AsyncMock(return_value=FAKE_HASH)),
|
|
patch("pyfragment.domains.base.fragment_request", mock_request),
|
|
):
|
|
await client.call("anyMethod", {"key": "value", "num": 7})
|
|
|
|
_, _, _, sent_data = mock_request.call_args.args
|
|
assert sent_data == {"method": "anyMethod", "key": "value", "num": 7}
|
|
|
|
|
|
# fragment_request HTTP status tests
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fragment_request_non_200_raises() -> None:
|
|
response = MagicMock(spec=Response)
|
|
response.status_code = 429
|
|
|
|
session = AsyncMock(spec=AsyncSession)
|
|
session.post = AsyncMock(return_value=response)
|
|
|
|
with pytest.raises(FragmentPageError, match="429"):
|
|
await fragment_request(session, FAKE_HASH, {}, {"method": "anyMethod"})
|