e41ecfad4c
Include Loyalty/Communication/Sports Center backends and registry updates alongside production nginx and compose wiring. Co-authored-by: Cursor <cursoragent@cursor.com>
69 lines
2.4 KiB
Python
69 lines
2.4 KiB
Python
"""Mock SMS provider for tests and local development."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from app.providers.contracts import (
|
|
BalanceResult,
|
|
ProviderHealth,
|
|
SendRequest,
|
|
SendResult,
|
|
)
|
|
|
|
|
|
class MockSMSProvider:
|
|
provider_kind = "mock"
|
|
channel = "sms"
|
|
|
|
def __init__(self) -> None:
|
|
self.sent: list[SendRequest] = []
|
|
self.fail_next: bool = False
|
|
self.fail_message: str = "mock_failure"
|
|
|
|
async def send(self, request: SendRequest, credentials: dict[str, Any]) -> SendResult:
|
|
self.sent.append(request)
|
|
if self.fail_next or credentials.get("force_fail"):
|
|
self.fail_next = False
|
|
return SendResult(
|
|
success=False,
|
|
error_code="mock_failure",
|
|
error_message=self.fail_message,
|
|
latency_ms=1,
|
|
)
|
|
return SendResult(
|
|
success=True,
|
|
provider_message_id=f"mock-{uuid4()}",
|
|
raw_response={"status": "ok"},
|
|
latency_ms=1,
|
|
)
|
|
|
|
async def get_balance(self, credentials: dict[str, Any]) -> BalanceResult:
|
|
return BalanceResult(balance=float(credentials.get("balance", 1000)), currency="IRR")
|
|
|
|
async def health_check(self, credentials: dict[str, Any]) -> ProviderHealth:
|
|
if credentials.get("unhealthy"):
|
|
return ProviderHealth(healthy=False, detail="forced unhealthy")
|
|
return ProviderHealth(healthy=True, detail="ok")
|
|
|
|
|
|
class StubFutureChannelProvider:
|
|
"""Placeholder for email/push/whatsapp/telegram/rubika/voice until implemented."""
|
|
|
|
def __init__(self, provider_kind: str, channel: str) -> None:
|
|
self.provider_kind = provider_kind
|
|
self.channel = channel
|
|
|
|
async def send(self, request: SendRequest, credentials: dict[str, Any]) -> SendResult:
|
|
return SendResult(
|
|
success=False,
|
|
error_code="channel_not_implemented",
|
|
error_message=f"Channel {self.channel} is registered but not yet implemented",
|
|
)
|
|
|
|
async def get_balance(self, credentials: dict[str, Any]) -> BalanceResult:
|
|
return BalanceResult(balance=None)
|
|
|
|
async def health_check(self, credentials: dict[str, Any]) -> ProviderHealth:
|
|
return ProviderHealth(healthy=False, detail="not implemented")
|