155 lines
5.1 KiB
Python
155 lines
5.1 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from easybill_client import AsyncEasybillClient, EasybillClient
|
|
|
|
|
|
class WorkflowTransport(httpx.MockTransport):
|
|
def __init__(self):
|
|
self.rate_limit_hits = 0
|
|
super().__init__(self._handler)
|
|
|
|
def _handler(self, request: httpx.Request) -> httpx.Response:
|
|
if request.url.path == "/positions/5":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"id": 5,
|
|
"number": "ART-5",
|
|
"type": "PRODUCT",
|
|
"description": "Premium Support",
|
|
"price": 4900,
|
|
},
|
|
)
|
|
|
|
if request.url.path == "/document-payments":
|
|
if request.method == "POST":
|
|
body = request.read().decode()
|
|
assert "INV-1" in body
|
|
return httpx.Response(
|
|
201,
|
|
json={
|
|
"id": 88,
|
|
"document_id": 10,
|
|
"amount": 1999,
|
|
"reference": "INV-1",
|
|
"type": "CASH",
|
|
},
|
|
)
|
|
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"page": 1,
|
|
"pages": 1,
|
|
"limit": 100,
|
|
"total": 1,
|
|
"items": [
|
|
{
|
|
"id": 88,
|
|
"document_id": 10,
|
|
"amount": 1999,
|
|
"reference": "INV-1",
|
|
"type": "CASH",
|
|
}
|
|
],
|
|
},
|
|
)
|
|
|
|
if request.url.path == "/customers":
|
|
page = request.url.params.get("page", "1")
|
|
if page == "1":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"page": 1,
|
|
"pages": 2,
|
|
"limit": 1,
|
|
"total": 2,
|
|
"items": [{"id": 1, "company_name": "ACME GmbH"}],
|
|
},
|
|
)
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"page": 2,
|
|
"pages": 2,
|
|
"limit": 1,
|
|
"total": 2,
|
|
"items": [{"id": 2, "company_name": "Example AG"}],
|
|
},
|
|
)
|
|
|
|
if request.url.path == "/documents":
|
|
if self.rate_limit_hits == 0:
|
|
self.rate_limit_hits += 1
|
|
return httpx.Response(429, headers={"Retry-After": "0"}, json={"error": "slow down"})
|
|
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"page": 1,
|
|
"pages": 1,
|
|
"limit": 100,
|
|
"total": 1,
|
|
"items": [{"id": 10, "number": "RE-10", "amount": 1999}],
|
|
},
|
|
)
|
|
|
|
return httpx.Response(404, json={"error": "not found"})
|
|
|
|
|
|
class AsyncWorkflowTransport(httpx.MockTransport):
|
|
def __init__(self):
|
|
super().__init__(self._handler)
|
|
|
|
@staticmethod
|
|
def _handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path == "/positions":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"page": 1,
|
|
"pages": 1,
|
|
"limit": 100,
|
|
"total": 1,
|
|
"items": [{"id": 7, "number": "ART-7", "description": "Hosting"}],
|
|
},
|
|
)
|
|
return httpx.Response(404, json={"error": "not found"})
|
|
|
|
|
|
def test_sync_workflow_helpers_cover_positions_payments_pagination_and_retry():
|
|
transport = WorkflowTransport()
|
|
client = EasybillClient(api_key="token", transport=transport, max_retries=1, retry_backoff=0)
|
|
try:
|
|
position = client.get_position(5)
|
|
payments = client.list_document_payments(document_id=10)
|
|
created_payment = client.create_document_payment(document_id=10, amount=1999, reference="INV-1")
|
|
all_customers = list(client.iter_all_customers(limit=1))
|
|
all_payments = list(client.iter_all_document_payments(document_id=10))
|
|
documents = client.list_documents()
|
|
finally:
|
|
client.close()
|
|
|
|
assert position.id == 5
|
|
assert position.price == 4900
|
|
assert payments.items[0].reference == "INV-1"
|
|
assert created_payment.id == 88
|
|
assert [customer.id for customer in all_customers] == [1, 2]
|
|
assert [payment.id for payment in all_payments] == [88]
|
|
assert documents.items[0].number == "RE-10"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_workflow_helpers_list_positions_as_models():
|
|
client = AsyncEasybillClient(api_key="token", transport=AsyncWorkflowTransport())
|
|
try:
|
|
positions = await client.list_positions()
|
|
iterated = [item async for item in client.iter_all_positions()]
|
|
finally:
|
|
await client.aclose()
|
|
|
|
assert positions.total == 1
|
|
assert positions.items[0].number == "ART-7"
|
|
assert iterated[0].id == 7
|