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
4 changes: 3 additions & 1 deletion src/shade/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
ShadeError,
SignatureVerificationError,
)
from .models import Merchant, ShadeObject
from .models import Merchant, ShadeObject, Transfer, TransferStatus

__version__ = "0.1.0"

Expand All @@ -39,6 +39,8 @@
"SignatureVerificationError",
"ShadeObject",
"SyncHTTPClient",
"Transfer",
"TransferStatus",
"config",
"api_base",
"environment",
Expand Down
3 changes: 2 additions & 1 deletion src/shade/models/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,5 +3,6 @@
"""
from .base import ShadeObject
from .merchant import Merchant
from .transfer import Transfer, TransferStatus

__all__ = ["Merchant", "ShadeObject"]
__all__ = ["Merchant", "ShadeObject", "Transfer", "TransferStatus"]
82 changes: 82 additions & 0 deletions src/shade/models/transfer.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
"""
Transfer model.

Represents a payout of funds from the merchant wallet to a destination
address on the Stellar network. Field names are converted from ``camelCase``
(backend/JSON) to ``snake_case`` (Python) via pydantic field aliases, matching
the convention established by :class:`~shade.models.merchant.Merchant`.
"""
from __future__ import annotations

from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import Optional

from pydantic import Field, field_validator
from stellar_sdk.strkey import StrKey

from .base import ShadeObject


class TransferStatus(str, Enum):
"""Lifecycle status of a transfer."""

PENDING = "pending"
PROCESSING = "processing"
COMPLETED = "completed"
FAILED = "failed"


class Transfer(ShadeObject):
"""A payout of funds from the merchant wallet to a destination address.

Build one from an API response with :meth:`ShadeObject.from_dict`, which
maps camelCase JSON keys to the snake_case fields below. ``asset`` falls
back to ``"XLM"`` when the API omits it (or sends it as ``null``), and
``status`` is always coerced to a :class:`TransferStatus` member.
"""

id: str
source_address: str = Field(alias="sourceAddress")
destination_address: str = Field(alias="destinationAddress")
amount: Decimal
asset: str = "XLM"
status: TransferStatus
stellar_tx_hash: Optional[str] = Field(default=None, alias="stellarTxHash")
fee: Optional[Decimal] = None
created_at: datetime = Field(alias="createdAt")

@field_validator("asset", mode="before")
@classmethod
def _default_asset(cls, value: object) -> object:
# Covers both a missing key (pydantic would already default it) and an
# API response that sends the key as an explicit null/empty string.
# Other falsy-but-wrong types (e.g. False, 0) are left alone so
# pydantic's normal type validation rejects them.
if value is None or value == "":
return "XLM"
return value

@field_validator("source_address", "destination_address")
@classmethod
def _validate_stellar_address(cls, value: str) -> str:
if not StrKey.is_valid_ed25519_public_key(value):
raise ValueError(
"must be a valid Stellar public key (starts with 'G', 56 characters)"
)
return value

@field_validator("amount")
@classmethod
def _amount_must_be_positive(cls, value: Decimal) -> Decimal:
if value <= 0:
raise ValueError("amount must be greater than 0")
return value

@field_validator("fee")
@classmethod
def _fee_must_not_be_negative(cls, value: Optional[Decimal]) -> Optional[Decimal]:
if value is not None and value < 0:
raise ValueError("fee must not be negative")
return value
154 changes: 154 additions & 0 deletions tests/test_transfer.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,154 @@
from datetime import datetime
from decimal import Decimal

import pytest
from stellar_sdk import Keypair

import shade
from shade import InvalidRequestError, ShadeObject, Transfer, TransferStatus

SOURCE_ADDRESS = Keypair.random().public_key
DESTINATION_ADDRESS = Keypair.random().public_key


def _api_response(**overrides):
"""A representative camelCase backend payload."""
data = {
"id": "trf_123",
"sourceAddress": SOURCE_ADDRESS,
"destinationAddress": DESTINATION_ADDRESS,
"amount": "150.25",
"asset": "USDC",
"status": "pending",
"stellarTxHash": None,
"fee": "0.5",
"createdAt": "2026-07-20T12:00:00Z",
}
data.update(overrides)
return data


def test_from_dict_maps_camelcase_to_snake_case():
transfer = Transfer.from_dict(_api_response())

assert transfer.id == "trf_123"
assert transfer.source_address == SOURCE_ADDRESS
assert transfer.destination_address == DESTINATION_ADDRESS
assert transfer.amount == Decimal("150.25")
assert transfer.asset == "USDC"
assert transfer.status == TransferStatus.PENDING
assert transfer.stellar_tx_hash is None
assert transfer.fee == Decimal("0.5")
assert transfer.created_at == datetime.fromisoformat("2026-07-20T12:00:00+00:00")


def test_amount_and_fee_are_decimal():
transfer = Transfer.from_dict(_api_response(amount="99.99", fee="1.10"))
assert isinstance(transfer.amount, Decimal)
assert isinstance(transfer.fee, Decimal)


def test_asset_defaults_to_xlm_when_absent():
payload = _api_response()
del payload["asset"]
transfer = Transfer.from_dict(payload)
assert transfer.asset == "XLM"


def test_asset_defaults_to_xlm_when_null():
transfer = Transfer.from_dict(_api_response(asset=None))
assert transfer.asset == "XLM"


def test_malformed_asset_is_rejected_not_defaulted():
# False/0 are falsy but must not be silently coerced to "XLM" — they are
# the wrong type and should surface as a validation error instead.
with pytest.raises(InvalidRequestError) as exc_info:
Transfer.from_dict(_api_response(asset=False))
assert exc_info.value.param == "asset"


def test_status_is_transfer_status_enum():
for raw in ("pending", "processing", "completed", "failed"):
transfer = Transfer.from_dict(_api_response(status=raw))
assert isinstance(transfer.status, TransferStatus)
assert transfer.status.value == raw


def test_invalid_status_raises():
with pytest.raises(InvalidRequestError) as exc_info:
Transfer.from_dict(_api_response(status="bogus"))
assert exc_info.value.param == "status"


def test_completed_transfer_has_stellar_tx_hash():
transfer = Transfer.from_dict(
_api_response(status="completed", stellarTxHash="a" * 64)
)
assert transfer.status is TransferStatus.COMPLETED
assert transfer.stellar_tx_hash == "a" * 64


def test_fee_defaults_to_none():
payload = _api_response()
del payload["fee"]
transfer = Transfer.from_dict(payload)
assert transfer.fee is None


def test_transfer_is_exported_from_package():
assert shade.Transfer is Transfer
assert shade.TransferStatus is TransferStatus
assert issubclass(Transfer, ShadeObject)


def test_from_dict_requires_a_mapping():
with pytest.raises(InvalidRequestError):
Transfer.from_dict([("id", "x")]) # type: ignore[arg-type]


def test_missing_id_raises_clear_validation_error():
payload = _api_response()
del payload["id"]
with pytest.raises(InvalidRequestError) as exc_info:
Transfer.from_dict(payload)
assert exc_info.value.param == "id"


def test_invalid_source_address_is_rejected():
with pytest.raises(InvalidRequestError) as exc_info:
Transfer.from_dict(_api_response(sourceAddress="not-a-stellar-key"))
assert exc_info.value.param == "sourceAddress"


def test_invalid_destination_address_is_rejected():
with pytest.raises(InvalidRequestError) as exc_info:
Transfer.from_dict(_api_response(destinationAddress="not-a-stellar-key"))
assert exc_info.value.param == "destinationAddress"


def test_non_positive_amount_is_rejected():
with pytest.raises(InvalidRequestError) as exc_info:
Transfer.from_dict(_api_response(amount="0"))
assert exc_info.value.param == "amount"


def test_negative_fee_is_rejected():
with pytest.raises(InvalidRequestError) as exc_info:
Transfer.from_dict(_api_response(fee="-1"))
assert exc_info.value.param == "fee"


def test_from_dict_preserves_unknown_keys():
transfer = Transfer.from_dict(_api_response(description="Payout for order #9"))
assert transfer.to_dict()["description"] == "Payout for order #9"


def test_to_dict_round_trips_to_camelcase():
payload = _api_response()
transfer = Transfer.from_dict(payload)
round_tripped = transfer.to_dict()
assert round_tripped["sourceAddress"] == SOURCE_ADDRESS
assert round_tripped["destinationAddress"] == DESTINATION_ADDRESS
assert round_tripped["stellarTxHash"] is None
assert Transfer.from_dict(round_tripped) == transfer
Loading