From 033384c9ccc00482512a8dd5f43ea670347af61b Mon Sep 17 00:00:00 2001 From: Chang Huan Date: Mon, 12 Jan 2026 22:23:33 -0800 Subject: [PATCH] Added payment app --- docs/apps/index.md | 1 + docs/apps/payment.md | 85 +++ pas/apps/__init__.py | 3 + pas/apps/payment/__init__.py | 33 + pas/apps/payment/app.py | 1103 +++++++++++++++++++++++++++++ pas/apps/payment/states.py | 648 +++++++++++++++++ tests/apps/test_payment_states.py | 529 ++++++++++++++ 7 files changed, 2402 insertions(+) create mode 100644 docs/apps/payment.md create mode 100644 pas/apps/payment/__init__.py create mode 100644 pas/apps/payment/app.py create mode 100644 pas/apps/payment/states.py create mode 100644 tests/apps/test_payment_states.py diff --git a/docs/apps/index.md b/docs/apps/index.md index 65f14a9..8ba8fde 100644 --- a/docs/apps/index.md +++ b/docs/apps/index.md @@ -13,6 +13,7 @@ Use these pages to explore the current user-facing tool surface for each PAS sta - [Stateful Reminder App](./reminder.md) - [Stateful Shopping App](./shopping.md) - [Stateful Note App](./notes.md) +- [Stateful Payment App](./payment.md) ## Navigation Framework Recap diff --git a/docs/apps/payment.md b/docs/apps/payment.md new file mode 100644 index 0000000..a7d8d42 --- /dev/null +++ b/docs/apps/payment.md @@ -0,0 +1,85 @@ +# Stateful Payment App + +`pas.apps.payment.app.StatefulPaymentApp` extends PAS navigation with a Payment application that manages payments, requests, transactions, contacts, and payment methods. It launches in `HomeView` and navigates between different views based on user actions and completed backend operations. + +## Navigation States + +### HomeView + +| Tool | Backend call(s) | Returns | Navigation effect | +| --- | --- | --- | --- | +| `get_feed()` | `StatefulPaymentApp.get_feed()` | List of feed items with transaction details, sorted by date (newest first) | Remains in `HomeView` | +| `get_balance()` | `StatefulPaymentApp.get_balance()` | Dictionary containing balance amount and currency | Remains in `HomeView` | +| `view_contacts()` | `StatefulPaymentApp.list_contacts()` | List of user's Payment contacts with names and usernames | Completed event transitions to `ContactListView` | +| `view_transactions()` | `StatefulPaymentApp.list_transactions()` | List of all user's transactions sorted by date | Completed event transitions to `TransactionListView` | +| `view_payment_methods()` | `StatefulPaymentApp.list_payment_methods()` | List of linked payment methods (bank accounts, cards) | Completed event transitions to `PaymentMethodsView` | + +### PaymentView + +| Tool | Backend call(s) | Returns | Navigation effect | +| --- | --- | --- | --- | +| `search_users(query: str)` | `StatefulPaymentApp.search_users(query=query)` | List of matching user profiles | Remains in `PaymentView` | +| `get_user(user_id: str)` | `StatefulPaymentApp.get_user(user_id=user_id)` | User profile information including name, username, and profile picture | Remains in `PaymentView` | +| `send_payment(recipient_id: str, amount: float, note: str, privacy: str = "friends", payment_method_id: str | None = None)` | `StatefulPaymentApp.send_payment(...)` | Transaction ID of the completed payment | Completed event transitions to `TransactionDetail(transaction_id)` | +| `request_payment(recipient_id: str, amount: float, note: str, privacy: str = "friends")` | `StatefulPaymentApp.request_payment(...)` | Request ID of the created payment request | Completed event transitions to `TransactionDetail(transaction_id)` | + +### TransactionListView + +| Tool | Backend call(s) | Returns | Navigation effect | +| --- | --- | --- | --- | +| `list_transactions(filter_type: str | None = None)` | `StatefulPaymentApp.list_transactions(filter_type=filter_type)` | List of transaction summaries sorted by date (newest first) | Remains in `TransactionListView` | +| `open_transaction(transaction_id: str)` | `StatefulPaymentApp.get_transaction(transaction_id=transaction_id)` | Complete transaction details including participants, amount, and status | Completed event transitions to `TransactionDetail(transaction_id)` | +| `list_pending_requests()` | `StatefulPaymentApp.list_pending_requests()` | List of pending requests that need action | Remains in `TransactionListView` | + +### TransactionDetail + +| Tool | Backend call(s) | Returns | Navigation effect | +| --- | --- | --- | --- | +| `get_transaction(transaction_id: str)` | `StatefulPaymentApp.get_transaction(transaction_id=transaction_id)` | Complete transaction information including sender, recipient, amount, note, status, and timestamps | Remains in `TransactionDetail` | +| `pay_request(request_id: str, payment_method_id: str | None = None)` | `StatefulPaymentApp.pay_request(request_id=request_id, payment_method_id=payment_method_id)` | Transaction ID of the completed payment | Remains in `TransactionDetail` (shows updated status) | +| `decline_request(request_id: str)` | `StatefulPaymentApp.decline_request(request_id=request_id)` | Confirmation message that request was declined | Completed event transitions to `TransactionListView` | +| `cancel_request(request_id: str)` | `StatefulPaymentApp.cancel_request(request_id=request_id)` | Confirmation message that request was cancelled | Completed event transitions to `TransactionListView` | + +### ContactListView + +| Tool | Backend call(s) | Returns | Navigation effect | +| --- | --- | --- | --- | +| `list_contacts()` | `StatefulPaymentApp.list_contacts()` | List of contact profiles with names and usernames | Remains in `ContactListView` | +| `search_users(query: str)` | `StatefulPaymentApp.search_users(query=query)` | List of matching user profiles | Remains in `ContactListView` | +| `open_contact(user_id: str)` | `StatefulPaymentApp.get_user_profile(user_id=user_id)` | User profile with recent transaction history | Completed event transitions to `UserProfile(user_id)` | +| `add_friend(user_id: str)` | `StatefulPaymentApp.add_friend(user_id=user_id)` | Confirmation message that friend request was sent | Remains in `ContactListView` | +| `remove_friend(user_id: str)` | `StatefulPaymentApp.remove_friend(user_id=user_id)` | Confirmation message that friend was removed | Remains in `ContactListView` | + +### UserProfile + +| Tool | Backend call(s) | Returns | Navigation effect | +| --- | --- | --- | --- | +| `get_user_profile(user_id: str)` | `StatefulPaymentApp.get_user_profile(user_id=user_id)` | User profile including name, username, friend status, and recent transaction history with this user | Remains in `UserProfile` | +| `get_transaction_history(user_id: str)` | `StatefulPaymentApp.get_transaction_history(user_id=user_id)` | List of transactions between current user and specified user | Remains in `UserProfile` | +| `pay_user(user_id: str)` | Internal navigation helper | None | Transitions to `PaymentView(recipient_id=user_id)` | + +### PaymentMethodsView + +| Tool | Backend call(s) | Returns | Navigation effect | +| --- | --- | --- | --- | +| `list_payment_methods()` | `StatefulPaymentApp.list_payment_methods()` | List of payment methods including bank accounts and cards with masked numbers and verification status | Remains in `PaymentMethodsView` | +| `add_bank_account(account_number: str, routing_number: str, account_type: str)` | `StatefulPaymentApp.add_bank_account(...)` | Payment method ID of the newly added account | Remains in `PaymentMethodsView` | +| `add_card(card_number: str, expiry: str, cvv: str, billing_zip: str)` | `StatefulPaymentApp.add_card(...)` | Payment method ID of the newly added card | Remains in `PaymentMethodsView` | +| `set_default_payment_method(payment_method_id: str)` | `StatefulPaymentApp.set_default_payment_method(payment_method_id=payment_method_id)` | Confirmation message with the updated default method | Remains in `PaymentMethodsView` | +| `remove_payment_method(payment_method_id: str)` | `StatefulPaymentApp.remove_payment_method(payment_method_id=payment_method_id)` | Confirmation message that payment method was removed | Remains in `PaymentMethodsView` | +| `view_transfer()` | `StatefulPaymentApp.get_balance()` | Current balance information | Completed event transitions to `TransferView` | + +### TransferView + +| Tool | Backend call(s) | Returns | Navigation effect | +| --- | --- | --- | --- | +| `get_balance()` | `StatefulPaymentApp.get_balance()` | Dictionary containing balance amount and available transfer options | Remains in `TransferView` | +| `transfer_to_bank(amount: float, bank_account_id: str, speed: str = "standard")` | `StatefulPaymentApp.transfer_to_bank(amount=amount, bank_account_id=bank_account_id, speed=speed)` | Confirmation message with transfer amount and estimated completion time | Remains in `TransferView` | +| `add_money_from_bank(amount: float, bank_account_id: str)` | `StatefulPaymentApp.add_money_from_bank(amount=amount, bank_account_id=bank_account_id)` | Confirmation message with transfer amount and estimated completion time | Remains in `TransferView` | + +## Navigation Helpers + +- `go_back()` appears automatically when navigation history exists and pops to the prior screen, returning messages such as `Navigated back to the state HomeView`. +- All forward transitions are triggered by completed Meta-ARE events via `handle_state_transition()`. +- `PaymentView` can be initialized with a pre-selected `recipient_id` when navigating from `UserProfile`. +- `TransactionDetail` remains in place after `pay_request` to show the updated transaction status, but transitions back to `TransactionListView` after declining or cancelling requests. diff --git a/pas/apps/__init__.py b/pas/apps/__init__.py index c2288be..afa5e56 100644 --- a/pas/apps/__init__.py +++ b/pas/apps/__init__.py @@ -12,6 +12,7 @@ from pas.apps.email.app import StatefulEmailApp from pas.apps.messaging.app import StatefulMessagingApp from pas.apps.note import StatefulNotesApp +from pas.apps.payment import StatefulPaymentApp from pas.apps.proactive_aui import PASAgentUserInterface from pas.apps.reminder import StatefulReminderApp from pas.apps.shopping import StatefulShoppingApp @@ -31,6 +32,7 @@ "StatefulEmailApp", "StatefulMessagingApp", "StatefulNotesApp", + "StatefulPaymentApp", "StatefulReminderApp", "StatefulShoppingApp", "user_tool", @@ -48,4 +50,5 @@ StatefulMessagingApp, StatefulNotesApp, StatefulReminderApp, + StatefulPaymentApp, ] diff --git a/pas/apps/payment/__init__.py b/pas/apps/payment/__init__.py new file mode 100644 index 0000000..8c24616 --- /dev/null +++ b/pas/apps/payment/__init__.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +from pas.apps.payment.app import ( + PaymentMethod, + PaymentUser, + StatefulPaymentApp, + Transaction, +) +from pas.apps.payment.states import ( + ContactListView, + HomeView, + PaymentMethodsView, + PaymentView, + TransactionDetail, + TransactionListView, + TransferView, + UserProfile, +) + +__all__ = [ + "ContactListView", + "HomeView", + "PaymentMethod", + "PaymentMethodsView", + "PaymentUser", + "PaymentView", + "StatefulPaymentApp", + "Transaction", + "TransactionDetail", + "TransactionListView", + "TransferView", + "UserProfile", +] diff --git a/pas/apps/payment/app.py b/pas/apps/payment/app.py new file mode 100644 index 0000000..1031ddb --- /dev/null +++ b/pas/apps/payment/app.py @@ -0,0 +1,1103 @@ +"""Stateful Payment app with PAS navigation.""" + +from __future__ import annotations + +import contextlib +import logging +import textwrap +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import TYPE_CHECKING, Any + +from are.simulation.tool_utils import OperationType, app_tool, data_tool +from are.simulation.types import EventType +from are.simulation.utils import get_state_dict, uuid_hex +from are.simulation.utils.type_utils import type_check + +if TYPE_CHECKING: + from are.simulation.types import CompletedEvent + +from pas.apps.core import StatefulApp +from pas.apps.payment.states import ( + ContactListView, + HomeView, + PaymentMethodsView, + PaymentView, + TransactionDetail, + TransactionListView, + TransferView, + UserProfile, +) +from pas.apps.tool_decorators import pas_event_registered + +logger = logging.getLogger(__name__) + + +@dataclass +class PaymentUser: + """Payment user profile model.""" + + user_id: str + username: str + display_name: str + phone: str = "" + email: str = "" + profile_picture: str = "" + is_friend: bool = False + + def __str__(self) -> str: + return textwrap.dedent( + f""" + ID: {self.user_id} + Username: @{self.username} + Name: {self.display_name} + Phone: {self.phone} + Email: {self.email} + Friend: {self.is_friend} + """ + ) + + def get_state(self) -> dict[str, Any]: + """Serialize user state.""" + return get_state_dict( + self, ["user_id", "username", "display_name", "phone", "email", "profile_picture", "is_friend"] + ) + + def load_state(self, state_dict: dict[str, Any]) -> None: + """Restore user from serialized state.""" + self.user_id = state_dict["user_id"] + self.username = state_dict["username"] + self.display_name = state_dict["display_name"] + self.phone = state_dict.get("phone", "") + self.email = state_dict.get("email", "") + self.profile_picture = state_dict.get("profile_picture", "") + self.is_friend = state_dict.get("is_friend", False) + + +@dataclass +class PaymentMethod: + """Payment method model (bank account or card).""" + + payment_method_id: str + method_type: str # "bank_account" or "card" + last_four: str + name: str = "" + is_default: bool = False + verified: bool = False + + def __str__(self) -> str: + return textwrap.dedent( + f""" + ID: {self.payment_method_id} + Type: {self.method_type} + Name: {self.name} + Last Four: ****{self.last_four} + Default: {self.is_default} + Verified: {self.verified} + """ + ) + + def get_state(self) -> dict[str, Any]: + """Serialize payment method state.""" + return get_state_dict(self, ["payment_method_id", "method_type", "last_four", "name", "is_default", "verified"]) + + def load_state(self, state_dict: dict[str, Any]) -> None: + """Restore payment method from serialized state.""" + self.payment_method_id = state_dict["payment_method_id"] + self.method_type = state_dict["method_type"] + self.last_four = state_dict["last_four"] + self.name = state_dict.get("name", "") + self.is_default = state_dict.get("is_default", False) + self.verified = state_dict.get("verified", False) + + +@dataclass +class Transaction: + """Transaction model for payments and requests.""" + + transaction_id: str + transaction_type: str # "payment", "request", "transfer" + sender_id: str + recipient_id: str + amount: float + note: str + status: str # "completed", "pending", "cancelled", "declined" + privacy: str # "public", "friends", "private" + created_at: datetime | float + payment_method_id: str | None = None + + def __str__(self) -> str: + created_str = ( + datetime.fromtimestamp(self.created_at, tz=UTC).strftime("%Y-%m-%d %H:%M:%S") + if isinstance(self.created_at, (int, float)) + else str(self.created_at) + ) + return textwrap.dedent( + f""" + ID: {self.transaction_id} + Type: {self.transaction_type} + From: {self.sender_id} + To: {self.recipient_id} + Amount: ${self.amount:.2f} + Note: {self.note} + Status: {self.status} + Created: {created_str} + """ + ) + + def get_state(self) -> dict[str, Any]: + """Serialize transaction state.""" + return { + "transaction_id": self.transaction_id, + "transaction_type": self.transaction_type, + "sender_id": self.sender_id, + "recipient_id": self.recipient_id, + "amount": self.amount, + "note": self.note, + "status": self.status, + "privacy": self.privacy, + "created_at": self.created_at.isoformat() if isinstance(self.created_at, datetime) else self.created_at, + "payment_method_id": self.payment_method_id, + } + + def load_state(self, state_dict: dict[str, Any]) -> None: + """Restore transaction from serialized state.""" + self.transaction_id = state_dict["transaction_id"] + self.transaction_type = state_dict["transaction_type"] + self.sender_id = state_dict["sender_id"] + self.recipient_id = state_dict["recipient_id"] + self.amount = state_dict["amount"] + self.note = state_dict["note"] + self.status = state_dict["status"] + self.privacy = state_dict["privacy"] + self.payment_method_id = state_dict.get("payment_method_id") + + if isinstance(state_dict["created_at"], str): + try: + self.created_at = datetime.fromisoformat(state_dict["created_at"]) + except ValueError: + # If ISO format parsing fails, try alternative format and convert to timestamp + try: + dt = datetime.strptime(state_dict["created_at"], "%Y-%m-%d %H:%M:%S") + self.created_at = dt.replace(tzinfo=UTC).timestamp() + except ValueError: + # Last resort: use current timestamp + self.created_at = datetime.now(UTC).timestamp() + else: + self.created_at = state_dict["created_at"] + + +@dataclass +class StatefulPaymentApp(StatefulApp): + """A Payment application that manages payments, requests, transactions, contacts, and payment methods with state-aware transitions. + + Key Features: + - User Management: Search users, manage contacts and friends + - Payment Management: Send money, request money with privacy settings + - Transaction Management: View history, pay/decline requests, cancel requests + - Payment Methods: Add/remove bank accounts and cards, set defaults + - Balance Management: View balance, transfer to/from bank accounts + - Feed: View public/friends transaction feed + + Notes: + - Current user is identified by user_id + - All monetary values are in USD + - Transaction IDs are automatically generated + - Privacy settings: "public", "friends", "private" + """ + + name: str | None = None + user_id: str = "" # Current user ID + users: dict[str, PaymentUser] = field(default_factory=dict) + transactions: dict[str, Transaction] = field(default_factory=dict) + payment_methods: dict[str, PaymentMethod] = field(default_factory=dict) + balance: float = 0.0 + + def __post_init__(self) -> None: + """Initialize the Payment app.""" + super().__init__(self.name or "payment") + self.load_root_state() + + def create_root_state(self) -> HomeView: + """Create the root navigation state.""" + return HomeView() + + def reset(self) -> None: + """Reset the app to empty state.""" + super().reset() + self.users.clear() + self.transactions.clear() + self.payment_methods.clear() + self.balance = 0.0 + + def _get_user_by_id(self, user_id: str) -> PaymentUser: + """Get user by ID with validation.""" + if user_id not in self.users: + raise KeyError(f"User {user_id} not found") + return self.users[user_id] + + def _get_transaction_by_id(self, transaction_id: str) -> Transaction: + """Get transaction by ID with validation.""" + if transaction_id not in self.transactions: + raise KeyError(f"Transaction {transaction_id} not found") + return self.transactions[transaction_id] + + def _get_payment_method_by_id(self, payment_method_id: str) -> PaymentMethod: + """Get payment method by ID with validation.""" + if payment_method_id not in self.payment_methods: + raise KeyError(f"Payment method {payment_method_id} not found") + return self.payment_methods[payment_method_id] + + @type_check + @data_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def create_user( + self, + username: str, + display_name: str, + phone: str = "", + email: str = "", + is_friend: bool = False, + ) -> str: + """Create a new Payment user.""" + if not isinstance(username, str) or len(username.strip()) == 0: + raise ValueError("Username must be non-empty string") + if not isinstance(display_name, str) or len(display_name.strip()) == 0: + raise ValueError("Display name must be non-empty string") + + user_id = uuid_hex(self.rng) + user = PaymentUser( + user_id=user_id, + username=username, + display_name=display_name, + phone=phone, + email=email, + is_friend=is_friend, + ) + self.users[user_id] = user + return user_id + + @type_check + @data_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def set_current_user(self, user_id: str) -> str: + """Set the current user for the app session.""" + if user_id not in self.users: + raise KeyError(f"User {user_id} not found") + self.user_id = user_id + return f"Current user set to {self.users[user_id].display_name}" + + @type_check + @data_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def create_transaction_with_time( + self, + transaction_type: str, + sender_id: str, + recipient_id: str, + amount: float, + note: str, + status: str, + created_at: str, + privacy: str = "friends", + payment_method_id: str | None = None, + ) -> str: + """Create a transaction with specific timestamp.""" + try: + timestamp = datetime.strptime(created_at, "%Y-%m-%d %H:%M:%S").replace(tzinfo=UTC).timestamp() + except ValueError as e: + raise ValueError("Invalid datetime format. Use YYYY-MM-DD HH:MM:SS") from e + + if transaction_type not in ["payment", "request", "transfer"]: + raise ValueError("Transaction type must be 'payment', 'request', or 'transfer'") + if status not in ["completed", "pending", "cancelled", "declined"]: + raise ValueError("Status must be 'completed', 'pending', 'cancelled', or 'declined'") + if privacy not in ["public", "friends", "private"]: + raise ValueError("Privacy must be 'public', 'friends', or 'private'") + if amount <= 0: + raise ValueError("Amount must be positive") + + transaction_id = uuid_hex(self.rng) + transaction = Transaction( + transaction_id=transaction_id, + transaction_type=transaction_type, + sender_id=sender_id, + recipient_id=recipient_id, + amount=amount, + note=note, + status=status, + privacy=privacy, + created_at=timestamp, + payment_method_id=payment_method_id, + ) + self.transactions[transaction_id] = transaction + return transaction_id + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT) + def get_balance(self) -> dict[str, Any]: + """Get current Payment balance.""" + return { + "balance": self.balance, + "currency": "USD", + } + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT) + def get_feed(self, limit: int = 20) -> list[dict[str, Any]]: + """Get transaction feed showing recent public/friends transactions.""" + if not isinstance(limit, int) or limit <= 0: + raise ValueError("Limit must be positive integer") + + visible_transactions = [ + t for t in self.transactions.values() if t.privacy in ["public", "friends"] and t.status == "completed" + ] + + sorted_transactions = sorted( + visible_transactions, + key=lambda t: t.created_at.timestamp() + if isinstance(t.created_at, datetime) + else (t.created_at if isinstance(t.created_at, (int, float)) else 0), + reverse=True, + )[:limit] + + return [ + { + "transaction_id": t.transaction_id, + "sender": self.users[t.sender_id].display_name if t.sender_id in self.users else "Unknown", + "recipient": self.users[t.recipient_id].display_name if t.recipient_id in self.users else "Unknown", + "amount": t.amount, + "note": t.note, + "created_at": t.created_at, + } + for t in sorted_transactions + ] + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT) + def search_users(self, query: str) -> list[dict[str, Any]]: + """Search for Payment users by username, name, or phone.""" + if not isinstance(query, str) or len(query.strip()) == 0: + raise ValueError("Query must be non-empty string") + + query_lower = query.lower() + return [ + { + "user_id": u.user_id, + "username": u.username, + "display_name": u.display_name, + "profile_picture": u.profile_picture, + "is_friend": u.is_friend, + } + for u in self.users.values() + if u.user_id != self.user_id + and (query_lower in u.username.lower() or query_lower in u.display_name.lower() or query_lower in u.phone) + ] + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT) + def get_user(self, user_id: str) -> dict[str, Any]: + """Get basic user profile information.""" + if not isinstance(user_id, str) or len(user_id) == 0: + raise ValueError("User ID must be non-empty string") + + user = self._get_user_by_id(user_id) + return { + "user_id": user.user_id, + "username": user.username, + "display_name": user.display_name, + "profile_picture": user.profile_picture, + "is_friend": user.is_friend, + } + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT) + def get_user_profile(self, user_id: str) -> dict[str, Any]: + """Get detailed Payment user profile with transaction history.""" + if not isinstance(user_id, str) or len(user_id) == 0: + raise ValueError("User ID must be non-empty string") + + user = self._get_user_by_id(user_id) + + user_transactions = [ + t + for t in self.transactions.values() + if (t.sender_id == self.user_id and t.recipient_id == user_id) + or (t.sender_id == user_id and t.recipient_id == self.user_id) + ] + + sorted_transactions = sorted( + user_transactions, + key=lambda t: t.created_at.timestamp() + if isinstance(t.created_at, datetime) + else (t.created_at if isinstance(t.created_at, (int, float)) else 0), + reverse=True, + )[:10] + + return { + "user_id": user.user_id, + "username": user.username, + "display_name": user.display_name, + "profile_picture": user.profile_picture, + "is_friend": user.is_friend, + "recent_transactions": [ + { + "transaction_id": t.transaction_id, + "type": t.transaction_type, + "amount": t.amount, + "note": t.note, + "status": t.status, + "created_at": t.created_at, + } + for t in sorted_transactions + ], + } + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.WRITE, event_type=EventType.AGENT) + def send_payment( + self, + recipient_id: str, + amount: float, + note: str, + privacy: str = "friends", + payment_method_id: str | None = None, + ) -> str: + """Send money to another user.""" + if not isinstance(recipient_id, str) or len(recipient_id) == 0: + raise ValueError("Recipient ID must be non-empty string") + if not isinstance(amount, (int, float)) or amount <= 0: + raise ValueError("Amount must be positive number") + if not isinstance(note, str) or len(note.strip()) == 0: + raise ValueError("Note must be non-empty string") + if privacy not in ["public", "friends", "private"]: + raise ValueError("Privacy must be 'public', 'friends', or 'private'") + + recipient = self._get_user_by_id(recipient_id) + + if payment_method_id and payment_method_id not in self.payment_methods: + raise KeyError(f"Payment method {payment_method_id} not found") + + if self.balance < amount and not payment_method_id: + raise ValueError("Insufficient balance") + + transaction_id = uuid_hex(self.rng) + transaction = Transaction( + transaction_id=transaction_id, + transaction_type="payment", + sender_id=self.user_id, + recipient_id=recipient_id, + amount=amount, + note=note, + status="completed", + privacy=privacy, + created_at=self.time_manager.time(), + payment_method_id=payment_method_id, + ) + + self.transactions[transaction_id] = transaction + + if not payment_method_id: + self.balance -= amount + + return transaction_id + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.WRITE, event_type=EventType.AGENT) + def request_payment( + self, + recipient_id: str, + amount: float, + note: str, + privacy: str = "friends", + ) -> str: + """Request money from another user.""" + if not isinstance(recipient_id, str) or len(recipient_id) == 0: + raise ValueError("Recipient ID must be non-empty string") + if not isinstance(amount, (int, float)) or amount <= 0: + raise ValueError("Amount must be positive number") + if not isinstance(note, str) or len(note.strip()) == 0: + raise ValueError("Note must be non-empty string") + if privacy not in ["public", "friends", "private"]: + raise ValueError("Privacy must be 'public', 'friends', or 'private'") + + recipient = self._get_user_by_id(recipient_id) + + transaction_id = uuid_hex(self.rng) + transaction = Transaction( + transaction_id=transaction_id, + transaction_type="request", + sender_id=self.user_id, + recipient_id=recipient_id, + amount=amount, + note=note, + status="pending", + privacy=privacy, + created_at=self.time_manager.time(), + ) + + self.transactions[transaction_id] = transaction + return transaction_id + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT) + def list_transactions(self, filter_type: str | None = None) -> list[dict[str, Any]]: + """List user's transaction history with optional filtering.""" + if filter_type and filter_type not in ["sent", "received", "pending", "all"]: + raise ValueError("Filter type must be 'sent', 'received', 'pending', or 'all'") + + user_transactions = [] + + for t in self.transactions.values(): + is_sender = t.sender_id == self.user_id + is_recipient = t.recipient_id == self.user_id + + if not (is_sender or is_recipient): + continue + + if filter_type == "sent" and not is_sender: + continue + if filter_type == "received" and not is_recipient: + continue + if filter_type == "pending" and t.status != "pending": + continue + + user_transactions.append(t) + + sorted_transactions = sorted( + user_transactions, + key=lambda t: t.created_at.timestamp() + if isinstance(t.created_at, datetime) + else (t.created_at if isinstance(t.created_at, (int, float)) else 0), + reverse=True, + ) + + return [ + { + "transaction_id": t.transaction_id, + "type": t.transaction_type, + "sender": self.users[t.sender_id].display_name if t.sender_id in self.users else "Unknown", + "recipient": self.users[t.recipient_id].display_name if t.recipient_id in self.users else "Unknown", + "amount": t.amount, + "note": t.note, + "status": t.status, + "created_at": t.created_at, + } + for t in sorted_transactions + ] + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT) + def get_transaction(self, transaction_id: str) -> dict[str, Any]: + """Get complete details of a specific transaction.""" + if not isinstance(transaction_id, str) or len(transaction_id) == 0: + raise ValueError("Transaction ID must be non-empty string") + + transaction = self._get_transaction_by_id(transaction_id) + + return { + "transaction_id": transaction.transaction_id, + "type": transaction.transaction_type, + "sender_id": transaction.sender_id, + "sender_name": self.users[transaction.sender_id].display_name + if transaction.sender_id in self.users + else "Unknown", + "recipient_id": transaction.recipient_id, + "recipient_name": self.users[transaction.recipient_id].display_name + if transaction.recipient_id in self.users + else "Unknown", + "amount": transaction.amount, + "note": transaction.note, + "status": transaction.status, + "privacy": transaction.privacy, + "created_at": transaction.created_at, + "payment_method_id": transaction.payment_method_id, + } + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT) + def list_pending_requests(self) -> list[dict[str, Any]]: + """List all pending payment requests (sent and received).""" + pending = [ + t + for t in self.transactions.values() + if t.transaction_type == "request" + and t.status == "pending" + and (t.sender_id == self.user_id or t.recipient_id == self.user_id) + ] + + sorted_pending = sorted( + pending, + key=lambda t: t.created_at.timestamp() + if isinstance(t.created_at, datetime) + else (t.created_at if isinstance(t.created_at, (int, float)) else 0), + reverse=True, + ) + + return [ + { + "transaction_id": t.transaction_id, + "direction": "incoming" if t.recipient_id == self.user_id else "outgoing", + "other_user": self.users[other_user_id].display_name + if (other_user_id := (t.sender_id if t.recipient_id == self.user_id else t.recipient_id)) in self.users + else "Unknown", + "amount": t.amount, + "note": t.note, + "created_at": t.created_at, + } + for t in sorted_pending + ] + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.WRITE, event_type=EventType.AGENT) + def pay_request(self, request_id: str, payment_method_id: str | None = None) -> str: + """Pay a pending payment request.""" + if not isinstance(request_id, str) or len(request_id) == 0: + raise ValueError("Request ID must be non-empty string") + + transaction = self._get_transaction_by_id(request_id) + + if transaction.transaction_type != "request": + raise ValueError("Transaction is not a request") + if transaction.status != "pending": + raise ValueError(f"Request is not pending (status: {transaction.status})") + if transaction.recipient_id != self.user_id: + raise ValueError("Cannot pay request not directed to you") + + if payment_method_id and payment_method_id not in self.payment_methods: + raise KeyError(f"Payment method {payment_method_id} not found") + + if self.balance < transaction.amount and not payment_method_id: + raise ValueError("Insufficient balance") + + transaction.status = "completed" + transaction.payment_method_id = payment_method_id + + if not payment_method_id: + self.balance -= transaction.amount + + return transaction.transaction_id + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.WRITE, event_type=EventType.AGENT) + def decline_request(self, request_id: str) -> str: + """Decline a pending payment request.""" + if not isinstance(request_id, str) or len(request_id) == 0: + raise ValueError("Request ID must be non-empty string") + + transaction = self._get_transaction_by_id(request_id) + + if transaction.transaction_type != "request": + raise ValueError("Transaction is not a request") + if transaction.status != "pending": + raise ValueError(f"Request is not pending (status: {transaction.status})") + if transaction.recipient_id != self.user_id: + raise ValueError("Cannot decline request not directed to you") + + transaction.status = "declined" + return f"Request {request_id} declined" + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.WRITE, event_type=EventType.AGENT) + def cancel_request(self, request_id: str) -> str: + """Cancel a pending outgoing payment request.""" + if not isinstance(request_id, str) or len(request_id) == 0: + raise ValueError("Request ID must be non-empty string") + + transaction = self._get_transaction_by_id(request_id) + + if transaction.transaction_type != "request": + raise ValueError("Transaction is not a request") + if transaction.status != "pending": + raise ValueError(f"Request is not pending (status: {transaction.status})") + if transaction.sender_id != self.user_id: + raise ValueError("Cannot cancel request you didn't send") + + transaction.status = "cancelled" + return f"Request {request_id} cancelled" + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT) + def list_contacts(self) -> list[dict[str, Any]]: + """List all user's Payment contacts/friends.""" + friends = [u for u in self.users.values() if u.is_friend and u.user_id != self.user_id] + + return [ + { + "user_id": u.user_id, + "username": u.username, + "display_name": u.display_name, + "profile_picture": u.profile_picture, + } + for u in friends + ] + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT) + def get_transaction_history(self, user_id: str) -> list[dict[str, Any]]: + """Get transaction history with a specific user.""" + if not isinstance(user_id, str) or len(user_id) == 0: + raise ValueError("User ID must be non-empty string") + + user = self._get_user_by_id(user_id) + + user_transactions = [ + t + for t in self.transactions.values() + if (t.sender_id == self.user_id and t.recipient_id == user_id) + or (t.sender_id == user_id and t.recipient_id == self.user_id) + ] + + sorted_transactions = sorted( + user_transactions, + key=lambda t: t.created_at.timestamp() + if isinstance(t.created_at, datetime) + else (t.created_at if isinstance(t.created_at, (int, float)) else 0), + reverse=True, + ) + + return [ + { + "transaction_id": t.transaction_id, + "type": t.transaction_type, + "amount": t.amount, + "note": t.note, + "status": t.status, + "created_at": t.created_at, + } + for t in sorted_transactions + ] + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.WRITE, event_type=EventType.AGENT) + def add_friend(self, user_id: str) -> str: + """Send friend request to a user.""" + if not isinstance(user_id, str) or len(user_id) == 0: + raise ValueError("User ID must be non-empty string") + + user = self._get_user_by_id(user_id) + + if user.is_friend: + raise ValueError(f"User {user.display_name} is already a friend") + + user.is_friend = True + return f"Friend request sent to {user.display_name}" + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.WRITE, event_type=EventType.AGENT) + def remove_friend(self, user_id: str) -> str: + """Remove a user from friends list.""" + if not isinstance(user_id, str) or len(user_id) == 0: + raise ValueError("User ID must be non-empty string") + + user = self._get_user_by_id(user_id) + + if not user.is_friend: + raise ValueError(f"User {user.display_name} is not a friend") + + user.is_friend = False + return f"Removed {user.display_name} from friends" + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT) + def list_payment_methods(self) -> list[dict[str, Any]]: + """List all linked payment methods.""" + return [ + { + "payment_method_id": pm.payment_method_id, + "type": pm.method_type, + "name": pm.name, + "last_four": pm.last_four, + "is_default": pm.is_default, + "verified": pm.verified, + } + for pm in self.payment_methods.values() + ] + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.WRITE, event_type=EventType.AGENT) + def add_bank_account(self, account_number: str, routing_number: str, account_type: str) -> str: + """Link a new bank account.""" + if not isinstance(account_number, str) or len(account_number) < 4: + raise ValueError("Invalid account number") + if not isinstance(routing_number, str) or len(routing_number) != 9: + raise ValueError("Routing number must be 9 digits") + if account_type not in ["checking", "savings"]: + raise ValueError("Account type must be 'checking' or 'savings'") + + payment_method_id = uuid_hex(self.rng) + payment_method = PaymentMethod( + payment_method_id=payment_method_id, + method_type="bank_account", + last_four=account_number[-4:], + name=f"{account_type.capitalize()} Account", + is_default=len(self.payment_methods) == 0, + verified=True, + ) + + self.payment_methods[payment_method_id] = payment_method + return payment_method_id + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.WRITE, event_type=EventType.AGENT) + def add_card(self, card_number: str, expiry: str, cvv: str, billing_zip: str) -> str: + """Link a new debit or credit card.""" + if not isinstance(card_number, str) or len(card_number) < 13: + raise ValueError("Invalid card number") + if not isinstance(expiry, str) or len(expiry) != 5: + raise ValueError("Expiry must be in MM/YY format") + if not isinstance(cvv, str) or len(cvv) not in [3, 4]: + raise ValueError("Invalid CVV") + if not isinstance(billing_zip, str) or len(billing_zip) < 5: + raise ValueError("Invalid billing ZIP code") + + payment_method_id = uuid_hex(self.rng) + payment_method = PaymentMethod( + payment_method_id=payment_method_id, + method_type="card", + last_four=card_number[-4:], + name=f"Card ending in {card_number[-4:]}", + is_default=len(self.payment_methods) == 0, + verified=True, + ) + + self.payment_methods[payment_method_id] = payment_method + return payment_method_id + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.WRITE, event_type=EventType.AGENT) + def set_default_payment_method(self, payment_method_id: str) -> str: + """Set a payment method as default.""" + if not isinstance(payment_method_id, str) or len(payment_method_id) == 0: + raise ValueError("Payment method ID must be non-empty string") + + payment_method = self._get_payment_method_by_id(payment_method_id) + + for pm in self.payment_methods.values(): + pm.is_default = False + + payment_method.is_default = True + return f"Set {payment_method.name} as default payment method" + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.WRITE, event_type=EventType.AGENT) + def remove_payment_method(self, payment_method_id: str) -> str: + """Remove a linked payment method.""" + if not isinstance(payment_method_id, str) or len(payment_method_id) == 0: + raise ValueError("Payment method ID must be non-empty string") + + payment_method = self._get_payment_method_by_id(payment_method_id) + name = payment_method.name + + del self.payment_methods[payment_method_id] + return f"Removed {name}" + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.WRITE, event_type=EventType.AGENT) + def transfer_to_bank(self, amount: float, bank_account_id: str, speed: str = "standard") -> str: + """Transfer money from Payment to bank account.""" + if not isinstance(amount, (int, float)) or amount <= 0: + raise ValueError("Amount must be positive number") + if not isinstance(bank_account_id, str) or len(bank_account_id) == 0: + raise ValueError("Bank account ID must be non-empty string") + if speed not in ["standard", "instant"]: + raise ValueError("Speed must be 'standard' or 'instant'") + + bank_account = self._get_payment_method_by_id(bank_account_id) + + if bank_account.method_type != "bank_account": + raise ValueError("Payment method is not a bank account") + + if self.balance < amount: + raise ValueError("Insufficient balance") + + self.balance -= amount + + eta = "1-3 business days" if speed == "standard" else "within 30 minutes" + return f"Transfer of ${amount:.2f} to {bank_account.name} initiated. ETA: {eta}" + + @type_check + @app_tool() + @pas_event_registered(operation_type=OperationType.WRITE, event_type=EventType.AGENT) + def add_money_from_bank(self, amount: float, bank_account_id: str) -> str: + """Add money from bank account to Payment balance.""" + if not isinstance(amount, (int, float)) or amount <= 0: + raise ValueError("Amount must be positive number") + if not isinstance(bank_account_id, str) or len(bank_account_id) == 0: + raise ValueError("Bank account ID must be non-empty string") + + bank_account = self._get_payment_method_by_id(bank_account_id) + + if bank_account.method_type != "bank_account": + raise ValueError("Payment method is not a bank account") + + self.balance += amount + return f"Added ${amount:.2f} from {bank_account.name}. ETA: 1-3 business days" + + def navigate_to_payment(self, user_id: str) -> None: + """Navigate to payment view with pre-selected user.""" + self.set_current_state(PaymentView(recipient_id=user_id)) + + def get_state(self) -> dict[str, Any]: + """Serialize complete app state.""" + return { + "user_id": self.user_id, + "users": {k: v.get_state() for k, v in self.users.items()}, + "transactions": {k: v.get_state() for k, v in self.transactions.items()}, + "payment_methods": {k: v.get_state() for k, v in self.payment_methods.items()}, + "balance": self.balance, + } + + def load_state(self, state_dict: dict[str, Any]) -> None: + """Restore app state from serialized data.""" + self.users.clear() + self.transactions.clear() + self.payment_methods.clear() + + self.user_id = state_dict.get("user_id", "") + self.balance = state_dict.get("balance", 0.0) + + for user_id, user_data in state_dict.get("users", {}).items(): + user = PaymentUser( + user_id=user_data["user_id"], + username=user_data["username"], + display_name=user_data["display_name"], + ) + user.load_state(user_data) + self.users[user_id] = user + + for transaction_id, transaction_data in state_dict.get("transactions", {}).items(): + created_at = transaction_data["created_at"] + if isinstance(created_at, str): + with contextlib.suppress(ValueError): + created_at = datetime.fromisoformat(created_at) + + transaction = Transaction( + transaction_id=transaction_data["transaction_id"], + transaction_type=transaction_data["transaction_type"], + sender_id=transaction_data["sender_id"], + recipient_id=transaction_data["recipient_id"], + amount=transaction_data["amount"], + note=transaction_data["note"], + status=transaction_data["status"], + privacy=transaction_data["privacy"], + created_at=created_at, + ) + transaction.load_state(transaction_data) + self.transactions[transaction_id] = transaction + + for pm_id, pm_data in state_dict.get("payment_methods", {}).items(): + payment_method = PaymentMethod( + payment_method_id=pm_data["payment_method_id"], + method_type=pm_data["method_type"], + last_four=pm_data["last_four"], + ) + payment_method.load_state(pm_data) + self.payment_methods[pm_id] = payment_method + + def handle_state_transition(self, event: CompletedEvent) -> None: + """Handle navigation state transitions based on user actions.""" + current_state = self.current_state + fname = event.function_name() + + if current_state is None or fname is None: + return + + action = event.action + args = action.resolved_args or action.args + metadata_value = event.metadata.return_value if event.metadata else None + + if isinstance(current_state, HomeView): + self._handle_home_view_transition(fname, args, metadata_value) + elif isinstance(current_state, PaymentView): + self._handle_payment_view_transition(fname, args, metadata_value) + elif isinstance(current_state, TransactionListView): + self._handle_transaction_list_transition(fname, args, metadata_value) + elif isinstance(current_state, TransactionDetail): + self._handle_transaction_detail_transition(fname, args, metadata_value) + elif isinstance(current_state, ContactListView): + self._handle_contact_list_transition(fname, args, metadata_value) + elif isinstance(current_state, UserProfile): + self._handle_user_profile_transition(current_state, fname, args, metadata_value) + elif isinstance(current_state, PaymentMethodsView): + self._handle_payment_methods_transition(fname, args, metadata_value) + elif isinstance(current_state, TransferView): + self._handle_transfer_view_transition(fname, args, metadata_value) + + def _handle_home_view_transition(self, fname: str, args: dict[str, Any], metadata: object | None) -> None: + """Handle transitions from home view state.""" + if fname == "view_transactions": + self.set_current_state(TransactionListView()) + elif fname == "view_contacts": + self.set_current_state(ContactListView()) + elif fname == "view_payment_methods": + self.set_current_state(PaymentMethodsView()) + + def _handle_payment_view_transition(self, fname: str, args: dict[str, Any], metadata: object | None) -> None: + """Handle transitions from payment view state.""" + if fname in ["send_payment", "request_payment"]: + transaction_id = metadata if isinstance(metadata, str) else None + if transaction_id: + self.set_current_state(TransactionDetail(transaction_id=transaction_id)) + + def _handle_transaction_list_transition(self, fname: str, args: dict[str, Any], metadata: object | None) -> None: + """Handle transitions from transaction list state.""" + if fname == "open_transaction": + transaction_id = args.get("transaction_id") + if transaction_id: + self.set_current_state(TransactionDetail(transaction_id=str(transaction_id))) + + def _handle_transaction_detail_transition(self, fname: str, args: dict[str, Any], metadata: object | None) -> None: + """Handle transitions from transaction detail state.""" + # After paying a request, stay in TransactionDetail to show updated status + # After declining or cancelling a request, return to TransactionListView + if fname in ["decline_request", "cancel_request"]: + self.set_current_state(TransactionListView()) + # pay_request stays in TransactionDetail to show the completed payment + + def _handle_contact_list_transition(self, fname: str, args: dict[str, Any], metadata: object | None) -> None: + """Handle transitions from contact list state.""" + if fname == "open_contact": + user_id = args.get("user_id") + if user_id: + self.set_current_state(UserProfile(user_id=str(user_id))) + + def _handle_user_profile_transition( + self, current_state: UserProfile, fname: str, args: dict[str, Any], metadata: object | None + ) -> None: + """Handle transitions from user profile state.""" + if fname == "pay_user": + # Use user_id from current state since we're viewing this user's profile + self.set_current_state(PaymentView(recipient_id=current_state.user_id)) + + def _handle_payment_methods_transition(self, fname: str, args: dict[str, Any], metadata: object | None) -> None: + """Handle transitions from payment methods view.""" + if fname == "view_transfer": + self.set_current_state(TransferView()) + + def _handle_transfer_view_transition(self, fname: str, args: dict[str, Any], metadata: object | None) -> None: + """Handle transitions from transfer view state.""" + pass diff --git a/pas/apps/payment/states.py b/pas/apps/payment/states.py new file mode 100644 index 0000000..9786411 --- /dev/null +++ b/pas/apps/payment/states.py @@ -0,0 +1,648 @@ +"""State definitions for the stateful Payment app.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, cast + +from are.simulation.types import OperationType, disable_events + +from pas.apps.core import AppState +from pas.apps.tool_decorators import pas_event_registered, user_tool + +if TYPE_CHECKING: + from pas.apps.payment.app import StatefulPaymentApp + + +class HomeView(AppState): + """Home state for viewing feed and navigating to main features. + + Users can view their transaction feed, access payment features, + and navigate to other sections of the app. + """ + + def on_enter(self) -> None: + """Called when entering home view state.""" + pass + + def on_exit(self) -> None: + """Called when leaving home view state.""" + pass + + @user_tool() + @pas_event_registered() + def get_feed(self) -> list[dict[str, Any]]: + """View transaction feed showing recent payments and requests. + + Returns: + List of feed items with transaction details, sorted by date (newest first). + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).get_feed() + + @user_tool() + @pas_event_registered() + def get_balance(self) -> dict[str, Any]: + """Get current Payment balance. + + Returns: + Dictionary containing balance amount and currency. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).get_balance() + + @user_tool() + @pas_event_registered() + def view_contacts(self) -> list[dict[str, Any]]: + """Navigate to contacts list. + + Returns: + List of user's Payment contacts with names and usernames. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).list_contacts() + + @user_tool() + @pas_event_registered() + def view_transactions(self) -> list[dict[str, Any]]: + """Navigate to transaction history. + + Returns: + List of all user's transactions sorted by date. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).list_transactions() + + @user_tool() + @pas_event_registered() + def view_payment_methods(self) -> list[dict[str, Any]]: + """Navigate to payment methods management. + + Returns: + List of linked payment methods (bank accounts, cards). + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).list_payment_methods() + + +class PaymentView(AppState): + """State for sending or requesting money. + + Users can select recipients, enter amounts, add notes, and choose + whether to pay or request money. + + Attributes: + recipient_id: Optional pre-selected recipient ID. + """ + + def __init__(self, recipient_id: str | None = None) -> None: + """Initialize payment view state. + + Args: + recipient_id: Optional unique identifier of pre-selected recipient. + """ + super().__init__() + self.recipient_id = recipient_id + + def on_enter(self) -> None: + """Called when entering payment view state.""" + pass + + def on_exit(self) -> None: + """Called when leaving payment view state.""" + pass + + @user_tool() + @pas_event_registered() + def search_users(self, query: str) -> list[dict[str, Any]]: + """Search for Payment users to pay or request from. + + Args: + query: Search term to match against username, name, or phone. + + Returns: + List of matching user profiles. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).search_users(query) + + @user_tool() + @pas_event_registered() + def get_user(self, user_id: str) -> dict[str, Any]: + """Get details of a specific user. + + Args: + user_id: Unique identifier of the user. + + Returns: + User profile information including name, username, and profile picture. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).get_user(user_id) + + @user_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def send_payment( + self, + recipient_id: str, + amount: float, + note: str, + privacy: str = "friends", + payment_method_id: str | None = None, + ) -> str: + """Send money to another user. + + Args: + recipient_id: Unique identifier of the recipient. + amount: Amount to send in USD. + note: Payment description or message. + privacy: Privacy setting ("public", "friends", "private"). + payment_method_id: Optional specific payment method to use. + + Returns: + Transaction ID of the completed payment. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).send_payment( + recipient_id, amount, note, privacy, payment_method_id + ) + + @user_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def request_payment( + self, + recipient_id: str, + amount: float, + note: str, + privacy: str = "friends", + ) -> str: + """Request money from another user. + + Args: + recipient_id: Unique identifier of the person to request from. + amount: Amount to request in USD. + note: Request description or reason. + privacy: Privacy setting ("public", "friends", "private"). + + Returns: + Request ID of the created payment request. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).request_payment(recipient_id, amount, note, privacy) + + +class TransactionListView(AppState): + """State for viewing transaction history. + + Users can view all their past transactions, filter by type, + and select individual transactions to view details. + """ + + def on_enter(self) -> None: + """Called when entering transaction list state.""" + pass + + def on_exit(self) -> None: + """Called when leaving transaction list state.""" + pass + + @user_tool() + @pas_event_registered() + def list_transactions(self, filter_type: str | None = None) -> list[dict[str, Any]]: + """List user's transaction history. + + Args: + filter_type: Optional filter ("sent", "received", "pending", "all"). + + Returns: + List of transaction summaries sorted by date (newest first). + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).list_transactions(filter_type) + + @user_tool() + @pas_event_registered() + def open_transaction(self, transaction_id: str) -> dict[str, Any]: + """Open detail page for a specific transaction. + + Args: + transaction_id: Unique identifier of the transaction. + + Returns: + Complete transaction details including participants, amount, and status. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).get_transaction(transaction_id) + + @user_tool() + @pas_event_registered() + def list_pending_requests(self) -> list[dict[str, Any]]: + """View all pending payment requests (sent and received). + + Returns: + List of pending requests that need action. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).list_pending_requests() + + +class TransactionDetail(AppState): + """State for viewing a specific transaction's details. + + Users can view complete transaction information, refund payments, + or take action on pending requests. + + Attributes: + transaction_id: ID of the transaction being viewed. + """ + + def __init__(self, transaction_id: str) -> None: + """Initialize transaction detail state. + + Args: + transaction_id: Unique identifier of the transaction. + """ + super().__init__() + self.transaction_id = transaction_id + + def on_enter(self) -> None: + """Called when entering transaction detail state.""" + pass + + def on_exit(self) -> None: + """Called when leaving transaction detail state.""" + pass + + @user_tool() + @pas_event_registered() + def get_transaction(self, transaction_id: str) -> dict[str, Any]: + """Fetch full transaction details. + + Args: + transaction_id: Unique identifier of the transaction. + + Returns: + Complete transaction information including sender, recipient, + amount, note, status, and timestamps. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).get_transaction(transaction_id) + + @user_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def pay_request(self, request_id: str, payment_method_id: str | None = None) -> str: + """Pay a pending payment request. + + Args: + request_id: Unique identifier of the request to pay. + payment_method_id: Optional specific payment method to use. + + Returns: + Transaction ID of the completed payment. + + Note: + Only pending incoming requests can be paid. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).pay_request(request_id, payment_method_id) + + @user_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def decline_request(self, request_id: str) -> str: + """Decline a pending payment request. + + Args: + request_id: Unique identifier of the request to decline. + + Returns: + Confirmation message that request was declined. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).decline_request(request_id) + + @user_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def cancel_request(self, request_id: str) -> str: + """Cancel a pending outgoing payment request. + + Args: + request_id: Unique identifier of the request to cancel. + + Returns: + Confirmation message that request was cancelled. + + Note: + Only pending outgoing requests can be cancelled. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).cancel_request(request_id) + + +class ContactListView(AppState): + """State for managing contacts and friends. + + Users can view their contacts, search for new users, + and manage friend connections. + """ + + def on_enter(self) -> None: + """Called when entering contact list state.""" + pass + + def on_exit(self) -> None: + """Called when leaving contact list state.""" + pass + + @user_tool() + @pas_event_registered() + def list_contacts(self) -> list[dict[str, Any]]: + """List all user's Payment contacts. + + Returns: + List of contact profiles with names and usernames. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).list_contacts() + + @user_tool() + @pas_event_registered() + def search_users(self, query: str) -> list[dict[str, Any]]: + """Search for Payment users by name, username, or phone. + + Args: + query: Search term to find users. + + Returns: + List of matching user profiles. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).search_users(query) + + @user_tool() + @pas_event_registered() + def open_contact(self, user_id: str) -> dict[str, Any]: + """Open a contact's profile page. + + Args: + user_id: Unique identifier of the contact. + + Returns: + User profile with recent transaction history. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).get_user_profile(user_id) + + @user_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def add_friend(self, user_id: str) -> str: + """Send friend request to a user. + + Args: + user_id: Unique identifier of the user to add. + + Returns: + Confirmation message that friend request was sent. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).add_friend(user_id) + + @user_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def remove_friend(self, user_id: str) -> str: + """Remove a user from friends list. + + Args: + user_id: Unique identifier of the friend to remove. + + Returns: + Confirmation message that friend was removed. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).remove_friend(user_id) + + +class UserProfile(AppState): + """State for viewing a specific user's profile. + + Users can view profile information, recent transactions with this user, + and initiate payments or requests. + + Attributes: + user_id: ID of the user profile being viewed. + """ + + def __init__(self, user_id: str) -> None: + """Initialize user profile state. + + Args: + user_id: Unique identifier of the user. + """ + super().__init__() + self.user_id = user_id + + def on_enter(self) -> None: + """Called when entering user profile state.""" + pass + + def on_exit(self) -> None: + """Called when leaving user profile state.""" + pass + + @user_tool() + @pas_event_registered() + def get_user_profile(self, user_id: str) -> dict[str, Any]: + """Get detailed user profile information. + + Args: + user_id: Unique identifier of the user. + + Returns: + User profile including name, username, friend status, and recent + transaction history with this user. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).get_user_profile(user_id) + + @user_tool() + @pas_event_registered() + def get_transaction_history(self, user_id: str) -> list[dict[str, Any]]: + """Get transaction history with this user. + + Args: + user_id: Unique identifier of the user. + + Returns: + List of transactions between current user and specified user. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).get_transaction_history(user_id) + + @user_tool() + @pas_event_registered() + def pay_user(self, user_id: str) -> None: + """Navigate to payment view with this user pre-selected. + + Args: + user_id: Unique identifier of the user to pay. + + Returns: + None (navigates to PaymentView state via transition logic). + """ + pass + + +class PaymentMethodsView(AppState): + """State for managing payment methods. + + Users can view linked bank accounts and cards, add new payment methods, + set default methods, and remove existing ones. + """ + + def on_enter(self) -> None: + """Called when entering payment methods state.""" + pass + + def on_exit(self) -> None: + """Called when leaving payment methods state.""" + pass + + @user_tool() + @pas_event_registered() + def list_payment_methods(self) -> list[dict[str, Any]]: + """List all linked payment methods. + + Returns: + List of payment methods including bank accounts and cards + with masked numbers and verification status. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).list_payment_methods() + + @user_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def add_bank_account(self, account_number: str, routing_number: str, account_type: str) -> str: + """Link a new bank account. + + Args: + account_number: Bank account number. + routing_number: Bank routing number. + account_type: Type of account ("checking" or "savings"). + + Returns: + Payment method ID of the newly added account. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).add_bank_account(account_number, routing_number, account_type) + + @user_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def add_card(self, card_number: str, expiry: str, cvv: str, billing_zip: str) -> str: + """Link a new debit or credit card. + + Args: + card_number: Card number. + expiry: Expiration date in MM/YY format. + cvv: Card security code. + billing_zip: Billing ZIP code. + + Returns: + Payment method ID of the newly added card. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).add_card(card_number, expiry, cvv, billing_zip) + + @user_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def set_default_payment_method(self, payment_method_id: str) -> str: + """Set a payment method as default. + + Args: + payment_method_id: Unique identifier of the payment method. + + Returns: + Confirmation message with the updated default method. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).set_default_payment_method(payment_method_id) + + @user_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def remove_payment_method(self, payment_method_id: str) -> str: + """Remove a linked payment method. + + Args: + payment_method_id: Unique identifier of the payment method to remove. + + Returns: + Confirmation message that payment method was removed. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).remove_payment_method(payment_method_id) + + @user_tool() + @pas_event_registered() + def view_transfer(self) -> dict[str, Any]: + """Navigate to transfer view for transferring money to/from bank accounts. + + Returns: + Current balance information. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).get_balance() + + +class TransferView(AppState): + """State for transferring money to/from bank accounts. + + Users can transfer their Payment balance to linked bank accounts + or add money from bank accounts to their Payment balance. + """ + + def on_enter(self) -> None: + """Called when entering transfer view state.""" + pass + + def on_exit(self) -> None: + """Called when leaving transfer view state.""" + pass + + @user_tool() + @pas_event_registered() + def get_balance(self) -> dict[str, Any]: + """Get current Payment balance. + + Returns: + Dictionary containing balance amount and available transfer options. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).get_balance() + + @user_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def transfer_to_bank(self, amount: float, bank_account_id: str, speed: str = "standard") -> str: + """Transfer money from Payment to bank account. + + Args: + amount: Amount to transfer in USD. + bank_account_id: Unique identifier of destination bank account. + speed: Transfer speed ("standard" for 1-3 days, "instant" for immediate). + + Returns: + Transfer ID and estimated completion time. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).transfer_to_bank(amount, bank_account_id, speed) + + @user_tool() + @pas_event_registered(operation_type=OperationType.WRITE) + def add_money_from_bank(self, amount: float, bank_account_id: str) -> str: + """Add money from bank account to Payment balance. + + Args: + amount: Amount to add in USD. + bank_account_id: Unique identifier of source bank account. + + Returns: + Transfer ID and estimated completion time. + """ + with disable_events(): + return cast("StatefulPaymentApp", self.app).add_money_from_bank(amount, bank_account_id) diff --git a/tests/apps/test_payment_states.py b/tests/apps/test_payment_states.py new file mode 100644 index 0000000..93d9d75 --- /dev/null +++ b/tests/apps/test_payment_states.py @@ -0,0 +1,529 @@ +"""Tests for the stateful payment app navigation flow.""" + +from __future__ import annotations + +from typing import Any + +import pytest +from are.simulation.types import ( + Action, + CompletedEvent, + EventMetadata, + EventType, +) + +from pas.apps.proactive_aui import PASAgentUserInterface +from pas.apps.payment.app import StatefulPaymentApp +from pas.apps.payment.states import ( + ContactListView, + HomeView, + PaymentMethodsView, + PaymentView, + TransactionDetail, + TransactionListView, + TransferView, + UserProfile, +) +from pas.apps.system import HomeScreenSystemApp +from pas.environment import StateAwareEnvironmentWrapper + + + +def _home_state(app: StatefulPaymentApp) -> HomeView: + """Assert and return app is in HomeView state.""" + state = app.current_state + assert isinstance(state, HomeView) + return state + + +def _payment_state(app: StatefulPaymentApp) -> PaymentView: + """Assert and return app is in PaymentView state.""" + state = app.current_state + assert isinstance(state, PaymentView) + return state + + +def _transaction_list_state(app: StatefulPaymentApp) -> TransactionListView: + """Assert and return app is in TransactionListView state.""" + state = app.current_state + assert isinstance(state, TransactionListView) + return state + + +def _transaction_detail_state(app: StatefulPaymentApp) -> TransactionDetail: + """Assert and return app is in TransactionDetail state.""" + state = app.current_state + assert isinstance(state, TransactionDetail) + return state + + +def _contact_list_state(app: StatefulPaymentApp) -> ContactListView: + """Assert and return app is in ContactListView state.""" + state = app.current_state + assert isinstance(state, ContactListView) + return state + + +def _payment_methods_state(app: StatefulPaymentApp) -> PaymentMethodsView: + """Assert and return app is in PaymentMethodsView state.""" + state = app.current_state + assert isinstance(state, PaymentMethodsView) + return state + + +def _make_event( + app: StatefulPaymentApp, + func: callable, + result: Any | None = None, + **kwargs: Any, +) -> CompletedEvent: + """Utility to build a minimal CompletedEvent for state transition tests. + + Args: + app: The StatefulPaymentApp instance + func: The function/method being called + result: Optional return value for metadata + **kwargs: Arguments to pass in the action + """ + action = Action(function=func, args={"self": app, **kwargs}, app=app) + + metadata = EventMetadata() + metadata.return_value = result + + return CompletedEvent( + event_type=EventType.USER, + action=action, + metadata=metadata, + event_time=0, + event_id="payment-test-event", + ) + + +@pytest.fixture +def payment_app() -> StatefulPaymentApp: + """Create a payment app with test data.""" + app = StatefulPaymentApp(name="payment") + + # Create test users + user1_id = app.create_user( + username="alice", + display_name="Alice Smith", + phone="555-0100", + email="alice@example.com", + is_friend=True, + ) + user2_id = app.create_user( + username="bob", + display_name="Bob Jones", + phone="555-0200", + email="bob@example.com", + is_friend=False, + ) + user3_id = app.create_user( + username="charlie", + display_name="Charlie Brown", + phone="555-0300", + email="charlie@example.com", + is_friend=True, + ) + + # Set current user + app.set_current_user(user1_id) + app.balance = 100.0 + + # Add payment methods + bank_id = app.add_bank_account( + account_number="1234567890", + routing_number="123456789", + account_type="checking", + ) + card_id = app.add_card( + card_number="4111111111111111", + expiry="12/25", + cvv="123", + billing_zip="12345", + ) + + # Create some transactions + app.create_transaction_with_time( + transaction_type="payment", + sender_id=user1_id, + recipient_id=user2_id, + amount=25.0, + note="Lunch", + status="completed", + created_at="2025-01-15 12:00:00", + privacy="friends", + ) + app.create_transaction_with_time( + transaction_type="request", + sender_id=user2_id, + recipient_id=user1_id, + amount=50.0, + note="Dinner", + status="pending", + created_at="2025-01-16 18:00:00", + privacy="friends", + ) + + return app + + +@pytest.fixture +def env_with_payment() -> StateAwareEnvironmentWrapper: + """Create environment with payment app registered and opened.""" + env = StateAwareEnvironmentWrapper() + system_app = HomeScreenSystemApp(name="HomeScreen") + aui_app = PASAgentUserInterface() + payment_app = StatefulPaymentApp(name="payment") + + # Create test user and set as current + user_id = payment_app.create_user( + username="testuser", + display_name="Test User", + phone="555-0000", + email="test@example.com", + ) + payment_app.set_current_user(user_id) + payment_app.balance = 50.0 + + env.register_apps([system_app, aui_app, payment_app]) + env._open_app("payment") + return env + + + + +def test_starts_in_home(payment_app: StatefulPaymentApp) -> None: + """App should start in HomeView with empty navigation stack.""" + assert isinstance(payment_app.current_state, HomeView) + assert payment_app.navigation_stack == [] + + + + +def test_view_transactions_transition(payment_app: StatefulPaymentApp) -> None: + """Handler: view_transactions event transitions to TransactionListView.""" + event = _make_event(payment_app, payment_app.current_state.view_transactions) + payment_app.handle_state_transition(event) + + assert isinstance(payment_app.current_state, TransactionListView) + assert len(payment_app.navigation_stack) == 1 + + +def test_view_contacts_transition(payment_app: StatefulPaymentApp) -> None: + """Handler: view_contacts event transitions to ContactListView.""" + event = _make_event(payment_app, payment_app.current_state.view_contacts) + payment_app.handle_state_transition(event) + + assert isinstance(payment_app.current_state, ContactListView) + assert len(payment_app.navigation_stack) == 1 + + +def test_view_payment_methods_transition(payment_app: StatefulPaymentApp) -> None: + """Handler: view_payment_methods event transitions to PaymentMethodsView.""" + event = _make_event(payment_app, payment_app.current_state.view_payment_methods) + payment_app.handle_state_transition(event) + + assert isinstance(payment_app.current_state, PaymentMethodsView) + assert len(payment_app.navigation_stack) == 1 + + +def test_send_payment_transition(payment_app: StatefulPaymentApp) -> None: + """Handler: send_payment event transitions to TransactionDetail.""" + recipient_id = [uid for uid in payment_app.users.keys() if uid != payment_app.user_id][0] + + payment_state = PaymentView() + payment_app.set_current_state(payment_state) + + transaction_id = "test_txn_123" + event = _make_event( + payment_app, + payment_app.current_state.send_payment, + result=transaction_id, + recipient_id=recipient_id, + amount=10.0, + note="Test payment", + ) + payment_app.handle_state_transition(event) + + assert isinstance(payment_app.current_state, TransactionDetail) + assert payment_app.current_state.transaction_id == transaction_id + + +def test_request_payment_transition(payment_app: StatefulPaymentApp) -> None: + """Handler: request_payment event transitions to TransactionDetail.""" + recipient_id = [uid for uid in payment_app.users.keys() if uid != payment_app.user_id][0] + + payment_state = PaymentView() + payment_app.set_current_state(payment_state) + + transaction_id = "test_request_123" + event = _make_event( + payment_app, + payment_app.current_state.request_payment, + result=transaction_id, + recipient_id=recipient_id, + amount=20.0, + note="Test request", + ) + payment_app.handle_state_transition(event) + + assert isinstance(payment_app.current_state, TransactionDetail) + assert payment_app.current_state.transaction_id == transaction_id + + +def test_open_transaction_transition(payment_app: StatefulPaymentApp) -> None: + """Handler: open_transaction event transitions to TransactionDetail.""" + transaction_list_state = TransactionListView() + payment_app.set_current_state(transaction_list_state) + + transaction_id = list(payment_app.transactions.keys())[0] + event = _make_event( + payment_app, + payment_app.current_state.open_transaction, + transaction_id=transaction_id, + ) + payment_app.handle_state_transition(event) + + assert isinstance(payment_app.current_state, TransactionDetail) + assert payment_app.current_state.transaction_id == transaction_id + + +def test_decline_request_transition(payment_app: StatefulPaymentApp) -> None: + """Handler: decline_request event transitions to TransactionListView.""" + pending_requests = [ + t for t in payment_app.transactions.values() + if t.transaction_type == "request" and t.status == "pending" + ] + if not pending_requests: + pytest.skip("No pending requests in test data") + + request = pending_requests[0] + detail_state = TransactionDetail(transaction_id=request.transaction_id) + payment_app.set_current_state(detail_state) + + event = _make_event( + payment_app, + payment_app.current_state.decline_request, + result=f"Request {request.transaction_id} declined", + request_id=request.transaction_id, + ) + payment_app.handle_state_transition(event) + + assert isinstance(payment_app.current_state, TransactionListView) + + +def test_open_contact_transition(payment_app: StatefulPaymentApp) -> None: + """Handler: open_contact event transitions to UserProfile.""" + contact_list_state = ContactListView() + payment_app.set_current_state(contact_list_state) + + friend_id = [ + uid for uid, user in payment_app.users.items() + if user.is_friend and uid != payment_app.user_id + ][0] + + event = _make_event( + payment_app, + payment_app.current_state.open_contact, + user_id=friend_id, + ) + payment_app.handle_state_transition(event) + + assert isinstance(payment_app.current_state, UserProfile) + assert payment_app.current_state.user_id == friend_id + + +def test_pay_user_transition(payment_app: StatefulPaymentApp) -> None: + """Handler: pay_user event transitions to PaymentView with recipient_id.""" + other_user_id = [uid for uid in payment_app.users.keys() if uid != payment_app.user_id][0] + profile_state = UserProfile(user_id=other_user_id) + payment_app.set_current_state(profile_state) + + event = _make_event( + payment_app, + payment_app.current_state.pay_user, + user_id=other_user_id, + ) + payment_app.handle_state_transition(event) + + assert isinstance(payment_app.current_state, PaymentView) + assert payment_app.current_state.recipient_id == other_user_id + + +def test_view_transfer_transition(payment_app: StatefulPaymentApp) -> None: + """Handler: view_transfer event transitions to TransferView.""" + payment_methods_state = PaymentMethodsView() + payment_app.set_current_state(payment_methods_state) + + event = _make_event( + payment_app, + payment_app.current_state.view_transfer, + ) + payment_app.handle_state_transition(event) + + assert isinstance(payment_app.current_state, TransferView) + + + + +def test_send_payment_flow(env_with_payment: StateAwareEnvironmentWrapper) -> None: + """Integration: Home -> PaymentView -> send_payment -> TransactionDetail.""" + env = env_with_payment + app = env.get_app_with_class(StatefulPaymentApp) + + recipient_id = app.create_user( + username="recipient", + display_name="Recipient User", + phone="555-9999", + ) + + assert isinstance(app.current_state, HomeView) + + app.navigate_to_payment(recipient_id) + assert isinstance(app.current_state, PaymentView) + assert app.current_state.recipient_id == recipient_id + assert len(app.navigation_stack) == 1 + + initial_balance = app.balance + _payment_state(app).send_payment( + recipient_id=recipient_id, + amount=10.0, + note="Test payment", + ) + + assert isinstance(app.current_state, TransactionDetail) + assert len(app.navigation_stack) == 2 + assert app.balance == initial_balance - 10.0 + + +def test_go_back_navigation(env_with_payment: StateAwareEnvironmentWrapper) -> None: + """Integration: Test go_back through multiple states.""" + env = env_with_payment + app = env.get_app_with_class(StatefulPaymentApp) + + _home_state(app).view_transactions() + assert isinstance(app.current_state, TransactionListView) + assert len(app.navigation_stack) == 1 + + app.go_back() + assert isinstance(app.current_state, HomeView) + assert len(app.navigation_stack) == 0 + + _home_state(app).view_payment_methods() + _payment_methods_state(app).view_transfer() + assert isinstance(app.current_state, TransferView) + assert len(app.navigation_stack) == 2 + + app.go_back() + assert isinstance(app.current_state, PaymentMethodsView) + assert len(app.navigation_stack) == 1 + + app.go_back() + assert isinstance(app.current_state, HomeView) + assert len(app.navigation_stack) == 0 + + + + + +def test_send_payment(payment_app: StatefulPaymentApp) -> None: + """Test sending a payment.""" + recipient_id = [uid for uid in payment_app.users.keys() if uid != payment_app.user_id][0] + initial_balance = payment_app.balance + + transaction_id = payment_app.send_payment( + recipient_id=recipient_id, + amount=15.0, + note="Test payment", + ) + + assert transaction_id in payment_app.transactions + assert payment_app.balance == initial_balance - 15.0 + assert payment_app.transactions[transaction_id].status == "completed" + + +def test_request_payment(payment_app: StatefulPaymentApp) -> None: + """Test requesting a payment.""" + recipient_id = [uid for uid in payment_app.users.keys() if uid != payment_app.user_id][0] + + request_id = payment_app.request_payment( + recipient_id=recipient_id, + amount=30.0, + note="Test request", + ) + + assert request_id in payment_app.transactions + assert payment_app.transactions[request_id].transaction_type == "request" + assert payment_app.transactions[request_id].status == "pending" + + +def test_pay_request(payment_app: StatefulPaymentApp) -> None: + """Test paying a pending request.""" + pending_requests = [ + t for t in payment_app.transactions.values() + if t.transaction_type == "request" and t.status == "pending" + and t.recipient_id == payment_app.user_id + ] + if not pending_requests: + sender_id = [uid for uid in payment_app.users.keys() if uid != payment_app.user_id][0] + request_id = payment_app.create_transaction_with_time( + transaction_type="request", + sender_id=sender_id, + recipient_id=payment_app.user_id, + amount=20.0, + note="Test request", + status="pending", + created_at="2025-01-17 10:00:00", + ) + pending_requests = [payment_app.transactions[request_id]] + + request = pending_requests[0] + initial_balance = payment_app.balance + + payment_app.pay_request(request.transaction_id) + + assert payment_app.transactions[request.transaction_id].status == "completed" + assert payment_app.balance == initial_balance - request.amount + + + + +def test_send_payment_insufficient_balance(payment_app: StatefulPaymentApp) -> None: + """Test sending payment with insufficient balance (should raise error).""" + recipient_id = [uid for uid in payment_app.users.keys() if uid != payment_app.user_id][0] + payment_app.balance = 5.0 + + with pytest.raises(ValueError, match="Insufficient balance"): + payment_app.send_payment( + recipient_id=recipient_id, + amount=100.0, + note="Too much", + ) + + +def test_invalid_user_id_raises_error(payment_app: StatefulPaymentApp) -> None: + """Test that operations with invalid user IDs raise errors.""" + with pytest.raises(KeyError): + payment_app.set_current_user("nonexistent_user") + + with pytest.raises(KeyError): + payment_app.send_payment( + recipient_id="nonexistent_user", + amount=10.0, + note="Test", + ) + + +def test_pay_request_wrong_recipient(payment_app: StatefulPaymentApp) -> None: + """Test that paying a request not directed to current user raises error.""" + recipient_id = [uid for uid in payment_app.users.keys() if uid != payment_app.user_id][0] + request_id = payment_app.request_payment( + recipient_id=recipient_id, + amount=10.0, + note="Test", + ) + + with pytest.raises(ValueError, match="Cannot pay request not directed to you"): + payment_app.pay_request(request_id)