Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/hub/payments.md
Original file line number Diff line number Diff line change
Expand Up @@ -18,5 +18,6 @@ Access with `norbix.hub.payments`.
| `enable_payments_integration` | `PUT` | `/{version}/payments/integrations/{Id}/enable` | `project` |
| `get_payments_integration` | `GET` | `/{version}/payments/integrations/{id}` | `project` |
| `get_payments_integrations` | `GET` | `/{version}/payments/integrations` | `project` |
| `get_payments_webhook_log` | `GET` | `/{version}/payments/webhooks/log` | `project` |
| `save_payments_integration` | `POST` | `/{version}/payments/integrations` | `project` |
| `test_payments_integration` | `POST` | `/{version}/payments/integrations/test` | `project` |
26 changes: 26 additions & 0 deletions src/norbix_python/hub/payments.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,19 @@ def enable_payments(self, *, timeout: float | None = None, bearer_token: str | N
bearer_token=bearer_token,
)

def get_payments_webhook_log(self, *, timeout: float | None = None, bearer_token: str | None = None, **request: Any) -> Any:
"""GET /{version}/payments/webhooks/log — the received payment webhooks of the project, newest first (optional ``integration_id``, ``limit``)."""
return self._transport.send(
target="hub",
path="/{version}/payments/webhooks/log",
method="GET",
path_params={},
request=request,
scope="project",
timeout=timeout,
bearer_token=bearer_token,
)

def delete_payments_trigger(self, trigger_id: str, *, timeout: float | None = None, bearer_token: str | None = None, **request: Any) -> Any:
"""DELETE /{version}/payments/triggers/{triggerId}"""
return self._transport.send(
Expand Down Expand Up @@ -248,6 +261,19 @@ async def enable_payments(self, *, timeout: float | None = None, bearer_token: s
bearer_token=bearer_token,
)

async def get_payments_webhook_log(self, *, timeout: float | None = None, bearer_token: str | None = None, **request: Any) -> Any:
"""GET /{version}/payments/webhooks/log — the received payment webhooks of the project, newest first (optional ``integration_id``, ``limit``)."""
return await self._transport.send(
target="hub",
path="/{version}/payments/webhooks/log",
method="GET",
path_params={},
request=request,
scope="project",
timeout=timeout,
bearer_token=bearer_token,
)

async def delete_payments_trigger(self, trigger_id: str, *, timeout: float | None = None, bearer_token: str | None = None, **request: Any) -> Any:
"""DELETE /{version}/payments/triggers/{triggerId}"""
return await self._transport.send(
Expand Down
41 changes: 41 additions & 0 deletions tests/hub/test_payments.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ def test_hub_payments_module_surface() -> None:
module = client.hub.payments
assert callable(module.disable_payments)
assert callable(module.enable_payments)
assert callable(module.get_payments_webhook_log)
assert callable(module.delete_payments_trigger)
assert callable(module.disable_payments_trigger)
assert callable(module.enable_payments_trigger)
Expand Down Expand Up @@ -136,3 +137,43 @@ def test_hub_payments_test_payments_integration_request_shape() -> None:
assert transport.last_request['method'] == 'POST'
assert transport.last_request is not None
assert transport.last_request['url'].startswith('https://')


def test_hub_payments_get_payments_webhook_log_request_shape() -> None:
client, transport = make_client(account_id=None)
client.hub.payments.get_payments_webhook_log(integration_id="int_stripe", limit=20)
assert transport.last_request is not None
assert transport.last_request['method'] == 'GET'
assert '/payments/webhooks/log' in transport.last_request['url']
assert 'integration_id=int_stripe' in transport.last_request['url'] or 'integrationId=int_stripe' in transport.last_request['url']


def test_async_hub_payments_get_payments_webhook_log_hits_the_route() -> None:
import asyncio
from urllib.parse import parse_qs, urlparse

import httpx

from norbix_python import AsyncNorbix

seen: list[httpx.Request] = []
log = {"list": [{"integrationId": "int_stripe", "eventName": "payment_intent.succeeded", "statusCode": 200}]}

def handler(request: httpx.Request) -> httpx.Response:
seen.append(request)
return httpx.Response(200, json=log)

async def run() -> object:
client = AsyncNorbix(
project_id='test-project',
bearer_token='test-token',
http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
)
try:
return await client.hub.payments.get_payments_webhook_log(integration_id="int_stripe", limit=20)
finally:
await client.aclose()

assert asyncio.run(run()) == log
assert [(r.method, urlparse(str(r.url)).path) for r in seen] == [('GET', '/v3/payments/webhooks/log')]
assert parse_qs(urlparse(str(seen[0].url)).query).get('limit') == ['20']
Loading