Skip to content
Open
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
69 changes: 66 additions & 3 deletions code/sdk/python/ap2/sdk/constraints.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@

from abc import ABC, abstractmethod
from datetime import datetime, timedelta
from decimal import ROUND_FLOOR, Decimal

from ap2.sdk.generated.open_checkout_mandate import (
AllowedMerchants,
Expand All @@ -19,7 +20,7 @@
AgentRecurrence,
AllowedPayees,
AllowedPaymentInstruments,
AllowedPisps,

Check warning on line 23 in code/sdk/python/ap2/sdk/constraints.py

View workflow job for this annotation

GitHub Actions / spellcheck

Unknown word (Pisps)
AmountRange,
Budget,
ExecutionDate,
Expand All @@ -34,6 +35,66 @@
from pydantic import BaseModel


# ISO 4217 currencies whose minor-unit exponent differs from the default 2.
_ZERO_DECIMAL_CURRENCIES = frozenset(
{
'BIF',
'CLP',
'DJF',
'GNF',
'ISK',
'JPY',
'KMF',
'KRW',
'PYG',
'RWF',
'UGX',
'UYI',
'VND',
'VUV',
'XAF',
'XOF',
'XPF',
}
)
_THREE_DECIMAL_CURRENCIES = frozenset(
{
'BHD',
'IQD',
'JOD',
'KWD',
'LYD',
'OMR',
'TND',
}
)
_FOUR_DECIMAL_CURRENCIES = frozenset({'CLF', 'UYW'})


def minor_unit_exponent(currency: str) -> int:
"""Return the ISO 4217 minor-unit exponent for ``currency``."""
code = currency.upper()
if code in _ZERO_DECIMAL_CURRENCIES:
return 0
if code in _THREE_DECIMAL_CURRENCIES:
return 3
if code in _FOUR_DECIMAL_CURRENCIES:
return 4
return 2


def major_to_minor_units(value: float | int | str, currency: str) -> int:
"""Convert a major-unit amount to integer minor units for ``currency``.

Uses exact decimal arithmetic (``19.99`` USD -> ``1999``, not ``1998``)
and the ISO 4217 minor-unit exponent (``1000`` JPY -> ``1000``,
``1.5`` KWD -> ``1500``). Fractions of a minor unit are floored so a
budget can never be exceeded through rounding.
"""
scaled = Decimal(str(value)) * Decimal(10) ** minor_unit_exponent(currency)
return int(scaled.to_integral_value(rounding=ROUND_FLOOR))


class MandateContext(BaseModel):
"""Aggregated usage context for a mandate."""

Expand Down Expand Up @@ -251,10 +312,10 @@
return [f'Payment instrument {instrument.id} not in allowed list']


class AllowedPispEvaluator(PaymentConstraintEvaluator):

Check warning on line 315 in code/sdk/python/ap2/sdk/constraints.py

View workflow job for this annotation

GitHub Actions / spellcheck

Unknown word (Pisp)
"""Evaluates if the PISP is allowed."""

Check warning on line 316 in code/sdk/python/ap2/sdk/constraints.py

View workflow job for this annotation

GitHub Actions / spellcheck

Unknown word (PISP)

def __init__(self, constraint: AllowedPisps):

Check warning on line 318 in code/sdk/python/ap2/sdk/constraints.py

View workflow job for this annotation

GitHub Actions / spellcheck

Unknown word (Pisps)
self.constraint = constraint

def evaluate(
Expand All @@ -262,17 +323,17 @@
closed_mandate: PaymentMandate,
open_checkout_hash: str | None = None,
) -> list[str]:
if not closed_mandate.pisp:

Check warning on line 326 in code/sdk/python/ap2/sdk/constraints.py

View workflow job for this annotation

GitHub Actions / spellcheck

Unknown word (pisp)
return ['Missing PISP in closed mandate']

Check warning on line 327 in code/sdk/python/ap2/sdk/constraints.py

View workflow job for this annotation

GitHub Actions / spellcheck

Unknown word (PISP)

if any(
allowed.domain_name == closed_mandate.pisp.domain_name

Check warning on line 330 in code/sdk/python/ap2/sdk/constraints.py

View workflow job for this annotation

GitHub Actions / spellcheck

Unknown word (pisp)
and allowed.legal_name == closed_mandate.pisp.legal_name

Check warning on line 331 in code/sdk/python/ap2/sdk/constraints.py

View workflow job for this annotation

GitHub Actions / spellcheck

Unknown word (pisp)
and allowed.brand_name == closed_mandate.pisp.brand_name

Check warning on line 332 in code/sdk/python/ap2/sdk/constraints.py

View workflow job for this annotation

GitHub Actions / spellcheck

Unknown word (pisp)
for allowed in self.constraint.allowed
):
return []
return [f'PISP {closed_mandate.pisp} not in allowed list']

Check warning on line 336 in code/sdk/python/ap2/sdk/constraints.py

View workflow job for this annotation

GitHub Actions / spellcheck

Unknown word (PISP)


class BudgetEvaluator(PaymentConstraintEvaluator):
Expand Down Expand Up @@ -304,11 +365,13 @@
past_spend = self.mandate_context.total_amount
total_spend = past_spend + closed_mandate.payment_amount.amount

budget_max_cents = int(self.constraint.max * 100)
if total_spend > budget_max_cents:
budget_max_minor = major_to_minor_units(
self.constraint.max, self.constraint.currency
)
if total_spend > budget_max_minor:
return [
f'Cumulative spend {total_spend} exceeds '
f'budget limit {budget_max_cents} (past spend: {past_spend})'
f'budget limit {budget_max_minor} (past spend: {past_spend})'
]
return []

Expand Down
42 changes: 42 additions & 0 deletions code/sdk/python/ap2/tests/constraints_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -492,6 +492,48 @@ def test_payment_budget_missing_context():
)


def test_payment_budget_decimal_boundary_usd():
"""A budget of 19.99 USD admits a 1999-cent spend (no float truncation)."""
violations = check_payment_constraints(
_open_payment(
constraints=[
Budget(max=19.99, currency='USD'), # 1999 cents
]
),
_closed_payment(payment_amount=Amount(amount=1999, currency='USD')),
mandate_context=MandateContext(total_uses=0, total_amount=0),
)
assert violations == []


@pytest.mark.parametrize(
('budget_max', 'currency', 'amount', 'allowed'),
[
pytest.param(1000, 'JPY', 1000, True, id='jpy_zero_decimal_at_limit'),
pytest.param(1000, 'JPY', 1001, False, id='jpy_zero_decimal_over'),
pytest.param(50000, 'KRW', 50000, True, id='krw_zero_decimal_at_limit'),
pytest.param(1.5, 'KWD', 1500, True, id='kwd_three_decimal_at_limit'),
pytest.param(1.5, 'KWD', 1501, False, id='kwd_three_decimal_over'),
pytest.param(0.29, 'EUR', 29, True, id='eur_float_truncation'),
],
)
def test_payment_budget_iso4217_minor_units(
budget_max, currency, amount, allowed
):
"""Budget max (major units) is converted with the currency's exponent."""
violations = check_payment_constraints(
_open_payment(constraints=[Budget(max=budget_max, currency=currency)]),
_closed_payment(
payment_amount=Amount(amount=amount, currency=currency)
),
mandate_context=MandateContext(total_uses=0, total_amount=0),
)
if allowed:
assert violations == []
else:
assert any('exceeds budget limit' in v for v in violations)


# ── check_payment_constraints – execution_date ───────────────────────────


Expand Down
Loading