# /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: - simulate_redirect (NEW default): Returns a checkout URL for paid tiers. The frontend redirects to the mock checkout page; after user clicks "Pay Now", a webhook callback marks the PaymentIntent as COMPLETED. - auto_approve: All payments succeed immediately (legacy). - simulate_failure: All payments fail with a configurable error message. - simulate_timeout: Payments hang until a configurable timeout then succeed. Usage: gateway = MockPaymentGateway(mode="simulate_redirect") result = await gateway.create_intent(Decimal("100.00"), "EUR") """ def __init__( self, mode: str = "simulate_redirect", failure_message: str = "Mock payment declined (simulated failure)", timeout_seconds: int = 5, base_url: str = "http://localhost:8000", ): """ Initialize the mock gateway. Args: mode: Operation mode — "simulate_redirect" (default), "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. base_url: Base URL for generating mock checkout URLs. """ self.mode = mode self.failure_message = failure_message self.timeout_seconds = timeout_seconds self.base_url = base_url self._processed_intents: Dict[str, Dict[str, Any]] = {} logger.info( "MockPaymentGateway initialized: mode=%s, timeout=%ds, base_url=%s", self.mode, self.timeout_seconds, self.base_url, ) # ────────────────────────────────────────────────────────────────────────── # 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. Modes: - simulate_redirect (default): For paid amounts (>0), returns a checkout URL with status "requires_action". The frontend redirects the user to this URL. For $0 amounts, falls through to auto_approve. - auto_approve: Immediately returns a "completed" intent. - simulate_failure: Raises PaymentGatewayError. - simulate_timeout: Logs a warning and returns "processing". THOUGHT PROCESS: The simulate_redirect mode bridges the gap between "instant success" and a realistic payment flow. The returned checkout_url points to a mock page served by our own backend. When the user clicks "Pay Now", the mock page POSTs to a webhook callback endpoint that finalizes the PaymentIntent. This allows frontend testing of the full redirect → callback → success lifecycle without Stripe. Args: amount: The payment amount. currency: ISO 4217 currency code (default: EUR). metadata: Optional metadata dict. **kwargs: Supports 'base_url' override for the checkout URL base. Returns: Dict with mock payment intent details: - simulate_redirect: {id, status: "requires_action", checkout_url, ...} - auto_approve: {id, status: "completed", ...} Raises: PaymentGatewayError: In simulate_failure mode. """ intent_id = f"mock_intent_{uuid.uuid4().hex[:12]}" # ── Strict console log for audit trail ────────────────────────────── logger.info( "[MOCK_PAYMENT_REQUEST] Initiating transaction with external gateway " "for amount: %s %s, mode=%s, intent_id=%s", amount, currency, self.mode, intent_id, ) # ── simulate_failure: Always fail ─────────────────────────────────── if self.mode == "simulate_failure": logger.warning( "Mock payment FAILURE: intent=%s reason='%s'", intent_id, self.failure_message, ) raise PaymentGatewayError(self.failure_message) # ── simulate_timeout: Pretend to hang then return "processing" ────── 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. # ── simulate_redirect: Return a checkout URL for paid tiers ───────── # For $0 amounts, fall through to auto_approve behaviour. if self.mode == "simulate_redirect" and amount > 0: # Use the base_url from kwargs if provided, otherwise use self.base_url base_url = kwargs.get("base_url", self.base_url) # Generate the mock checkout page URL — the user will be redirected # here to complete the payment. After clicking "Pay Now", the mock # page POSTs to /billing/mock-payment/callback which updates the # PaymentIntent from PENDING → COMPLETED. checkout_url = f"{base_url}/api/v1/billing/mock-payment/checkout/{intent_id}" intent_data = { "id": intent_id, "status": "requires_action", "amount": float(amount), "currency": currency, "checkout_url": checkout_url, "method": "GET", # Frontend opens this URL in a new tab/window "created_at": datetime.utcnow().isoformat(), "completed_at": None, "metadata": metadata or {}, "gateway": "mock", } self._processed_intents[intent_id] = intent_data logger.info( "[MOCK_PAYMENT_REDIRECT] Checkout URL generated: intent=%s " "checkout_url=%s amount=%s %s", intent_id, checkout_url, amount, currency, ) return intent_data # ── auto_approve OR $0 amount: Immediate completion ───────────────── now = datetime.utcnow() intent_data = { "id": intent_id, "status": "completed", "amount": float(amount), "currency": currency, "created_at": now.isoformat(), "completed_at": now.isoformat(), "metadata": metadata or {}, "gateway": "mock", } self._processed_intents[intent_id] = intent_data logger.info( "Mock payment AUTO-APPROVED: intent=%s amount=%s %s", intent_id, amount, currency, ) 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: "simulate_redirect", "auto_approve", "simulate_failure", or "simulate_timeout". """ valid_modes = {"simulate_redirect", "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")