diff --git a/README.md b/README.md
index 1ea2776..9a9ca5d 100644
--- a/README.md
+++ b/README.md
@@ -2,8 +2,9 @@
Python client for the [epoint.az](https://epoint.az) payment gateway.
-Covers all 30 documented endpoints: payments, saved cards, refunds and payouts, split payments,
-pre-authorisation, installments, wallets, Apple Pay and Google Pay, invoices, and B2B transfers.
+Covers every documented endpoint: payments, the checkout form post, saved cards, refunds and
+payouts, split payments, pre-authorisation, installments, wallets, Apple Pay and Google Pay,
+invoices, and B2B transfers.
Sync and async, fully typed, one dependency.
```bash
@@ -94,10 +95,34 @@ Most methods return a `Response`. Read known fields as attributes, anything else
r = client.get_status("te0000000001")
r.status # "success", compares equal to Status.SUCCESS
r.ok # True
-r.get("rrn") # bank reference number
+r.rrn # bank reference number
+r.amount # 25.5, a float
r.raw # the full response dict
```
+Typed attributes: `status`, `ok`, `transaction`, `redirect_url`, `message`, `trace_id`, `code`,
+`bank_transaction`, `bank_response`, `operation_code`, `rrn`, `card_name`, `card_mask`, `amount`,
+`card_id`, `order_id` and `widget_url`. Each is `None` when the response does not carry the field.
+Text fields come back as `str`. `amount` is a `float` and accepts a number or a numeric string
+from the API; anything else reads as `None`. `get`, `[]` and `raw` still give the value as sent.
+
+`ok` is true for `success`, for the B2B `SUCCESS` and for the heartbeat's `ok`, the values in
+`OK_STATUSES`. It is false for `error`, `failed`, `server_error`, `returned` and everything else.
+
+### Payments that ended in error
+
+`get_status` returns a payment whose status is `error` instead of raising. The rule is exact: on
+`get_status`, a 2xx response whose body has `status: "error"` and a `transaction` field is a
+payment state and comes back as a `Response`. Any other `status: "error"` body, such as an unknown
+transaction, and any `4xx` or `5xx` answer still raise. Every other method raises on
+`status: "error"` as before.
+
+```python
+status = client.get_status(transaction)
+if status.status == Status.ERROR:
+ retry_later(status.order_id)
+```
+
`create_payment`, `reserve`, `create_split_payment` and the other checkout methods return
`redirect_url`. `get_installment_plans` returns a list and `list_wallets` returns a dict.
@@ -123,6 +148,7 @@ if status.status == Status.SUCCESS:
| `CardStatus` | new, active, pending, rejected, expired, session_expired |
| `InvoiceStatus` | waiting_for_payment, paid, canceled |
| `B2BStatus` | PENDING, PROCESSING, SUCCESS, FAILED |
+| `B2BWebhookStatus` | success, failed |
| `OperationCode` | 001 card registration, 100 payment, 200 registration with payment |
| `Language` | az, en, ru |
| `Currency` | AZN, USD, EUR, RUB |
@@ -131,8 +157,8 @@ Currency is not uniform across the API. Checkout takes all four, but split, pre-
reverse, payout and wallet take AZN and nothing else. `SUPPORTED_CURRENCIES` and `AZN_ONLY` hold
those two sets.
-`SETTLED_STATUSES` is what `ok` checks, and `USABLE_CARD_STATUSES` is the set a card has to be in
-before you can charge it.
+`OK_STATUSES` is what `Response.ok` checks. `SETTLED_STATUSES` is what `Callback.ok` checks, and
+`USABLE_CARD_STATUSES` is the set a card has to be in before you can charge it.
## Saving a card without a payment
@@ -152,13 +178,52 @@ store_card(registration["card_id"])
The customer comes back to `success_redirect_url` or `error_redirect_url`, or to the client's own
defaults when neither is given. A card can be charged once `get_card_status` reports it `active`.
+## Checkout form
+
+`/api/1/checkout` is a browser form post, not a JSON call. `checkout_form` takes the same
+arguments as `create_payment`, makes no request, and returns a `CheckoutForm` with the form action
+`url` and the signed `data` and `signature` fields. Render it as a form that submits itself, and
+epoint answers with a redirect to its card page:
+
+```python
+form = client.checkout_form(amount=30.75, order_id="order-1", description="Test order")
+html = f"""
+
+
+"""
+```
+
+`form.fields` gives the two fields as a dict. `checkout_form` is synchronous on both clients.
+
+## B2B webhooks
+
+A B2B transfer created with `webhook_url` reports its result there as JSON with `order_id`,
+`status` (`success` or `failed`), `amount`, `payee_iban`, `payee_name` and `error_message`.
+`parse_b2b_webhook` turns the body, as a string, bytes or a dict, into a typed `B2BWebhook`.
+Unknown fields are ignored, and an unknown status is kept as a string.
+
+The docs define no signature for this webhook, so anyone can post one. Confirm the result with
+`get_bank_transfer` before acting on it:
+
+```python
+from epoint import parse_b2b_webhook
+
+hook = parse_b2b_webhook(request.body)
+confirmed = client.get_bank_transfer(hook.order_id)
+if confirmed.ok:
+ mark_paid(hook.order_id)
+```
+
## Methods
| Group | Methods |
|---|---|
-| Checkout | `create_payment`, `create_payment_request`, `create_amex_payment`, `change_payment_sum` |
+| Checkout | `create_payment`, `checkout_form`, `create_payment_request`, `create_amex_payment`, `change_payment_sum` |
| Status | `get_status`, `get_card_status`, `get_bank_transfer` |
-| Split | `create_split_payment`, `split_charge_saved_card` |
+| Split | `create_split_payment`, `create_token_split_payment`, `split_charge_saved_card` |
| Pre-auth | `reserve`, `capture` |
| Saved cards | `register_card`, `register_card_and_pay`, `charge_saved_card` |
| Money back | `refund`, `reverse` |
@@ -168,12 +233,19 @@ defaults when neither is given. A card can be charged once `get_card_status` rep
| Invoices | `create_invoice`, `update_invoice`, `get_invoice`, `list_invoices`, `send_invoice_sms`, `send_invoice_email` |
| B2B | `create_bank_transfer`, `get_bank_transfer` |
| Health | `heartbeat` |
+| Webhooks | `verify_callback`, `parse_b2b_webhook` |
+
+`create_token_split_payment` posts to `/api/1/token/split-payment`, the split payment for Apple Pay
+and Google Pay, with the same arguments as `create_split_payment`. The docs give no field for the
+wallet token, so none is sent.
+
+Not supported: `invoice_images[]` on invoices. The docs give no way to send it.
## Errors
| Exception | Raised when |
|---|---|
-| `GatewayError` | epoint returned `status: error` or `failed`. Carries `code`, `message`, `payload`. |
+| `GatewayError` | epoint returned `status: error`, except a payment state on `get_status`. Carries `code`, `message`, `payload`. |
| `TransportError` | Network failure, or a non-JSON or `4xx`/`5xx` response |
| `SignatureError` | A callback's signature did not match, or its data would not decode |
diff --git a/src/epoint/__init__.py b/src/epoint/__init__.py
index 1cbf3cd..24bb146 100644
--- a/src/epoint/__init__.py
+++ b/src/epoint/__init__.py
@@ -1,14 +1,16 @@
from __future__ import annotations
from .aio import AsyncEpointClient
-from .callbacks import verify_callback
+from .callbacks import parse_b2b_webhook, verify_callback
from .client import EpointClient
from .enums import (
AZN_ONLY,
+ OK_STATUSES,
SETTLED_STATUSES,
SUPPORTED_CURRENCIES,
USABLE_CARD_STATUSES,
B2BStatus,
+ B2BWebhookStatus,
CardStatus,
Currency,
InvoiceStatus,
@@ -17,19 +19,23 @@
Status,
)
from .errors import EpointError, GatewayError, SignatureError, TransportError
-from .models import Callback, Response
+from .models import B2BWebhook, Callback, CheckoutForm, Response
__version__ = "0.3.0"
__all__ = [
"AZN_ONLY",
+ "OK_STATUSES",
"SETTLED_STATUSES",
"SUPPORTED_CURRENCIES",
"USABLE_CARD_STATUSES",
"AsyncEpointClient",
"B2BStatus",
+ "B2BWebhook",
+ "B2BWebhookStatus",
"Callback",
"CardStatus",
+ "CheckoutForm",
"Currency",
"EpointClient",
"EpointError",
@@ -42,5 +48,6 @@
"Status",
"TransportError",
"__version__",
+ "parse_b2b_webhook",
"verify_callback",
]
diff --git a/src/epoint/_endpoints.py b/src/epoint/_endpoints.py
index 2081177..5c8ca76 100644
--- a/src/epoint/_endpoints.py
+++ b/src/epoint/_endpoints.py
@@ -52,6 +52,12 @@ def _checkout(
return Call(path, payload)
+def _flag(value: bool | None) -> int | None:
+ if value is None:
+ return None
+ return 1 if value else 0
+
+
def create_payment(
d: Defaults,
amount: Amount,
@@ -62,6 +68,7 @@ def create_payment(
success_redirect_url: str | None,
error_redirect_url: str | None,
other_attr: Any,
+ is_installment: bool | None = None,
) -> Call:
return _checkout(
d,
@@ -74,6 +81,34 @@ def create_payment(
success_redirect_url,
error_redirect_url,
other_attr,
+ extra={"is_installment": _flag(is_installment)},
+ )
+
+
+def checkout_form(
+ d: Defaults,
+ amount: Amount,
+ order_id: str,
+ description: str | None,
+ language: str | None,
+ currency: str | None,
+ success_redirect_url: str | None,
+ error_redirect_url: str | None,
+ other_attr: Any,
+ is_installment: bool | None = None,
+) -> Call:
+ return _checkout(
+ d,
+ "/api/1/checkout",
+ amount,
+ order_id,
+ description,
+ language,
+ currency,
+ success_redirect_url,
+ error_redirect_url,
+ other_attr,
+ extra={"is_installment": _flag(is_installment)},
)
@@ -87,6 +122,7 @@ def create_payment_request(
success_redirect_url: str | None,
error_redirect_url: str | None,
other_attr: Any,
+ is_installment: bool | None = None,
) -> Call:
return _checkout(
d,
@@ -99,6 +135,7 @@ def create_payment_request(
success_redirect_url,
error_redirect_url,
other_attr,
+ extra={"is_installment": _flag(is_installment)},
)
@@ -205,6 +242,34 @@ def create_split_payment(
)
+def create_token_split_payment(
+ d: Defaults,
+ amount: Amount,
+ order_id: str,
+ split_account: str,
+ split_amount: Amount,
+ description: str | None,
+ language: str | None,
+ currency: str | None,
+ success_redirect_url: str | None,
+ error_redirect_url: str | None,
+ other_attr: Any,
+) -> Call:
+ return _checkout(
+ d,
+ "/api/1/token/split-payment",
+ amount,
+ order_id,
+ description,
+ language,
+ currency,
+ success_redirect_url,
+ error_redirect_url,
+ other_attr,
+ extra={"split_user": split_account, "split_amount": money(split_amount)},
+ )
+
+
def pay_by_installment(
d: Defaults,
amount: Amount,
@@ -216,6 +281,7 @@ def pay_by_installment(
currency: str | None,
success_redirect_url: str | None,
error_redirect_url: str | None,
+ other_attr: Any = None,
) -> Call:
return _checkout(
d,
@@ -227,7 +293,7 @@ def pay_by_installment(
currency,
success_redirect_url,
error_redirect_url,
- None,
+ other_attr,
extra={
"installment_card_id": installment_card_id,
"installment_month": installment_month,
@@ -383,7 +449,13 @@ def reverse(
def get_installment_plans(
- d: Defaults, order_id: str, amount: Amount | None, language: str | None, currency: str | None
+ d: Defaults,
+ order_id: str,
+ amount: Amount | None,
+ language: str | None,
+ currency: str | None,
+ description: str | None = None,
+ other_attr: Any = None,
) -> Call:
return Call(
"/api/1/get-installment-request",
@@ -392,6 +464,8 @@ def get_installment_plans(
"amount": money(amount) if amount is not None else None,
"currency": currency or d.currency,
"language": language or d.language,
+ "description": description,
+ "other_attr": other_attr,
},
)
@@ -500,6 +574,9 @@ def update_invoice(
display: bool,
save_as_template: bool,
allow_installment: bool,
+ tin: str | None = None,
+ contract_number: str | None = None,
+ merchant_order_id: str | None = None,
) -> Call:
payload = _invoice(
amount,
@@ -512,7 +589,12 @@ def update_invoice(
display,
save_as_template,
allow_installment,
- {"id": invoice_id},
+ {
+ "id": invoice_id,
+ "inn": tin,
+ "contract_number": contract_number,
+ "merchant_order_id": merchant_order_id,
+ },
)
return Call("/api/1/invoices/update", payload)
@@ -546,6 +628,7 @@ def create_bank_transfer(
address: str | None,
additional_info: str | None,
webhook_url: str | None,
+ bulk_description: str | None = None,
) -> Call:
return Call(
"/api/1/b2b/payment",
@@ -562,5 +645,6 @@ def create_bank_transfer(
"address": address,
"additional_info": additional_info,
"webhook_url": webhook_url,
+ "bulk_description": bulk_description,
},
)
diff --git a/src/epoint/_transport.py b/src/epoint/_transport.py
index 344be03..5302101 100644
--- a/src/epoint/_transport.py
+++ b/src/epoint/_transport.py
@@ -18,10 +18,17 @@ def build_json(public_key: str, private_key: str, payload: dict[str, Any]) -> di
return build_form(public_key, private_key, payload)
-def parse(status_code: int, body: Any) -> Response:
+def is_payment_state(status_code: int, body: dict[str, Any]) -> bool:
+ return 200 <= status_code < 300 and body.get("status") == Status.ERROR and "transaction" in body
+
+
+def parse(status_code: int, body: Any, *, payment_status: bool = False) -> Response:
if not isinstance(body, dict):
raise TransportError(f"unexpected response body: {body!r}", status_code=status_code)
+ if payment_status and is_payment_state(status_code, body):
+ return Response(body)
+
if body.get("status") == Status.ERROR:
raise GatewayError(
str(body.get("message") or "request refused"),
diff --git a/src/epoint/aio.py b/src/epoint/aio.py
index cd6c587..4e0a180 100644
--- a/src/epoint/aio.py
+++ b/src/epoint/aio.py
@@ -13,7 +13,7 @@
from .client import PRODUCTION_URL
from .enums import Currency, Language
from .errors import TransportError
-from .models import Callback, Response
+from .models import Callback, CheckoutForm, Response
class AsyncEpointClient:
@@ -70,9 +70,12 @@ async def _send(self, method: str, path: str, **kwargs: Any) -> tuple[int, Any]:
) from exc
return response.status_code, body
- async def _post(self, path: str, payload: dict[str, Any]) -> Response:
+ async def _post(
+ self, path: str, payload: dict[str, Any], *, payment_status: bool = False
+ ) -> Response:
form = _transport.build_form(self.public_key, self.private_key, payload)
- return _transport.parse(*await self._send("POST", path, data=form))
+ status, body = await self._send("POST", path, data=form)
+ return _transport.parse(status, body, payment_status=payment_status)
async def _post_raw(self, path: str, payload: dict[str, Any]) -> Any:
form = _transport.build_form(self.public_key, self.private_key, payload)
@@ -96,6 +99,7 @@ async def create_payment(
success_redirect_url: str | None = None,
error_redirect_url: str | None = None,
other_attr: Any = None,
+ is_installment: bool | None = None,
) -> Response:
"""Start a payment and return the redirect URL the customer should be sent to."""
return await self._post(
@@ -109,9 +113,38 @@ async def create_payment(
success_redirect_url,
error_redirect_url,
other_attr,
+ is_installment,
)
)
+ def checkout_form(
+ self,
+ amount: Amount,
+ order_id: str,
+ *,
+ description: str | None = None,
+ language: str | None = None,
+ currency: str | None = None,
+ success_redirect_url: str | None = None,
+ error_redirect_url: str | None = None,
+ other_attr: Any = None,
+ is_installment: bool | None = None,
+ ) -> CheckoutForm:
+ path, payload = ep.checkout_form(
+ self._defaults,
+ amount,
+ order_id,
+ description,
+ language,
+ currency,
+ success_redirect_url,
+ error_redirect_url,
+ other_attr,
+ is_installment,
+ )
+ form = _transport.build_form(self.public_key, self.private_key, payload)
+ return CheckoutForm(f"{self.base_url}{path}", form["data"], form["signature"])
+
async def create_payment_request(
self,
amount: Amount,
@@ -123,6 +156,7 @@ async def create_payment_request(
success_redirect_url: str | None = None,
error_redirect_url: str | None = None,
other_attr: Any = None,
+ is_installment: bool | None = None,
) -> Response:
"""Same as create_payment, for the payment types epoint documents separately."""
return await self._post(
@@ -136,6 +170,7 @@ async def create_payment_request(
success_redirect_url,
error_redirect_url,
other_attr,
+ is_installment,
)
)
@@ -226,7 +261,7 @@ async def capture(self, transaction: str, amount: Amount) -> Response:
async def get_status(self, transaction: str) -> Response:
"""Look up a transaction by its epoint id."""
- return await self._post(*ep.get_status(transaction))
+ return await self._post(*ep.get_status(transaction), payment_status=True)
async def get_card_status(self, card_id: str) -> Response:
"""Look up a saved card by its id."""
@@ -263,6 +298,36 @@ async def create_split_payment(
)
)
+ async def create_token_split_payment(
+ self,
+ amount: Amount,
+ order_id: str,
+ split_account: str,
+ split_amount: Amount,
+ *,
+ description: str | None = None,
+ language: str | None = None,
+ currency: str | None = None,
+ success_redirect_url: str | None = None,
+ error_redirect_url: str | None = None,
+ other_attr: Any = None,
+ ) -> Response:
+ return await self._post(
+ *ep.create_token_split_payment(
+ self._defaults,
+ amount,
+ order_id,
+ split_account,
+ split_amount,
+ description,
+ language,
+ currency,
+ success_redirect_url,
+ error_redirect_url,
+ other_attr,
+ )
+ )
+
async def split_charge_saved_card(
self,
card_id: str,
@@ -402,10 +467,20 @@ async def get_installment_plans(
amount: Amount | None = None,
language: str | None = None,
currency: str | None = None,
+ description: str | None = None,
+ other_attr: Any = None,
) -> list[dict[str, Any]]:
"""List the installment plans available for an order."""
result = await self._post_raw(
- *ep.get_installment_plans(self._defaults, order_id, amount, language, currency)
+ *ep.get_installment_plans(
+ self._defaults,
+ order_id,
+ amount,
+ language,
+ currency,
+ description,
+ other_attr,
+ )
)
return list(result) if isinstance(result, list) else []
@@ -421,6 +496,7 @@ async def pay_by_installment(
currency: str | None = None,
success_redirect_url: str | None = None,
error_redirect_url: str | None = None,
+ other_attr: Any = None,
) -> Response:
"""Start an installment payment on a chosen plan."""
return await self._post(
@@ -435,6 +511,7 @@ async def pay_by_installment(
currency,
success_redirect_url,
error_redirect_url,
+ other_attr,
)
)
@@ -517,6 +594,9 @@ async def update_invoice(
description: str | None = None,
phone: str | None = None,
email: str | None = None,
+ tin: str | None = None,
+ contract_number: str | None = None,
+ merchant_order_id: str | None = None,
display: bool = True,
save_as_template: bool = False,
allow_installment: bool = False,
@@ -535,6 +615,9 @@ async def update_invoice(
display,
save_as_template,
allow_installment,
+ tin,
+ contract_number,
+ merchant_order_id,
)
)
@@ -569,6 +652,7 @@ async def create_bank_transfer(
address: str | None = None,
additional_info: str | None = None,
webhook_url: str | None = None,
+ bulk_description: str | None = None,
) -> Response:
"""Send a B2B bank transfer to an IBAN."""
return await self._post_json(
@@ -585,6 +669,7 @@ async def create_bank_transfer(
address,
additional_info,
webhook_url,
+ bulk_description,
)
)
diff --git a/src/epoint/callbacks.py b/src/epoint/callbacks.py
index 583891f..b095f7a 100644
--- a/src/epoint/callbacks.py
+++ b/src/epoint/callbacks.py
@@ -1,11 +1,28 @@
from __future__ import annotations
+import json
+from collections.abc import Mapping
+from typing import Any
+
from . import _signing
-from .errors import SignatureError
-from .models import Callback
+from .errors import EpointError, SignatureError
+from .models import B2BWebhook, Callback
def verify_callback(private_key: str, data: str, signature: str) -> Callback:
if not _signing.matches(private_key, data, signature):
raise SignatureError("callback signature does not match")
return Callback.from_dict(_signing.decode_data(data))
+
+
+def parse_b2b_webhook(body: str | bytes | Mapping[str, Any]) -> B2BWebhook:
+ if isinstance(body, Mapping):
+ data: Any = dict(body)
+ else:
+ try:
+ data = json.loads(body)
+ except ValueError as exc:
+ raise EpointError("b2b webhook body is not valid JSON") from exc
+ if not isinstance(data, dict):
+ raise EpointError("b2b webhook body is not a JSON object")
+ return B2BWebhook.from_dict(data)
diff --git a/src/epoint/client.py b/src/epoint/client.py
index d517ebd..b6f66da 100644
--- a/src/epoint/client.py
+++ b/src/epoint/client.py
@@ -12,7 +12,7 @@
from .callbacks import verify_callback
from .enums import Currency, Language
from .errors import TransportError
-from .models import Callback, Response
+from .models import Callback, CheckoutForm, Response
PRODUCTION_URL = "https://epoint.az"
@@ -71,9 +71,12 @@ def _send(self, method: str, path: str, **kwargs: Any) -> tuple[int, Any]:
) from exc
return response.status_code, body
- def _post(self, path: str, payload: dict[str, Any]) -> Response:
+ def _post(
+ self, path: str, payload: dict[str, Any], *, payment_status: bool = False
+ ) -> Response:
form = _transport.build_form(self.public_key, self.private_key, payload)
- return _transport.parse(*self._send("POST", path, data=form))
+ status, body = self._send("POST", path, data=form)
+ return _transport.parse(status, body, payment_status=payment_status)
def _post_raw(self, path: str, payload: dict[str, Any]) -> Any:
form = _transport.build_form(self.public_key, self.private_key, payload)
@@ -97,6 +100,7 @@ def create_payment(
success_redirect_url: str | None = None,
error_redirect_url: str | None = None,
other_attr: Any = None,
+ is_installment: bool | None = None,
) -> Response:
"""Start a payment and return the redirect URL the customer should be sent to."""
return self._post(
@@ -110,9 +114,38 @@ def create_payment(
success_redirect_url,
error_redirect_url,
other_attr,
+ is_installment,
)
)
+ def checkout_form(
+ self,
+ amount: Amount,
+ order_id: str,
+ *,
+ description: str | None = None,
+ language: str | None = None,
+ currency: str | None = None,
+ success_redirect_url: str | None = None,
+ error_redirect_url: str | None = None,
+ other_attr: Any = None,
+ is_installment: bool | None = None,
+ ) -> CheckoutForm:
+ path, payload = ep.checkout_form(
+ self._defaults,
+ amount,
+ order_id,
+ description,
+ language,
+ currency,
+ success_redirect_url,
+ error_redirect_url,
+ other_attr,
+ is_installment,
+ )
+ form = _transport.build_form(self.public_key, self.private_key, payload)
+ return CheckoutForm(f"{self.base_url}{path}", form["data"], form["signature"])
+
def create_payment_request(
self,
amount: Amount,
@@ -124,6 +157,7 @@ def create_payment_request(
success_redirect_url: str | None = None,
error_redirect_url: str | None = None,
other_attr: Any = None,
+ is_installment: bool | None = None,
) -> Response:
"""Same as create_payment, for the payment types epoint documents separately."""
return self._post(
@@ -137,6 +171,7 @@ def create_payment_request(
success_redirect_url,
error_redirect_url,
other_attr,
+ is_installment,
)
)
@@ -227,7 +262,7 @@ def capture(self, transaction: str, amount: Amount) -> Response:
def get_status(self, transaction: str) -> Response:
"""Look up a transaction by its epoint id."""
- return self._post(*ep.get_status(transaction))
+ return self._post(*ep.get_status(transaction), payment_status=True)
def get_card_status(self, card_id: str) -> Response:
"""Look up a saved card by its id."""
@@ -264,6 +299,36 @@ def create_split_payment(
)
)
+ def create_token_split_payment(
+ self,
+ amount: Amount,
+ order_id: str,
+ split_account: str,
+ split_amount: Amount,
+ *,
+ description: str | None = None,
+ language: str | None = None,
+ currency: str | None = None,
+ success_redirect_url: str | None = None,
+ error_redirect_url: str | None = None,
+ other_attr: Any = None,
+ ) -> Response:
+ return self._post(
+ *ep.create_token_split_payment(
+ self._defaults,
+ amount,
+ order_id,
+ split_account,
+ split_amount,
+ description,
+ language,
+ currency,
+ success_redirect_url,
+ error_redirect_url,
+ other_attr,
+ )
+ )
+
def split_charge_saved_card(
self,
card_id: str,
@@ -401,6 +466,8 @@ def get_installment_plans(
amount: Amount | None = None,
language: str | None = None,
currency: str | None = None,
+ description: str | None = None,
+ other_attr: Any = None,
) -> list[dict[str, Any]]:
"""List the installment plans available for an order."""
result = self._post_raw(
@@ -410,6 +477,8 @@ def get_installment_plans(
amount,
language,
currency,
+ description,
+ other_attr,
)
)
return list(result) if isinstance(result, list) else []
@@ -426,6 +495,7 @@ def pay_by_installment(
currency: str | None = None,
success_redirect_url: str | None = None,
error_redirect_url: str | None = None,
+ other_attr: Any = None,
) -> Response:
"""Start an installment payment on a chosen plan."""
return self._post(
@@ -440,6 +510,7 @@ def pay_by_installment(
currency,
success_redirect_url,
error_redirect_url,
+ other_attr,
)
)
@@ -522,6 +593,9 @@ def update_invoice(
description: str | None = None,
phone: str | None = None,
email: str | None = None,
+ tin: str | None = None,
+ contract_number: str | None = None,
+ merchant_order_id: str | None = None,
display: bool = True,
save_as_template: bool = False,
allow_installment: bool = False,
@@ -540,6 +614,9 @@ def update_invoice(
display,
save_as_template,
allow_installment,
+ tin,
+ contract_number,
+ merchant_order_id,
)
)
@@ -574,6 +651,7 @@ def create_bank_transfer(
address: str | None = None,
additional_info: str | None = None,
webhook_url: str | None = None,
+ bulk_description: str | None = None,
) -> Response:
"""Send a B2B bank transfer to an IBAN."""
return self._post_json(
@@ -590,6 +668,7 @@ def create_bank_transfer(
address,
additional_info,
webhook_url,
+ bulk_description,
)
)
diff --git a/src/epoint/enums.py b/src/epoint/enums.py
index 86ceb10..994c75b 100644
--- a/src/epoint/enums.py
+++ b/src/epoint/enums.py
@@ -54,6 +54,16 @@ class B2BStatus(StrEnum):
FAILED = "FAILED"
+class B2BWebhookStatus(StrEnum):
+ SUCCESS = "success"
+ FAILED = "failed"
+
+
+HEARTBEAT_OK = "ok"
+
+OK_STATUSES = frozenset({Status.SUCCESS.value, B2BStatus.SUCCESS.value, HEARTBEAT_OK})
+
+
class OperationCode(StrEnum):
"""What the transaction did, as it comes back on a callback."""
diff --git a/src/epoint/models.py b/src/epoint/models.py
index 14ca547..70a60b1 100644
--- a/src/epoint/models.py
+++ b/src/epoint/models.py
@@ -1,9 +1,28 @@
from __future__ import annotations
from dataclasses import dataclass
+from decimal import Decimal, InvalidOperation
from typing import Any
-from .enums import SETTLED_STATUSES
+from .enums import OK_STATUSES, SETTLED_STATUSES, B2BWebhookStatus
+
+
+def _text(value: Any) -> str | None:
+ return str(value) if value is not None else None
+
+
+def _number(value: Any) -> float | None:
+ if value is None or isinstance(value, bool):
+ return None
+ if isinstance(value, (int, float, Decimal)):
+ return float(value)
+ if isinstance(value, str):
+ try:
+ parsed = Decimal(value.strip())
+ except InvalidOperation:
+ return None
+ return float(parsed) if parsed.is_finite() else None
+ return None
class Response:
@@ -17,7 +36,7 @@ def status(self) -> str | None:
@property
def ok(self) -> bool:
- return self.raw.get("status") in SETTLED_STATUSES
+ return self.raw.get("status") in OK_STATUSES
@property
def transaction(self) -> str | None:
@@ -34,6 +53,54 @@ def message(self) -> str | None:
value = self.raw.get("message")
return str(value) if value is not None else None
+ @property
+ def trace_id(self) -> str | None:
+ return _text(self.raw.get("trace_id"))
+
+ @property
+ def code(self) -> str | None:
+ return _text(self.raw.get("code"))
+
+ @property
+ def bank_transaction(self) -> str | None:
+ return _text(self.raw.get("bank_transaction"))
+
+ @property
+ def bank_response(self) -> str | None:
+ return _text(self.raw.get("bank_response"))
+
+ @property
+ def operation_code(self) -> str | None:
+ return _text(self.raw.get("operation_code"))
+
+ @property
+ def rrn(self) -> str | None:
+ return _text(self.raw.get("rrn"))
+
+ @property
+ def card_name(self) -> str | None:
+ return _text(self.raw.get("card_name"))
+
+ @property
+ def card_mask(self) -> str | None:
+ return _text(self.raw.get("card_mask"))
+
+ @property
+ def card_id(self) -> str | None:
+ return _text(self.raw.get("card_id"))
+
+ @property
+ def order_id(self) -> str | None:
+ return _text(self.raw.get("order_id"))
+
+ @property
+ def widget_url(self) -> str | None:
+ return _text(self.raw.get("widget_url"))
+
+ @property
+ def amount(self) -> float | None:
+ return _number(self.raw.get("amount"))
+
def get(self, key: str, default: Any = None) -> Any:
return self.raw.get(key, default)
@@ -62,6 +129,9 @@ class Callback:
amount: float
other_attr: Any
raw: dict[str, Any]
+ bank_response: str | None = None
+ trace_id: str | None = None
+ card_id: str | None = None
@property
def ok(self) -> bool:
@@ -83,4 +153,45 @@ def from_dict(cls, data: dict[str, Any]) -> Callback:
amount=float(data["amount"]) if data.get("amount") is not None else 0.0,
other_attr=data.get("other_attr"),
raw=data,
+ bank_response=_text(data.get("bank_response")),
+ trace_id=_text(data.get("trace_id")),
+ card_id=_text(data.get("card_id")),
+ )
+
+
+@dataclass(frozen=True)
+class CheckoutForm:
+ url: str
+ data: str
+ signature: str
+
+ @property
+ def fields(self) -> dict[str, str]:
+ return {"data": self.data, "signature": self.signature}
+
+
+@dataclass(frozen=True)
+class B2BWebhook:
+ order_id: str
+ status: str
+ amount: float | None
+ payee_iban: str | None
+ payee_name: str | None
+ error_message: str | None
+ raw: dict[str, Any]
+
+ @property
+ def ok(self) -> bool:
+ return self.status == B2BWebhookStatus.SUCCESS
+
+ @classmethod
+ def from_dict(cls, data: dict[str, Any]) -> B2BWebhook:
+ return cls(
+ order_id=str(data.get("order_id", "")),
+ status=str(data.get("status", "")),
+ amount=_number(data.get("amount")),
+ payee_iban=_text(data.get("payee_iban")),
+ payee_name=_text(data.get("payee_name")),
+ error_message=_text(data.get("error_message")),
+ raw=data,
)
diff --git a/tests/test_async.py b/tests/test_async.py
index 734f513..5a7b6d4 100644
--- a/tests/test_async.py
+++ b/tests/test_async.py
@@ -2,11 +2,13 @@
import json
import os
import uuid
+from typing import Any
import httpx
import pytest
from epoint import AsyncEpointClient, GatewayError
+from epoint._signing import matches
PUBLIC_KEY = "i000000001"
PRIVATE_KEY = "sandbox_private_key_0000000001"
@@ -79,6 +81,176 @@ def handler(request: httpx.Request) -> httpx.Response:
assert result.get("code") == "116"
+def async_recording_client(seen: dict, response: Any = None) -> AsyncEpointClient:
+ def handler(request: httpx.Request) -> httpx.Response:
+ seen.clear()
+ seen["path"] = request.url.path
+ if request.headers.get("content-type", "").startswith("application/json"):
+ body = json.loads(request.content)
+ seen["payload"] = json.loads(base64.b64decode(body["data"]))
+ else:
+ form = httpx.QueryParams(request.content.decode())
+ seen["payload"] = json.loads(base64.b64decode(form["data"]))
+ return httpx.Response(200, json={"status": "success"} if response is None else response)
+
+ return AsyncEpointClient(
+ PUBLIC_KEY,
+ PRIVATE_KEY,
+ base_url="https://epoint.test",
+ success_redirect_url="https://shop.test/default-ok",
+ error_redirect_url="https://shop.test/default-fail",
+ transport=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
+ )
+
+
+async def test_get_status_returns_a_payment_that_ended_in_error_async():
+ body = {
+ "status": "error",
+ "code": "",
+ "message": "An error occurred during payment",
+ "transaction": "te1",
+ }
+
+ async with async_client_with(lambda request: httpx.Response(200, json=body)) as client:
+ result = await client.get_status("te1")
+
+ assert result.status == "error"
+ assert result.transaction == "te1"
+ assert not result.ok
+
+
+async def test_get_status_still_raises_on_a_non_2xx_error_async():
+ body = {"status": "error", "message": "denied", "transaction": "te1"}
+
+ async with async_client_with(lambda request: httpx.Response(403, json=body)) as client:
+ with pytest.raises(GatewayError):
+ await client.get_status("te1")
+
+
+async def test_other_calls_still_raise_on_error_with_a_transaction_async():
+ body = {"status": "error", "message": "bad", "transaction": "te1"}
+
+ async with async_client_with(lambda request: httpx.Response(200, json=body)) as client:
+ with pytest.raises(GatewayError):
+ await client.create_payment(1, "o")
+
+
+async def test_ok_and_typed_fields_async():
+ body = {"status": "SUCCESS", "order_id": "b1", "trace_id": "tr1", "amount": "1250.00"}
+
+ async with async_client_with(lambda request: httpx.Response(200, json=body)) as client:
+ result = await client.get_bank_transfer("b1")
+
+ assert result.ok
+ assert result.order_id == "b1"
+ assert result.trace_id == "tr1"
+ assert result.amount == 1250.0
+
+
+async def test_token_split_payment_payload_async():
+ seen: dict = {}
+ async with async_recording_client(seen) as client:
+ await client.create_token_split_payment(
+ amount=100, order_id="o-1", split_account="i000000002", split_amount=30
+ )
+
+ assert seen["path"] == "/api/1/token/split-payment"
+ payload = seen["payload"]
+ assert payload["amount"] == "100.00"
+ assert payload["split_user"] == "i000000002"
+ assert payload["split_amount"] == "30.00"
+ assert payload["currency"] == "AZN"
+ assert payload["order_id"] == "o-1"
+ assert payload["success_redirect_url"] == "https://shop.test/default-ok"
+ assert payload["error_redirect_url"] == "https://shop.test/default-fail"
+
+
+async def test_checkout_form_is_signed_and_makes_no_request_async():
+ def handler(request: httpx.Request) -> httpx.Response:
+ raise AssertionError("checkout_form must not send a request")
+
+ async with AsyncEpointClient(
+ PUBLIC_KEY,
+ PRIVATE_KEY,
+ base_url="https://epoint.test",
+ transport=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
+ ) as client:
+ form = client.checkout_form(amount=5, order_id="o-1", is_installment=False)
+
+ assert form.url == "https://epoint.test/api/1/checkout"
+ assert matches(PRIVATE_KEY, form.data, form.signature)
+ payload = json.loads(base64.b64decode(form.data))
+ assert payload == {
+ "public_key": PUBLIC_KEY,
+ "amount": "5.00",
+ "currency": "AZN",
+ "language": "az",
+ "order_id": "o-1",
+ "is_installment": 0,
+ }
+
+
+@pytest.mark.parametrize(("value", "sent"), [(True, 1), (False, 0), (None, None)])
+async def test_is_installment_is_sent_as_an_integer_or_omitted_async(value, sent):
+ seen: dict = {}
+ async with async_recording_client(seen) as client:
+ for call, path in (
+ (client.create_payment, "/api/1/request"),
+ (client.create_payment_request, "/api/1/payment-request"),
+ ):
+ await call(amount=1, order_id="o", is_installment=value)
+ assert seen["path"] == path
+ assert seen["payload"].get("is_installment") == sent
+ assert ("is_installment" in seen["payload"]) == (value is not None)
+
+
+async def test_new_optional_parameters_async():
+ seen: dict = {}
+ async with async_recording_client(seen) as client:
+ await client.create_bank_transfer(
+ order_id="b1",
+ amount=100,
+ description="fees",
+ iban="AZ21NABZ00000000137010001944",
+ name="ACME",
+ tin="1234567892",
+ bank_code="505141",
+ bulk_description="July payroll",
+ )
+ assert seen["payload"]["bulk_description"] == "July payroll"
+
+ await client.pay_by_installment(
+ amount=200,
+ order_id="o",
+ installment_card_id="kapital",
+ installment_month=6,
+ other_attr={"k": "v"},
+ )
+ assert seen["path"] == "/api/1/installment-request"
+ assert seen["payload"]["other_attr"] == {"k": "v"}
+
+ await client.update_invoice(
+ 7,
+ amount=11,
+ period_from="2026-01-01",
+ period_to="2026-12-31",
+ tin="1234567892",
+ contract_number="C-1",
+ merchant_order_id="M-1",
+ )
+ assert seen["payload"]["inn"] == "1234567892"
+ assert seen["payload"]["contract_number"] == "C-1"
+ assert seen["payload"]["merchant_order_id"] == "M-1"
+
+ async with async_recording_client(seen, []) as client:
+ plans = await client.get_installment_plans(
+ order_id="o", amount=200, description="sofa", other_attr={"k": "v"}
+ )
+ assert plans == []
+ assert seen["payload"]["description"] == "sofa"
+ assert seen["payload"]["other_attr"] == {"k": "v"}
+
+
def sandbox_up() -> bool:
try:
return httpx.get(f"{SANDBOX_URL}/api/heartbeat", timeout=2).status_code == 200
@@ -111,3 +283,80 @@ async def test_payment_end_to_end_async():
status = await client.get_status(created.transaction)
assert status.status == "success"
assert status.get("card_mask") == "411111******1111"
+
+
+@pytest.mark.sandbox
+async def test_new_features_against_the_sandbox_async():
+ if not sandbox_up():
+ pytest.skip(f"no sandbox on {SANDBOX_URL}")
+
+ httpx.patch(
+ f"{SANDBOX_URL}/_sandbox/merchants/{PUBLIC_KEY}",
+ json={"token_payments_enabled": True},
+ )
+
+ async with AsyncEpointClient(
+ PUBLIC_KEY, PRIVATE_KEY, base_url=SANDBOX_URL, language="en"
+ ) as client:
+ assert (await client.heartbeat()).ok
+
+ errored = await client.create_payment(amount=5, order_id=f"aio-{uuid.uuid4().hex[:10]}")
+ pay(errored.redirect_url, card="4000000000009998")
+ status = await client.get_status(errored.transaction)
+ assert status.status == "error"
+ assert status.message == "An error occurred during payment"
+ assert not status.ok
+
+ with pytest.raises(GatewayError):
+ await client.get_status("te9999999999")
+
+ split = await client.create_token_split_payment(
+ amount=100,
+ order_id=f"aio-{uuid.uuid4().hex[:10]}",
+ split_account="i000000002",
+ split_amount=30,
+ )
+ assert split.redirect_url
+ assert split.transaction
+ pay(split.redirect_url)
+ assert (await client.get_status(split.transaction)).status == "success"
+
+ with pytest.raises(GatewayError):
+ await client.create_token_split_payment(
+ amount=100,
+ order_id=f"aio-{uuid.uuid4().hex[:10]}",
+ split_account="i000009999",
+ split_amount=30,
+ )
+
+ order_id = f"aio-{uuid.uuid4().hex[:10]}"
+ form = client.checkout_form(amount=7, order_id=order_id)
+ async with httpx.AsyncClient() as http:
+ answer = await http.post(form.url, data=form.fields, follow_redirects=False)
+ assert answer.status_code == 303
+ pay(answer.headers["location"])
+ rows = (
+ await http.get(
+ f"{SANDBOX_URL}/_sandbox/transactions",
+ params={"merchant": PUBLIC_KEY, "limit": 200},
+ )
+ ).json()
+ transaction = next(row["transaction"] for row in rows if row["order_id"] == order_id)
+ assert (await client.get_status(transaction)).status == "success"
+
+ invoice = await client.create_invoice(
+ amount=10, period_from="2026-01-01", period_to="2026-12-31"
+ )
+ await client.update_invoice(
+ invoice["id"],
+ amount=11,
+ period_from="2026-01-01",
+ period_to="2026-12-31",
+ tin="1234567892",
+ contract_number="C-88",
+ merchant_order_id="M-88",
+ )
+ viewed = (await client.get_invoice(invoice["id"]))["invoice"]
+ assert viewed["inn"] == "1234567892"
+ assert viewed["contract_number"] == "C-88"
+ assert viewed["merchant_order_id"] == "M-88"
diff --git a/tests/test_client_unit.py b/tests/test_client_unit.py
index 3d813bd..7c52fbc 100644
--- a/tests/test_client_unit.py
+++ b/tests/test_client_unit.py
@@ -1,5 +1,6 @@
import base64
import json
+from typing import Any
import httpx
import pytest
@@ -212,3 +213,190 @@ def handler(request: httpx.Request) -> httpx.Response:
client_with(handler).register_card()
assert "success_redirect_url" not in seen
assert "error_redirect_url" not in seen
+
+
+def recording_client(seen: dict, response: Any = None) -> EpointClient:
+ def handler(request: httpx.Request) -> httpx.Response:
+ seen.clear()
+ seen["path"] = request.url.path
+ if request.headers.get("content-type", "").startswith("application/json"):
+ body = json.loads(request.content)
+ seen["payload"] = json.loads(base64.b64decode(body["data"]))
+ else:
+ seen["payload"], _ = decode_sent(request)
+ return httpx.Response(200, json={"status": "success"} if response is None else response)
+
+ return EpointClient(
+ PUBLIC_KEY,
+ PRIVATE_KEY,
+ base_url="https://epoint.test",
+ success_redirect_url="https://shop.test/default-ok",
+ error_redirect_url="https://shop.test/default-fail",
+ transport=httpx.Client(transport=httpx.MockTransport(handler)),
+ )
+
+
+def test_get_status_returns_a_payment_that_ended_in_error():
+ body = {
+ "status": "error",
+ "code": "",
+ "message": "An error occurred during payment",
+ "transaction": "te1",
+ }
+ result = client_with(lambda request: httpx.Response(200, json=body)).get_status("te1")
+ assert result.status == "error"
+ assert result.message == "An error occurred during payment"
+ assert result.code == ""
+ assert not result.ok
+
+
+def test_other_calls_still_raise_on_error_with_a_transaction():
+ body = {"status": "error", "message": "bad", "transaction": "te1"}
+ with pytest.raises(GatewayError):
+ client_with(lambda request: httpx.Response(200, json=body)).create_payment(1, "o")
+
+
+def test_token_split_payment_payload():
+ seen: dict = {}
+ client = recording_client(seen)
+ client.create_token_split_payment(
+ amount=100, order_id="o-1", split_account="i000000002", split_amount=30
+ )
+ assert seen["path"] == "/api/1/token/split-payment"
+ payload = seen["payload"]
+ assert payload["amount"] == "100.00"
+ assert payload["split_user"] == "i000000002"
+ assert payload["split_amount"] == "30.00"
+ assert payload["currency"] == "AZN"
+ assert payload["language"] == "az"
+ assert payload["order_id"] == "o-1"
+ assert payload["success_redirect_url"] == "https://shop.test/default-ok"
+ assert payload["error_redirect_url"] == "https://shop.test/default-fail"
+
+ client.create_token_split_payment(
+ amount=10,
+ order_id="o-2",
+ split_account="i000000002",
+ split_amount=3,
+ description="share",
+ success_redirect_url="https://shop.test/ok",
+ other_attr={"k": "v"},
+ )
+ assert seen["payload"]["success_redirect_url"] == "https://shop.test/ok"
+ assert seen["payload"]["error_redirect_url"] == "https://shop.test/default-fail"
+ assert seen["payload"]["description"] == "share"
+ assert seen["payload"]["other_attr"] == {"k": "v"}
+
+
+def test_checkout_form_is_signed_and_makes_no_request():
+ def handler(request: httpx.Request) -> httpx.Response:
+ raise AssertionError("checkout_form must not send a request")
+
+ client = EpointClient(
+ PUBLIC_KEY,
+ PRIVATE_KEY,
+ base_url="https://epoint.test/",
+ success_redirect_url="https://shop.test/default-ok",
+ transport=httpx.Client(transport=httpx.MockTransport(handler)),
+ )
+ form = client.checkout_form(
+ amount=30.75, order_id="o-1", description="Kilim", is_installment=True
+ )
+ assert form.url == "https://epoint.test/api/1/checkout"
+ assert matches(PRIVATE_KEY, form.data, form.signature)
+ assert form.fields == {"data": form.data, "signature": form.signature}
+ payload = json.loads(base64.b64decode(form.data))
+ assert payload == {
+ "public_key": PUBLIC_KEY,
+ "amount": "30.75",
+ "currency": "AZN",
+ "language": "az",
+ "order_id": "o-1",
+ "description": "Kilim",
+ "success_redirect_url": "https://shop.test/default-ok",
+ "is_installment": 1,
+ }
+
+
+@pytest.mark.parametrize(("value", "sent"), [(True, 1), (False, 0), (None, None)])
+def test_is_installment_is_sent_as_an_integer_or_omitted(value, sent):
+ seen: dict = {}
+ client = recording_client(seen)
+ for call, path in (
+ (client.create_payment, "/api/1/request"),
+ (client.create_payment_request, "/api/1/payment-request"),
+ ):
+ call(amount=1, order_id="o", is_installment=value)
+ assert seen["path"] == path
+ assert seen["payload"].get("is_installment") == sent
+ assert ("is_installment" in seen["payload"]) == (value is not None)
+
+ form = client.checkout_form(amount=1, order_id="o", is_installment=value)
+ payload = json.loads(base64.b64decode(form.data))
+ assert payload.get("is_installment") == sent
+ assert ("is_installment" in payload) == (value is not None)
+
+
+def test_bank_transfer_sends_bulk_description():
+ seen: dict = {}
+ recording_client(seen).create_bank_transfer(
+ order_id="b1",
+ amount=100,
+ description="fees",
+ iban="AZ21NABZ00000000137010001944",
+ name="ACME",
+ tin="1234567892",
+ bank_code="505141",
+ bulk_description="July payroll",
+ )
+ assert seen["path"] == "/api/1/b2b/payment"
+ assert seen["payload"]["bulk_description"] == "July payroll"
+
+
+def test_installment_plans_send_description_and_other_attr():
+ seen: dict = {}
+ recording_client(seen, []).get_installment_plans(
+ order_id="o", amount=200, description="sofa", other_attr={"k": "v"}
+ )
+ assert seen["path"] == "/api/1/get-installment-request"
+ assert seen["payload"]["description"] == "sofa"
+ assert seen["payload"]["other_attr"] == {"k": "v"}
+
+
+def test_pay_by_installment_sends_other_attr():
+ seen: dict = {}
+ recording_client(seen).pay_by_installment(
+ amount=200,
+ order_id="o",
+ installment_card_id="kapital",
+ installment_month=6,
+ other_attr={"k": "v"},
+ )
+ assert seen["path"] == "/api/1/installment-request"
+ assert seen["payload"]["other_attr"] == {"k": "v"}
+ assert seen["payload"]["installment_month"] == 6
+
+
+def test_update_invoice_sends_tin_contract_and_merchant_order():
+ seen: dict = {}
+ client = recording_client(seen)
+ client.update_invoice(
+ 7,
+ amount=11,
+ period_from="2026-01-01",
+ period_to="2026-12-31",
+ tin="1234567892",
+ contract_number="C-1",
+ merchant_order_id="M-1",
+ )
+ assert seen["path"] == "/api/1/invoices/update"
+ payload = seen["payload"]
+ assert payload["id"] == 7
+ assert payload["inn"] == "1234567892"
+ assert payload["contract_number"] == "C-1"
+ assert payload["merchant_order_id"] == "M-1"
+
+ client.update_invoice(7, amount=11, period_from="2026-01-01", period_to="2026-12-31")
+ assert "inn" not in seen["payload"]
+ assert "contract_number" not in seen["payload"]
+ assert "merchant_order_id" not in seen["payload"]
diff --git a/tests/test_models.py b/tests/test_models.py
new file mode 100644
index 0000000..5e1ef9a
--- /dev/null
+++ b/tests/test_models.py
@@ -0,0 +1,204 @@
+import json
+
+import pytest
+
+from epoint import (
+ OK_STATUSES,
+ B2BWebhook,
+ B2BWebhookStatus,
+ Callback,
+ EpointError,
+ GatewayError,
+ TransportError,
+ parse_b2b_webhook,
+)
+from epoint._transport import parse
+from epoint.models import Response
+
+
+def test_get_status_returns_a_payment_that_ended_in_error():
+ body = {"status": "error", "code": "", "message": "An error occurred", "transaction": "te1"}
+ result = parse(200, body, payment_status=True)
+ assert result.status == "error"
+ assert result.transaction == "te1"
+ assert not result.ok
+
+
+def test_get_status_still_raises_for_an_error_with_no_transaction():
+ with pytest.raises(GatewayError) as info:
+ parse(200, {"status": "error", "message": "Unknown transaction"}, payment_status=True)
+ assert "Unknown transaction" in str(info.value)
+
+
+def test_get_status_still_raises_for_an_error_with_a_non_2xx_status():
+ body = {"status": "error", "message": "denied", "transaction": "te1"}
+ with pytest.raises(GatewayError):
+ parse(403, body, payment_status=True)
+
+
+def test_get_status_still_raises_transport_error_on_http_500():
+ with pytest.raises(TransportError):
+ parse(500, {"detail": "boom", "transaction": "te1"}, payment_status=True)
+
+
+def test_other_endpoints_still_raise_on_error_with_a_transaction():
+ with pytest.raises(GatewayError):
+ parse(200, {"status": "error", "message": "bad", "transaction": "te1"})
+
+
+@pytest.mark.parametrize("status", ["success", "SUCCESS", "ok"])
+def test_ok_is_true_for_the_success_statuses(status):
+ assert Response({"status": status}).ok
+ assert status in OK_STATUSES
+
+
+@pytest.mark.parametrize(
+ "status",
+ ["error", "failed", "server_error", "returned", "new", "FAILED", "PENDING", "OK", None],
+)
+def test_ok_is_false_for_everything_else(status):
+ assert not Response({"status": status}).ok
+
+
+def test_callback_ok_stays_lower_case_success_only():
+ base = {"order_id": "o", "transaction": "te1"}
+ assert Callback.from_dict({**base, "status": "success"}).ok
+ assert not Callback.from_dict({**base, "status": "SUCCESS"}).ok
+ assert not Callback.from_dict({**base, "status": "ok"}).ok
+
+
+def test_typed_accessors_read_the_documented_fields():
+ r = Response(
+ {
+ "status": "success",
+ "trace_id": "tr1",
+ "code": "000",
+ "bank_transaction": "BT1",
+ "bank_response": "000",
+ "operation_code": "100",
+ "rrn": 123456789012,
+ "card_name": "TESTER",
+ "card_mask": "411111******1111",
+ "amount": 25.5,
+ "card_id": "ce1",
+ "order_id": "o-1",
+ "widget_url": "https://epoint.test/widget",
+ }
+ )
+ assert r.trace_id == "tr1"
+ assert r.code == "000"
+ assert r.bank_transaction == "BT1"
+ assert r.bank_response == "000"
+ assert r.operation_code == "100"
+ assert r.rrn == "123456789012"
+ assert r.card_name == "TESTER"
+ assert r.card_mask == "411111******1111"
+ assert r.amount == 25.5
+ assert r.card_id == "ce1"
+ assert r.order_id == "o-1"
+ assert r.widget_url == "https://epoint.test/widget"
+ assert r.raw["rrn"] == 123456789012
+ assert r.get("rrn") == 123456789012
+
+
+def test_typed_accessors_are_none_when_missing():
+ r = Response({"status": "success"})
+ for name in (
+ "trace_id",
+ "code",
+ "bank_transaction",
+ "bank_response",
+ "operation_code",
+ "rrn",
+ "card_name",
+ "card_mask",
+ "amount",
+ "card_id",
+ "order_id",
+ "widget_url",
+ ):
+ assert getattr(r, name) is None
+
+
+@pytest.mark.parametrize(
+ ("value", "expected"),
+ [(10, 10.0), (30.75, 30.75), ("30.75", 30.75), (" 5 ", 5.0), ("abc", None), (True, None)],
+)
+def test_amount_accepts_a_number_or_a_numeric_string(value, expected):
+ assert Response({"amount": value}).amount == expected
+
+
+def test_callback_carries_the_new_optional_fields():
+ callback = Callback.from_dict(
+ {
+ "order_id": "o",
+ "status": "success",
+ "transaction": "te1",
+ "bank_response": "000",
+ "trace_id": "tr1",
+ "card_id": "ce1",
+ "amount": "12.50",
+ }
+ )
+ assert callback.bank_response == "000"
+ assert callback.trace_id == "tr1"
+ assert callback.card_id == "ce1"
+ assert callback.amount == 12.5
+
+
+def test_callback_new_fields_default_to_none():
+ callback = Callback.from_dict({"order_id": "o", "status": "success"})
+ assert callback.bank_response is None
+ assert callback.trace_id is None
+ assert callback.card_id is None
+
+
+WEBHOOK = {
+ "order_id": "b2b-1",
+ "status": "success",
+ "amount": 1250.0,
+ "payee_iban": "AZ21NABZ00000000137010001944",
+ "payee_name": "Logistika MMC",
+ "error_message": None,
+}
+
+
+def test_b2b_webhook_is_parsed_from_a_json_string():
+ hook = parse_b2b_webhook(json.dumps(WEBHOOK))
+ assert isinstance(hook, B2BWebhook)
+ assert hook.order_id == "b2b-1"
+ assert hook.status == B2BWebhookStatus.SUCCESS
+ assert hook.ok
+ assert hook.amount == 1250.0
+ assert hook.payee_iban == "AZ21NABZ00000000137010001944"
+ assert hook.payee_name == "Logistika MMC"
+ assert hook.error_message is None
+
+
+def test_b2b_webhook_accepts_bytes_and_a_dict():
+ assert parse_b2b_webhook(json.dumps(WEBHOOK).encode()).order_id == "b2b-1"
+ assert parse_b2b_webhook(WEBHOOK).order_id == "b2b-1"
+
+
+def test_b2b_webhook_failure_carries_the_error_message():
+ hook = parse_b2b_webhook(
+ {**WEBHOOK, "status": "failed", "amount": "1250.00", "error_message": "IBAN closed"}
+ )
+ assert hook.status == B2BWebhookStatus.FAILED
+ assert not hook.ok
+ assert hook.amount == 1250.0
+ assert hook.error_message == "IBAN closed"
+
+
+def test_b2b_webhook_ignores_unknown_fields_and_keeps_an_unknown_status():
+ hook = parse_b2b_webhook({**WEBHOOK, "status": "reversed", "extra": 1})
+ assert hook.status == "reversed"
+ assert not hook.ok
+ assert hook.raw["extra"] == 1
+
+
+def test_b2b_webhook_rejects_a_body_that_is_not_a_json_object():
+ with pytest.raises(EpointError):
+ parse_b2b_webhook("not json")
+ with pytest.raises(EpointError):
+ parse_b2b_webhook("[1, 2]")
diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py
index aecf6d0..8febd74 100644
--- a/tests/test_sandbox.py
+++ b/tests/test_sandbox.py
@@ -199,3 +199,151 @@ def submit(redirect_url: str, card: str) -> httpx.Response:
assert failed.status_code == 303
assert failed.headers["location"] == "https://shop.test/card-fail"
assert client.get_card_status(approved["card_id"]).get("status") == CardStatus.ACTIVE
+
+
+def enable_token_payments(client: EpointClient) -> None:
+ httpx.patch(
+ f"{client.base_url}/_sandbox/merchants/{client.public_key}",
+ json={"token_payments_enabled": True},
+ )
+
+
+def transaction_for(client: EpointClient, order_id: str) -> str:
+ rows = httpx.get(
+ f"{client.base_url}/_sandbox/transactions",
+ params={"merchant": client.public_key, "limit": 200},
+ ).json()
+ return next(row["transaction"] for row in rows if row["order_id"] == order_id)
+
+
+def test_a_payment_that_ended_in_error_reads_back_without_raising(client, pay):
+ created = client.create_payment(amount=5, order_id=order())
+ pay(client, created.redirect_url, card="4000000000009998")
+
+ status = client.get_status(created.transaction)
+ assert status.status == Status.ERROR
+ assert status.message == "An error occurred during payment"
+ assert status.code == ""
+ assert status.transaction == created.transaction
+ assert not status.ok
+
+
+def test_an_unknown_transaction_still_raises(client):
+ with pytest.raises(GatewayError):
+ client.get_status("te9999999999")
+
+
+def test_server_error_and_returned_read_back(client, pay):
+ timed_out = client.create_payment(amount=5, order_id=order())
+ pay(client, timed_out.redirect_url, card="4000000000009999")
+ server_error = client.get_status(timed_out.transaction)
+ assert server_error.status == Status.SERVER_ERROR
+ assert not server_error.ok
+
+ paid = client.create_payment(amount=5, order_id=order())
+ pay(client, paid.redirect_url)
+ client.reverse(paid.transaction)
+ returned = client.get_status(paid.transaction)
+ assert returned.status == Status.RETURNED
+ assert not returned.ok
+
+
+def test_typed_fields_read_from_a_settled_payment(client, pay):
+ order_id = order()
+ created = client.create_payment(amount=25.5, order_id=order_id)
+ assert created.trace_id
+ pay(client, created.redirect_url)
+
+ status = client.get_status(created.transaction)
+ assert status.ok
+ assert status.code == "000"
+ assert status.bank_response == "000"
+ assert status.bank_transaction
+ assert status.operation_code == "100"
+ assert status.rrn
+ assert status.card_mask == "411111******1111"
+ assert status.card_name == "TESTER"
+ assert status.amount == 25.5
+ assert status.trace_id
+
+
+def test_token_split_payment(client, pay):
+ enable_token_payments(client)
+ created = client.create_token_split_payment(
+ amount=100, order_id=order(), split_account=PARTNER_KEY, split_amount=30
+ )
+ assert created.ok
+ assert created.redirect_url
+ assert created.transaction
+ assert created.trace_id
+
+ pay(client, created.redirect_url)
+ assert client.get_status(created.transaction).status == Status.SUCCESS
+
+
+def test_token_split_payment_with_an_unknown_split_user_is_refused(client):
+ enable_token_payments(client)
+ with pytest.raises(GatewayError) as info:
+ client.create_token_split_payment(
+ amount=100, order_id=order(), split_account="i000009999", split_amount=30
+ )
+ assert "split_user" in str(info.value)
+
+
+def test_checkout_form_posts_to_the_card_page(client, pay):
+ order_id = order()
+ form = client.checkout_form(amount=12.5, order_id=order_id, is_installment=False)
+ assert form.url == f"{SANDBOX_URL}/api/1/checkout"
+
+ answer = httpx.post(form.url, data=form.fields, follow_redirects=False)
+ assert answer.status_code == 303
+ location = answer.headers["location"]
+ assert "/checkout/" in location
+
+ pay(client, location)
+ status = client.get_status(transaction_for(client, order_id))
+ assert status.status == Status.SUCCESS
+ assert status.amount == 12.5
+
+
+def test_updated_invoice_keeps_tin_contract_and_merchant_order(client):
+ created = client.create_invoice(amount=10, period_from="2026-01-01", period_to="2026-12-31")
+ invoice_id = created["id"]
+ updated = client.update_invoice(
+ invoice_id,
+ amount=11,
+ period_from="2026-01-01",
+ period_to="2026-12-31",
+ tin="1234567892",
+ contract_number="C-77",
+ merchant_order_id="M-77",
+ )
+ assert updated.ok
+
+ invoice = client.get_invoice(invoice_id)["invoice"]
+ assert invoice["inn"] == "1234567892"
+ assert invoice["contract_number"] == "C-77"
+ assert invoice["merchant_order_id"] == "M-77"
+
+
+def test_new_optional_parameters_are_accepted(client, pay):
+ grant_all(client)
+ plans = client.get_installment_plans(
+ order_id=order(), amount=200, description="sofa", other_attr={"k": "v"}
+ )
+ assert plans
+
+ request = client.create_payment_request(amount=5, order_id=order(), is_installment=True)
+ assert request.redirect_url
+
+ transfer = client.create_bank_transfer(
+ order_id=order(),
+ amount=100,
+ description="fees",
+ iban="AZ21NABZ00000000137010001944",
+ name="Logistika MMC",
+ tin="1234567892",
+ bank_code="505141",
+ bulk_description="July payroll",
+ )
+ assert transfer.ok