123 lines
3.8 KiB
Python
123 lines
3.8 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from easybill_client import AsyncEasybillClient, EasybillClient
|
|
|
|
|
|
class MockSyncTransport(httpx.MockTransport):
|
|
def __init__(self):
|
|
super().__init__(self._handler)
|
|
|
|
@staticmethod
|
|
def _handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path == "/customers/1":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"id": 1,
|
|
"company_name": "ACME GmbH",
|
|
"last_name": "Mustermann",
|
|
"number": "K-100",
|
|
},
|
|
)
|
|
|
|
if request.url.path == "/documents":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"page": 1,
|
|
"pages": 1,
|
|
"limit": 100,
|
|
"total": 1,
|
|
"items": [
|
|
{
|
|
"id": 10,
|
|
"number": "RE-10",
|
|
"customer_id": 1,
|
|
"type": "INVOICE",
|
|
"status": "DONE",
|
|
"amount": 1999,
|
|
"is_draft": False,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
|
|
if request.url.path == "/webhooks" and request.method == "POST":
|
|
body = request.read().decode()
|
|
assert "document.completed" in body
|
|
return httpx.Response(
|
|
201,
|
|
json={
|
|
"id": 77,
|
|
"url": "https://example.com/webhook",
|
|
"description": "Connector Hub",
|
|
"secret": "secret",
|
|
"content_type": "json",
|
|
"events": ["document.completed"],
|
|
"is_active": True,
|
|
},
|
|
)
|
|
|
|
return httpx.Response(404, json={"error": "not found"})
|
|
|
|
|
|
class MockAsyncTransport(httpx.MockTransport):
|
|
def __init__(self):
|
|
super().__init__(self._handler)
|
|
|
|
@staticmethod
|
|
def _handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path == "/customers":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"page": 1,
|
|
"pages": 1,
|
|
"limit": 100,
|
|
"total": 1,
|
|
"items": [
|
|
{
|
|
"id": 2,
|
|
"company_name": "Example AG",
|
|
"last_name": "Musterfrau",
|
|
"number": "K-200",
|
|
}
|
|
],
|
|
},
|
|
)
|
|
return httpx.Response(404, json={"error": "not found"})
|
|
|
|
|
|
def test_sync_client_returns_customer_and_documents_as_models():
|
|
client = EasybillClient(api_key="token", transport=MockSyncTransport())
|
|
try:
|
|
customer = client.get_customer(1)
|
|
documents = client.list_documents()
|
|
webhook = client.create_webhook(
|
|
url="https://example.com/webhook",
|
|
events=["document.completed"],
|
|
secret="secret",
|
|
description="Connector Hub",
|
|
)
|
|
finally:
|
|
client.close()
|
|
|
|
assert customer.id == 1
|
|
assert customer.company_name == "ACME GmbH"
|
|
assert documents.total == 1
|
|
assert documents.items[0].amount == 1999
|
|
assert webhook.id == 77
|
|
assert webhook.events == ["document.completed"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_client_lists_customers_as_models():
|
|
client = AsyncEasybillClient(api_key="token", transport=MockAsyncTransport())
|
|
try:
|
|
customers = await client.list_customers()
|
|
finally:
|
|
await client.aclose()
|
|
|
|
assert customers.total == 1
|
|
assert customers.items[0].company_name == "Example AG"
|