245 lines
8.7 KiB
Python
245 lines
8.7 KiB
Python
# /opt/docker/dev/service_finder/backend/app/services/mock_payment_gateway.py
|
|
"""
|
|
Mock Payment Gateway — Development/Testing implementation of BasePaymentGateway.
|
|
|
|
Simulates successful payment processing without requiring real Stripe keys.
|
|
Can be configured to simulate failures for testing error handling paths.
|
|
|
|
THOUGHT PROCESS:
|
|
- Implements BasePaymentGateway abstract interface (Strategy Pattern).
|
|
- Default mode is "auto_approve" — all payments succeed immediately.
|
|
- Supports "simulate_failure" mode for testing error paths.
|
|
- Generates deterministic mock session IDs for traceability.
|
|
- Logs all payment attempts for debugging.
|
|
- Decoupled from FinancialManager via dependency injection.
|
|
"""
|
|
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime, timedelta
|
|
from decimal import Decimal
|
|
from typing import Optional, Dict, Any
|
|
|
|
from app.services.financial_interfaces import (
|
|
BasePaymentGateway,
|
|
PaymentGatewayError,
|
|
)
|
|
|
|
logger = logging.getLogger("mock-payment-gateway")
|
|
|
|
|
|
class MockPaymentGateway(BasePaymentGateway):
|
|
"""
|
|
Mock payment gateway for development and testing.
|
|
|
|
Modes:
|
|
- auto_approve (default): All payments succeed immediately.
|
|
- simulate_failure: All payments fail with a configurable error message.
|
|
- simulate_timeout: Payments hang until a configurable timeout then succeed.
|
|
|
|
Usage:
|
|
gateway = MockPaymentGateway(mode="auto_approve")
|
|
result = await gateway.create_intent(Decimal("100.00"), "EUR")
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
mode: str = "auto_approve",
|
|
failure_message: str = "Mock payment declined (simulated failure)",
|
|
timeout_seconds: int = 5,
|
|
):
|
|
"""
|
|
Initialize the mock gateway.
|
|
|
|
Args:
|
|
mode: Operation mode — "auto_approve", "simulate_failure", or "simulate_timeout".
|
|
failure_message: Custom error message for simulate_failure mode.
|
|
timeout_seconds: Simulated processing delay for simulate_timeout mode.
|
|
"""
|
|
self.mode = mode
|
|
self.failure_message = failure_message
|
|
self.timeout_seconds = timeout_seconds
|
|
self._processed_intents: Dict[str, Dict[str, Any]] = {}
|
|
|
|
logger.info(
|
|
"MockPaymentGateway initialized: mode=%s, timeout=%ds",
|
|
self.mode, self.timeout_seconds,
|
|
)
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────
|
|
# BasePaymentGateway Implementation
|
|
# ──────────────────────────────────────────────────────────────────────────
|
|
|
|
async def create_intent(
|
|
self,
|
|
amount: Decimal,
|
|
currency: str = "EUR",
|
|
metadata: Optional[Dict[str, Any]] = None,
|
|
**kwargs,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Create a mock payment intent.
|
|
|
|
In auto_approve mode, immediately returns a "completed" intent.
|
|
In simulate_failure mode, raises PaymentGatewayError.
|
|
In simulate_timeout mode, logs a warning and returns "processing".
|
|
|
|
Args:
|
|
amount: The payment amount.
|
|
currency: ISO 4217 currency code (default: EUR).
|
|
metadata: Optional metadata dict.
|
|
**kwargs: Additional parameters (ignored in mock).
|
|
|
|
Returns:
|
|
Dict with mock payment intent details.
|
|
|
|
Raises:
|
|
PaymentGatewayError: In simulate_failure mode.
|
|
"""
|
|
intent_id = f"mock_intent_{uuid.uuid4().hex[:12]}"
|
|
logger.info(
|
|
"Mock create_intent: id=%s amount=%s %s mode=%s",
|
|
intent_id, amount, currency, self.mode,
|
|
)
|
|
|
|
if self.mode == "simulate_failure":
|
|
logger.warning(
|
|
"Mock payment FAILURE: intent=%s reason='%s'",
|
|
intent_id, self.failure_message,
|
|
)
|
|
raise PaymentGatewayError(self.failure_message)
|
|
|
|
if self.mode == "simulate_timeout":
|
|
logger.warning(
|
|
"Mock payment TIMEOUT simulation: intent=%s delay=%ds",
|
|
intent_id, self.timeout_seconds,
|
|
)
|
|
# In a real scenario we'd sleep; here we just return "processing"
|
|
# so the caller can handle the pending state.
|
|
|
|
now = datetime.utcnow()
|
|
intent_data = {
|
|
"id": intent_id,
|
|
"status": "completed" if self.mode == "auto_approve" else "processing",
|
|
"amount": float(amount),
|
|
"currency": currency,
|
|
"created_at": now.isoformat(),
|
|
"completed_at": now.isoformat() if self.mode == "auto_approve" else None,
|
|
"metadata": metadata or {},
|
|
"gateway": "mock",
|
|
}
|
|
|
|
self._processed_intents[intent_id] = intent_data
|
|
return intent_data
|
|
|
|
async def verify_payment(
|
|
self,
|
|
payment_intent_id: str,
|
|
**kwargs,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Verify a mock payment's status.
|
|
|
|
Args:
|
|
payment_intent_id: The mock intent ID to verify.
|
|
**kwargs: Additional parameters (ignored in mock).
|
|
|
|
Returns:
|
|
Dict with verification details.
|
|
|
|
Raises:
|
|
PaymentGatewayError: If the intent ID is unknown.
|
|
"""
|
|
intent = self._processed_intents.get(payment_intent_id)
|
|
if not intent:
|
|
logger.warning(
|
|
"Mock verify_payment: unknown intent_id=%s",
|
|
payment_intent_id,
|
|
)
|
|
raise PaymentGatewayError(
|
|
f"Mock payment intent '{payment_intent_id}' not found"
|
|
)
|
|
|
|
logger.info(
|
|
"Mock verify_payment: intent=%s status=%s",
|
|
payment_intent_id, intent["status"],
|
|
)
|
|
|
|
return {
|
|
"verified": intent["status"] == "completed",
|
|
"intent_id": payment_intent_id,
|
|
"status": intent["status"],
|
|
"amount": intent["amount"],
|
|
"currency": intent["currency"],
|
|
}
|
|
|
|
async def refund_payment(
|
|
self,
|
|
payment_intent_id: str,
|
|
amount: Optional[Decimal] = None,
|
|
**kwargs,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Refund a mock payment.
|
|
|
|
Args:
|
|
payment_intent_id: The mock intent ID to refund.
|
|
amount: Optional partial refund amount (None = full refund).
|
|
**kwargs: Additional parameters (ignored in mock).
|
|
|
|
Returns:
|
|
Dict with refund details.
|
|
|
|
Raises:
|
|
PaymentGatewayError: If the intent ID is unknown.
|
|
"""
|
|
intent = self._processed_intents.get(payment_intent_id)
|
|
if not intent:
|
|
logger.warning(
|
|
"Mock refund_payment: unknown intent_id=%s",
|
|
payment_intent_id,
|
|
)
|
|
raise PaymentGatewayError(
|
|
f"Mock payment intent '{payment_intent_id}' not found"
|
|
)
|
|
|
|
refund_amount = float(amount) if amount is not None else intent["amount"]
|
|
refund_id = f"mock_refund_{uuid.uuid4().hex[:12]}"
|
|
|
|
logger.info(
|
|
"Mock refund: intent=%s amount=%s refund_id=%s",
|
|
payment_intent_id, refund_amount, refund_id,
|
|
)
|
|
|
|
return {
|
|
"refunded": True,
|
|
"refund_id": refund_id,
|
|
"intent_id": payment_intent_id,
|
|
"amount": refund_amount,
|
|
"status": "succeeded",
|
|
}
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────
|
|
# Mock-specific Helpers
|
|
# ──────────────────────────────────────────────────────────────────────────
|
|
|
|
def set_mode(self, mode: str) -> None:
|
|
"""
|
|
Change the gateway's operating mode at runtime.
|
|
|
|
Args:
|
|
mode: "auto_approve", "simulate_failure", or "simulate_timeout".
|
|
"""
|
|
valid_modes = {"auto_approve", "simulate_failure", "simulate_timeout"}
|
|
if mode not in valid_modes:
|
|
raise ValueError(
|
|
f"Invalid mock mode '{mode}'. Valid modes: {valid_modes}"
|
|
)
|
|
self.mode = mode
|
|
logger.info("MockPaymentGateway mode changed to: %s", self.mode)
|
|
|
|
def reset(self) -> None:
|
|
"""Clear all processed intents (useful between tests)."""
|
|
self._processed_intents.clear()
|
|
logger.info("MockPaymentGateway reset: all intents cleared")
|