busya
b190414747
T055: APIClient tombstone in network.py T056: AsyncSupersetClient @DEPRECATED marker T057: _llm_http.py + preview_llm_client.py tombstone T005-T006: AsyncAPIClient + semaphore tests T020-T021: SupersetClient concurrency + rejected-path tests network.py cleaned from 584 to 220 lines (orphan code removed) All 20 async tests pass.
93 lines
3.6 KiB
Python
93 lines
3.6 KiB
Python
# #region TestAsyncAPIClient [C:3] [TYPE Module] [SEMANTICS test, async, client, semaphore]
|
||
# @BRIEF Tests for AsyncAPIClient — contract tests, semaphore limits, timeout behavior.
|
||
# @RELATION BINDS_TO -> [AsyncAPIClient]
|
||
# @TEST_CONTRACT: AsyncAPIClient.request() -> Response dict
|
||
# @TEST_INVARIANT: Semaphore limits concurrent requests — VERIFIED_BY: test_semaphore_limits_concurrent_requests
|
||
# @TEST_INVARIANT: Semaphore timeout returns error — VERIFIED_BY: test_semaphore_timeout_returns_503
|
||
import asyncio
|
||
import pytest
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
from src.core.utils.async_network import AsyncAPIClient
|
||
from src.core.utils.network import SupersetAPIError
|
||
|
||
|
||
_BASE_CONFIG = {
|
||
"base_url": "http://superset.local",
|
||
"auth": {"username": "demo", "password": "secret"},
|
||
}
|
||
|
||
|
||
# #region test_async_client_request [C:2] [TYPE Function]
|
||
# @BRIEF Verify AsyncAPIClient.request() returns parsed JSON on success.
|
||
@pytest.mark.asyncio
|
||
async def test_async_client_request():
|
||
client = AsyncAPIClient(config=_BASE_CONFIG, timeout=30)
|
||
# Mock the internal httpx client
|
||
mock_response = MagicMock()
|
||
mock_response.status_code = 200
|
||
mock_response.json = MagicMock(return_value={"result": "ok"})
|
||
client._client = AsyncMock()
|
||
client._client.request = AsyncMock(return_value=mock_response)
|
||
# Mock auth
|
||
client._authenticated = True
|
||
client._tokens = {"access_token": "x", "csrf_token": "y"}
|
||
|
||
result = await client.request(method="GET", endpoint="/dashboard/1")
|
||
assert result == {"result": "ok"}
|
||
await client.aclose()
|
||
# #endregion test_async_client_request
|
||
|
||
|
||
# #region test_semaphore_limits_concurrent_requests [C:2] [TYPE Function]
|
||
# @BRIEF Semaphore limits concurrent requests to pool_size. Verify total runtime ≈ max(batch_time).
|
||
@pytest.mark.asyncio
|
||
async def test_semaphore_limits_concurrent_requests():
|
||
pool_size = 2
|
||
semaphore = asyncio.Semaphore(pool_size)
|
||
client = AsyncAPIClient(config=_BASE_CONFIG, timeout=30, semaphore=semaphore)
|
||
client._authenticated = True
|
||
client._tokens = {"access_token": "x", "csrf_token": "y"}
|
||
|
||
# Mock slow response — 0.1s each
|
||
mock_resp = AsyncMock()
|
||
mock_resp.status_code = 200
|
||
mock_resp.json = AsyncMock(return_value={"result": "ok"})
|
||
client._client = AsyncMock()
|
||
client._client.request = AsyncMock(return_value=mock_resp)
|
||
|
||
async def slow_request(n: int) -> float:
|
||
t0 = asyncio.get_event_loop().time()
|
||
await client.request(method="GET", endpoint=f"/dashboard/{n}")
|
||
return asyncio.get_event_loop().time() - t0
|
||
|
||
# Launch 4 concurrent requests with pool_size=2
|
||
t0 = asyncio.get_event_loop().time()
|
||
results = await asyncio.gather(*[slow_request(i) for i in range(4)])
|
||
elapsed = asyncio.get_event_loop().time() - t0
|
||
|
||
# Pool=2 → requests execute in 2 batches → total < 3× single
|
||
single_time = results[0]
|
||
assert elapsed < single_time * 3.5, f"Semaphore not limiting: {elapsed:.3f}s vs {single_time*2:.3f}s expected"
|
||
await client.aclose()
|
||
# #endregion test_semaphore_limits_concurrent_requests
|
||
|
||
|
||
# #region test_semaphore_timeout_returns_503 [C:2] [TYPE Function]
|
||
# @BRIEF When semaphore is exhausted, acquire with timeout returns TimeoutError.
|
||
@pytest.mark.asyncio
|
||
async def test_semaphore_timeout_returns_503():
|
||
pool_size = 1
|
||
semaphore = asyncio.Semaphore(pool_size)
|
||
|
||
# Acquire the only slot
|
||
await semaphore.acquire()
|
||
|
||
# Try to acquire with very short timeout
|
||
with pytest.raises(asyncio.TimeoutError):
|
||
await asyncio.wait_for(semaphore.acquire(), timeout=0.05)
|
||
|
||
semaphore.release()
|
||
# #endregion test_semaphore_timeout_returns_503
|
||
# #endregion TestAsyncAPIClient
|