Раис Юсупалиев
77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
import asyncio
|
|
|
|
import httpx
|
|
|
|
from app.adapters.delivery_providers.cdek.auth import CDEKAuthClient
|
|
|
|
|
|
def test_auth_client_reuses_valid_token_until_expiration() -> None:
|
|
call_count = 0
|
|
current_time = {"value": 100.0}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
nonlocal call_count
|
|
call_count += 1
|
|
assert request.url.path == "/v2/oauth/token"
|
|
return httpx.Response(
|
|
200,
|
|
json={"access_token": "token-1", "expires_in": 120},
|
|
request=request,
|
|
)
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.MockTransport(handler)
|
|
async with httpx.AsyncClient(transport=transport) as http_client:
|
|
auth_client = CDEKAuthClient(
|
|
http_client=http_client,
|
|
base_url="https://api.cdek.test/v2",
|
|
client_id="client-id",
|
|
client_secret="client-secret",
|
|
expiry_skew_seconds=5.0,
|
|
clock=lambda: current_time["value"],
|
|
)
|
|
|
|
first = await auth_client.get_access_token()
|
|
second = await auth_client.get_access_token()
|
|
|
|
assert first == "token-1"
|
|
assert second == "token-1"
|
|
|
|
asyncio.run(scenario())
|
|
|
|
assert call_count == 1
|
|
|
|
|
|
def test_auth_client_refreshes_expired_token() -> None:
|
|
responses = iter(
|
|
[
|
|
{"access_token": "token-1", "expires_in": 10},
|
|
{"access_token": "token-2", "expires_in": 10},
|
|
]
|
|
)
|
|
current_time = {"value": 0.0}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json=next(responses), request=request)
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.MockTransport(handler)
|
|
async with httpx.AsyncClient(transport=transport) as http_client:
|
|
auth_client = CDEKAuthClient(
|
|
http_client=http_client,
|
|
base_url="https://api.cdek.test/v2",
|
|
client_id="client-id",
|
|
client_secret="client-secret",
|
|
expiry_skew_seconds=0.0,
|
|
clock=lambda: current_time["value"],
|
|
)
|
|
|
|
first = await auth_client.get_access_token()
|
|
current_time["value"] = 11.0
|
|
second = await auth_client.get_access_token()
|
|
|
|
assert first == "token-1"
|
|
assert second == "token-2"
|
|
|
|
asyncio.run(scenario())
|