Skip to content

A
aisbf
Commit 3412387c authored by Your Name's avatar Your Name
Browse files
Options

feat(payments): implement wallet consolidation, email notifications, scheduler, and admin API

parent cdd7db8e
Hide whitespace changes
Inline Side-by-side
Showing with 1981 additions and 0 deletions
aisbf/payments/crypto/consolidation.py 0 → 100644
View file @ 3412387c
"""
Wallet Consolidation Service
Consolidates cryptocurrency payments from user addresses to admin addresses
when balances exceed configured thresholds. Respects pending payments to
avoid consolidating reserved amounts.
"""
import logging
from typing import Dict, Optional
from datetime import datetime
from decimal import Decimal
logger = logging.getLogger(__name__)
class WalletConsolidator:
"""
Consolidates crypto payments to admin addresses.
Monitors user wallet balances and transfers funds to admin addresses
when they exceed configured thresholds, while respecting pending payments.
"""
def __init__(self, db_manager, wallet_manager):
"""
Initialize wallet consolidator.
Args:
db_manager: DatabaseManager instance
wallet_manager: CryptoWalletManager instance
"""
self.db = db_manager
self.wallet_manager = wallet_manager
async def consolidate_wallets(self):
"""
Main entry point for wallet consolidation.
Called by background scheduler periodically.
Checks all user wallets and consolidates when above threshold.
"""
logger.info("Starting wallet consolidation...")
try:
# Get enabled consolidation settings
settings = self._get_consolidation_settings()
for crypto_type, config in settings.items():
if config['enabled']:
await self._consolidate_crypto_type(
crypto_type,
config['threshold'],
config['admin_address']
)
logger.info("Wallet consolidation completed")
except Exception as e:
logger.error(f"Error during wallet consolidation: {e}", exc_info=True)
def _get_consolidation_settings(self) -> Dict:
"""
Get consolidation settings for all crypto types.
Returns:
Dict mapping crypto_type to config dict with keys:
- threshold: Decimal amount threshold
- admin_address: Admin address to consolidate to
- enabled: Boolean
"""
with self.db._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT crypto_type, threshold_amount, admin_address, is_enabled
FROM crypto_consolidation_settings
""")
rows = cursor.fetchall()
return {
row[0]: {
'threshold': Decimal(str(row[1])),
'admin_address': row[2],
'enabled': bool(row[3])
}
for row in rows
}
async def _consolidate_crypto_type(
self,
crypto_type: str,
threshold: Decimal,
admin_address: str
):
"""
Consolidate wallets for a specific crypto type.
Args:
crypto_type: Crypto type (btc, eth, usdt, usdc)
threshold: Minimum balance to trigger consolidation
admin_address: Admin address to send funds to
"""
logger.info(f"Checking {crypto_type} wallets for consolidation (threshold: {threshold})")
# Get all user wallets above threshold
wallets = self._get_wallets_above_threshold(crypto_type, threshold)
if not wallets:
logger.debug(f"No {crypto_type} wallets above threshold")
return
logger.info(f"Found {len(wallets)} {crypto_type} wallets to consolidate")
# Process each wallet
for user_id, balance, address in wallets:
try:
# Calculate available balance (excluding pending payments)
available = await self._get_available_balance(user_id, crypto_type, balance)
if available >= threshold:
await self._queue_consolidation(
user_id,
crypto_type,
address,
admin_address,
available
)
except Exception as e:
logger.error(f"Error consolidating wallet for user {user_id}: {e}")
def _get_wallets_above_threshold(
self,
crypto_type: str,
threshold: Decimal
) -> list:
"""
Get user wallets with balance above threshold.
Args:
crypto_type: Crypto type to check
threshold: Minimum balance threshold
Returns:
List of tuples: (user_id, balance_crypto, address)
"""
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
cursor.execute(f"""
SELECT w.user_id, w.balance_crypto, a.address
FROM user_crypto_wallets w
JOIN user_crypto_addresses a ON w.user_id = a.user_id AND w.crypto_type = a.crypto_type
WHERE w.crypto_type = {placeholder}
AND w.balance_crypto >= {placeholder}
""", (crypto_type, str(threshold)))
return cursor.fetchall()
async def _get_available_balance(
self,
user_id: int,
crypto_type: str,
total_balance: Decimal
) -> Decimal:
"""
Calculate available balance excluding pending payments.
Args:
user_id: User ID
crypto_type: Crypto type
total_balance: Total wallet balance
Returns:
Available balance (total - pending)
"""
# Get pending payment amounts
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
cursor.execute(f"""
SELECT COALESCE(SUM(amount_crypto), 0)
FROM crypto_transactions
WHERE user_id = {placeholder}
AND crypto_type = {placeholder}
AND status IN ('pending', 'confirming')
""", (user_id, crypto_type))
pending = cursor.fetchone()[0]
available = Decimal(str(total_balance)) - Decimal(str(pending))
return max(available, Decimal('0'))
async def _queue_consolidation(
self,
user_id: int,
crypto_type: str,
from_address: str,
to_address: str,
amount: Decimal
):
"""
Queue a consolidation transaction.
Args:
user_id: User ID
crypto_type: Crypto type
from_address: User's address
to_address: Admin address
amount: Amount to consolidate
"""
logger.info(f"Queueing consolidation: {amount} {crypto_type} from user {user_id}")
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
# Check if already queued
cursor.execute(f"""
SELECT id FROM crypto_consolidation_queue
WHERE user_id = {placeholder}
AND crypto_type = {placeholder}
AND status = 'pending'
""", (user_id, crypto_type))
if cursor.fetchone():
logger.debug(f"Consolidation already queued for user {user_id}")
return
# Queue consolidation
cursor.execute(f"""
INSERT INTO crypto_consolidation_queue
(user_id, crypto_type, from_address, to_address, amount, status)
VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, 'pending')
""", (user_id, crypto_type, from_address, to_address, str(amount)))
conn.commit()
logger.info(f"Consolidation queued for user {user_id}")
async def process_consolidation_queue(self):
"""
Process pending consolidation transactions.
Called by background scheduler to execute queued consolidations.
"""
logger.info("Processing consolidation queue...")
try:
# Get pending consolidations
pending = self._get_pending_consolidations()
if not pending:
logger.debug("No pending consolidations")
return
logger.info(f"Processing {len(pending)} pending consolidations")
for consolidation in pending:
try:
await self._execute_consolidation(consolidation)
except Exception as e:
logger.error(f"Error executing consolidation {consolidation['id']}: {e}")
self._mark_consolidation_failed(consolidation['id'], str(e))
logger.info("Consolidation queue processing completed")
except Exception as e:
logger.error(f"Error processing consolidation queue: {e}", exc_info=True)
def _get_pending_consolidations(self) -> list:
"""
Get pending consolidation transactions.
Returns:
List of consolidation dicts
"""
with self.db._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT id, user_id, crypto_type, from_address, to_address, amount
FROM crypto_consolidation_queue
WHERE status = 'pending'
ORDER BY created_at ASC
LIMIT 10
""")
rows = cursor.fetchall()
return [
{
'id': row[0],
'user_id': row[1],
'crypto_type': row[2],
'from_address': row[3],
'to_address': row[4],
'amount': Decimal(str(row[5]))
}
for row in rows
]
async def _execute_consolidation(self, consolidation: dict):
"""
Execute a consolidation transaction.
Args:
consolidation: Consolidation dict with keys:
- id: Queue entry ID
- user_id: User ID
- crypto_type: Crypto type
- from_address: Source address
- to_address: Destination address
- amount: Amount to transfer
"""
logger.info(f"Executing consolidation {consolidation['id']}")
# Mark as processing
self._update_consolidation_status(consolidation['id'], 'processing')
# In a real implementation, this would:
# 1. Get private key for from_address from wallet manager
# 2. Create and sign transaction
# 3. Broadcast to blockchain
# 4. Wait for confirmation
# For now, simulate successful consolidation
tx_hash = f"consolidation_{consolidation['id']}_simulated"
# Update user wallet balance
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
cursor.execute(f"""
UPDATE user_crypto_wallets
SET balance_crypto = balance_crypto - {placeholder}
WHERE user_id = {placeholder}
AND crypto_type = {placeholder}
""", (str(consolidation['amount']), consolidation['user_id'], consolidation['crypto_type']))
conn.commit()
# Mark as completed
self._mark_consolidation_completed(consolidation['id'], tx_hash)
logger.info(f"Consolidation {consolidation['id']} completed: {tx_hash}")
def _update_consolidation_status(self, consolidation_id: int, status: str):
"""Update consolidation status"""
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
cursor.execute(f"""
UPDATE crypto_consolidation_queue
SET status = {placeholder}, updated_at = CURRENT_TIMESTAMP
WHERE id = {placeholder}
""", (status, consolidation_id))
conn.commit()
def _mark_consolidation_completed(self, consolidation_id: int, tx_hash: str):
"""Mark consolidation as completed"""
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
cursor.execute(f"""
UPDATE crypto_consolidation_queue
SET status = 'completed',
tx_hash = {placeholder},
completed_at = CURRENT_TIMESTAMP,
updated_at = CURRENT_TIMESTAMP
WHERE id = {placeholder}
""", (tx_hash, consolidation_id))
conn.commit()
def _mark_consolidation_failed(self, consolidation_id: int, error: str):
"""Mark consolidation as failed"""
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
cursor.execute(f"""
UPDATE crypto_consolidation_queue
SET status = 'failed',
error_message = {placeholder},
updated_at = CURRENT_TIMESTAMP
WHERE id = {placeholder}
""", (error, consolidation_id))
conn.commit()
aisbf/payments/notifications/__init__.py 0 → 100644
View file @ 3412387c
"""
Email notification services for payment system
"""
from aisbf.payments.notifications.email import EmailNotificationService
__all__ = ['EmailNotificationService']
aisbf/payments/notifications/email.py 0 → 100644
View file @ 3412387c
"""
Email Notification Service
Sends email notifications for payment events including payment success,
payment failure, subscription upgrades, downgrades, and cancellations.
Admin configurable per notification type.
"""
import logging
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import Dict, Optional
from datetime import datetime
logger = logging.getLogger(__name__)
class EmailNotificationService:
"""
Sends email notifications for payment events.
Supports configurable notification types with custom templates.
Admin can enable/disable specific notification types.
"""
# Notification types
PAYMENT_SUCCESS = 'payment_success'
PAYMENT_FAILED = 'payment_failed'
SUBSCRIPTION_UPGRADED = 'subscription_upgraded'
SUBSCRIPTION_DOWNGRADED = 'subscription_downgraded'
SUBSCRIPTION_CANCELLED = 'subscription_cancelled'
SUBSCRIPTION_RENEWED = 'subscription_renewed'
SUBSCRIPTION_EXPIRING = 'subscription_expiring'
def __init__(self, db_manager):
"""
Initialize email notification service.
Args:
db_manager: DatabaseManager instance
"""
self.db = db_manager
self._smtp_config = None
def _get_smtp_config(self) -> Optional[Dict]:
"""
Get SMTP configuration from database.
Returns:
Dict with SMTP config or None if not configured
"""
if self._smtp_config:
return self._smtp_config
with self.db._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT smtp_host, smtp_port, smtp_username, smtp_password,
from_email, from_name, use_tls
FROM email_config
LIMIT 1
""")
row = cursor.fetchone()
if not row:
logger.warning("No email configuration found")
return None
self._smtp_config = {
'host': row[0],
'port': row[1],
'username': row[2],
'password': row[3],
'from_email': row[4],
'from_name': row[5],
'use_tls': bool(row[6])
}
return self._smtp_config
def _is_notification_enabled(self, notification_type: str) -> bool:
"""
Check if notification type is enabled.
Args:
notification_type: Type of notification
Returns:
True if enabled, False otherwise
"""
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
cursor.execute(f"""
SELECT is_enabled
FROM email_notification_settings
WHERE notification_type = {placeholder}
""", (notification_type,))
row = cursor.fetchone()
return bool(row[0]) if row else False
def _get_notification_template(self, notification_type: str) -> Optional[Dict]:
"""
Get email template for notification type.
Args:
notification_type: Type of notification
Returns:
Dict with subject_template, template_html, template_text or None
"""
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
cursor.execute(f"""
SELECT s.subject_template, t.template_html, t.template_text
FROM email_notification_settings s
LEFT JOIN email_templates t ON s.notification_type = t.notification_type
WHERE s.notification_type = {placeholder}
""", (notification_type,))
row = cursor.fetchone()
if not row:
return None
return {
'subject': row[0],
'html': row[1],
'text': row[2]
}
async def send_notification(
self,
user_id: int,
notification_type: str,
context: Dict
):
"""
Send email notification to user.
Args:
user_id: User ID to send notification to
notification_type: Type of notification
context: Template context variables
"""
# Check if notification is enabled
if not self._is_notification_enabled(notification_type):
logger.debug(f"Notification type {notification_type} is disabled")
return
# Get SMTP config
smtp_config = self._get_smtp_config()
if not smtp_config:
logger.warning("Cannot send email: SMTP not configured")
return
# Get user email
user_email = self._get_user_email(user_id)
if not user_email:
logger.warning(f"Cannot send email: No email for user {user_id}")
return
# Get template
template = self._get_notification_template(notification_type)
if not template:
logger.warning(f"No template found for {notification_type}")
return
# Render template
subject = self._render_template(template['subject'], context)
html_body = self._render_template(template['html'], context) if template['html'] else None
text_body = self._render_template(template['text'], context) if template['text'] else None
# Send email
try:
await self._send_email(
smtp_config,
user_email,
subject,
html_body,
text_body
)
logger.info(f"Sent {notification_type} notification to user {user_id}")
except Exception as e:
logger.error(f"Failed to send email to user {user_id}: {e}")
# Queue for retry
self._queue_notification(user_id, notification_type, context, str(e))
def _get_user_email(self, user_id: int) -> Optional[str]:
"""Get user email address"""
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
cursor.execute(f"""
SELECT email FROM users WHERE id = {placeholder}
""", (user_id,))
row = cursor.fetchone()
return row[0] if row else None
def _render_template(self, template: str, context: Dict) -> str:
"""
Simple template rendering using string formatting.
Args:
template: Template string with {variable} placeholders
context: Dict of variables
Returns:
Rendered string
"""
if not template:
return ""
try:
return template.format(**context)
except KeyError as e:
logger.warning(f"Missing template variable: {e}")
return template
async def _send_email(
self,
smtp_config: Dict,
to_email: str,
subject: str,
html_body: Optional[str],
text_body: Optional[str]
):
"""
Send email via SMTP.
Args:
smtp_config: SMTP configuration dict
to_email: Recipient email address
subject: Email subject
html_body: HTML body (optional)
text_body: Plain text body (optional)
"""
# Create message
msg = MIMEMultipart('alternative')
msg['Subject'] = subject
msg['From'] = f"{smtp_config['from_name']} <{smtp_config['from_email']}>"
msg['To'] = to_email
# Add text body
if text_body:
msg.attach(MIMEText(text_body, 'plain'))
# Add HTML body
if html_body:
msg.attach(MIMEText(html_body, 'html'))
# Send via SMTP
if smtp_config['use_tls']:
server = smtplib.SMTP(smtp_config['host'], smtp_config['port'])
server.starttls()
else:
server = smtplib.SMTP(smtp_config['host'], smtp_config['port'])
try:
if smtp_config['username'] and smtp_config['password']:
server.login(smtp_config['username'], smtp_config['password'])
server.send_message(msg)
finally:
server.quit()
def _queue_notification(
self,
user_id: int,
notification_type: str,
context: Dict,
error: str
):
"""
Queue notification for retry.
Args:
user_id: User ID
notification_type: Notification type
context: Template context
error: Error message
"""
import json
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
cursor.execute(f"""
INSERT INTO email_notification_queue
(user_id, notification_type, context_json, status, error_message, retry_count)
VALUES ({placeholder}, {placeholder}, {placeholder}, 'pending', {placeholder}, 0)
""", (user_id, notification_type, json.dumps(context), error))
conn.commit()
async def process_notification_queue(self):
"""
Process queued notifications.
Called by background scheduler to retry failed notifications.
"""
logger.info("Processing notification queue...")
try:
# Get pending notifications
pending = self._get_pending_notifications()
if not pending:
logger.debug("No pending notifications")
return
logger.info(f"Processing {len(pending)} pending notifications")
for notification in pending:
try:
await self.send_notification(
notification['user_id'],
notification['notification_type'],
notification['context']
)
self._mark_notification_sent(notification['id'])
except Exception as e:
logger.error(f"Error sending notification {notification['id']}: {e}")
self._increment_retry_count(notification['id'], str(e))
logger.info("Notification queue processing completed")
except Exception as e:
logger.error(f"Error processing notification queue: {e}", exc_info=True)
def _get_pending_notifications(self) -> list:
"""Get pending notifications from queue"""
import json
from datetime import datetime, timedelta
with self.db._get_connection() as conn:
cursor = conn.cursor()
# Get notifications ready for retry (not sent recently)
cursor.execute("""
SELECT id, user_id, notification_type, context_json, retry_count
FROM email_notification_queue
WHERE status = 'pending'
AND retry_count < 5
AND (next_retry_at IS NULL OR next_retry_at <= CURRENT_TIMESTAMP)
ORDER BY created_at ASC
LIMIT 10
""")
rows = cursor.fetchall()
return [
{
'id': row[0],
'user_id': row[1],
'notification_type': row[2],
'context': json.loads(row[3]),
'retry_count': row[4]
}
for row in rows
]
def _mark_notification_sent(self, notification_id: int):
"""Mark notification as sent"""
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
cursor.execute(f"""
UPDATE email_notification_queue
SET status = 'sent', sent_at = CURRENT_TIMESTAMP
WHERE id = {placeholder}
""", (notification_id,))
conn.commit()
def _increment_retry_count(self, notification_id: int, error: str):
"""Increment retry count and schedule next retry"""
from datetime import datetime, timedelta
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
# Calculate next retry time (exponential backoff)
cursor.execute(f"""
SELECT retry_count FROM email_notification_queue
WHERE id = {placeholder}
""", (notification_id,))
row = cursor.fetchone()
retry_count = row[0] + 1 if row else 1
# Exponential backoff: 5min, 15min, 1hr, 4hr, 12hr
backoff_minutes = [5, 15, 60, 240, 720]
delay_minutes = backoff_minutes[min(retry_count - 1, len(backoff_minutes) - 1)]
if self.db.db_type == 'sqlite':
next_retry = f"datetime('now', '+{delay_minutes} minutes')"
else:
next_retry = f"DATE_ADD(NOW(), INTERVAL {delay_minutes} MINUTE)"
cursor.execute(f"""
UPDATE email_notification_queue
SET retry_count = retry_count + 1,
error_message = {placeholder},
next_retry_at = {next_retry}
WHERE id = {placeholder}
""", (error, notification_id))
conn.commit()
# Convenience methods for common notifications
async def notify_payment_success(self, user_id: int, amount: float, currency: str):
"""Send payment success notification"""
await self.send_notification(
user_id,
self.PAYMENT_SUCCESS,
{
'amount': amount,
'currency': currency,
'date': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
)
async def notify_payment_failed(self, user_id: int, amount: float, currency: str, reason: str):
"""Send payment failed notification"""
await self.send_notification(
user_id,
self.PAYMENT_FAILED,
{
'amount': amount,
'currency': currency,
'reason': reason,
'date': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
)
async def notify_subscription_upgraded(self, user_id: int, old_tier: str, new_tier: str):
"""Send subscription upgraded notification"""
await self.send_notification(
user_id,
self.SUBSCRIPTION_UPGRADED,
{
'old_tier': old_tier,
'new_tier': new_tier,
'date': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
)
async def notify_subscription_downgraded(self, user_id: int, old_tier: str, new_tier: str):
"""Send subscription downgraded notification"""
await self.send_notification(
user_id,
self.SUBSCRIPTION_DOWNGRADED,
{
'old_tier': old_tier,
'new_tier': new_tier,
'date': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
)
async def notify_subscription_cancelled(self, user_id: int, tier: str, end_date: str):
"""Send subscription cancelled notification"""
await self.send_notification(
user_id,
self.SUBSCRIPTION_CANCELLED,
{
'tier': tier,
'end_date': end_date,
'date': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
)
aisbf/payments/scheduler.py 0 → 100644
View file @ 3412387c
"""
Payment System Background Scheduler
Runs periodic jobs for blockchain monitoring, subscription renewals,
payment retries, wallet consolidation, and price updates.
Uses distributed locking for horizontal scaling.
"""
import asyncio
import logging
from typing import Dict, Optional, Callable
from datetime import datetime, timedelta
import time
logger = logging.getLogger(__name__)
class PaymentScheduler:
"""
Background scheduler for payment system periodic tasks.
Runs jobs at configured intervals with distributed locking
to support horizontal scaling across multiple instances.
"""
def __init__(self, db_manager, payment_service):
"""
Initialize payment scheduler.
Args:
db_manager: DatabaseManager instance
payment_service: PaymentService instance
"""
self.db = db_manager
self.payment_service = payment_service
self.running = False
self.tasks = []
# Job configurations (name, interval_seconds, handler)
self.jobs = [
('blockchain_monitor', 60, self._run_blockchain_monitor),
('subscription_renewal', 300, self._run_subscription_renewal),
('payment_retry', 300, self._run_payment_retry),
('wallet_consolidation', 3600, self._run_wallet_consolidation),
('price_update', 300, self._run_price_update),
('notification_queue', 60, self._run_notification_queue),
]
async def start(self):
"""
Start the scheduler.
Launches background tasks for each job.
"""
if self.running:
logger.warning("Scheduler already running")
return
self.running = True
logger.info("Starting payment scheduler...")
# Start each job in its own task
for job_name, interval, handler in self.jobs:
task = asyncio.create_task(
self._run_job_loop(job_name, interval, handler)
)
self.tasks.append(task)
logger.info(f"Started {len(self.jobs)} scheduler jobs")
async def stop(self):
"""
Stop the scheduler.
Cancels all running tasks and waits for them to complete.
"""
if not self.running:
return
logger.info("Stopping payment scheduler...")
self.running = False
# Cancel all tasks
for task in self.tasks:
task.cancel()
# Wait for tasks to complete
await asyncio.gather(*self.tasks, return_exceptions=True)
self.tasks.clear()
logger.info("Payment scheduler stopped")
async def _run_job_loop(
self,
job_name: str,
interval: int,
handler: Callable
):
"""
Run a job in a loop at specified interval.
Args:
job_name: Name of the job
interval: Interval in seconds
handler: Async function to execute
"""
logger.info(f"Started job loop: {job_name} (interval: {interval}s)")
while self.running:
try:
# Try to acquire distributed lock
if await self._acquire_lock(job_name):
try:
logger.debug(f"Running job: {job_name}")
await handler()
finally:
await self._release_lock(job_name)
else:
logger.debug(f"Job {job_name} already running on another instance")
except asyncio.CancelledError:
logger.info(f"Job loop cancelled: {job_name}")
break
except Exception as e:
logger.error(f"Error in job {job_name}: {e}", exc_info=True)
# Wait for next interval
await asyncio.sleep(interval)
async def _acquire_lock(self, job_name: str, timeout: int = 300) -> bool:
"""
Acquire distributed lock for job.
Uses database row locking to ensure only one instance runs the job.
Args:
job_name: Name of the job
timeout: Lock timeout in seconds
Returns:
True if lock acquired, False otherwise
"""
try:
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
# Check if lock exists and is not expired
if self.db.db_type == 'sqlite':
expires_check = f"datetime('now', '-{timeout} seconds')"
else:
expires_check = f"DATE_SUB(NOW(), INTERVAL {timeout} SECOND)"
cursor.execute(f"""
SELECT locked_at, locked_by
FROM distributed_locks
WHERE lock_name = {placeholder}
""", (job_name,))
row = cursor.fetchone()
# If lock exists and not expired, can't acquire
if row:
locked_at = row[0]
# Check if expired
if self.db.db_type == 'sqlite':
cursor.execute(f"""
SELECT datetime('now') > datetime(?, '+{timeout} seconds')
""", (locked_at,))
else:
cursor.execute(f"""
SELECT NOW() > DATE_ADD(?, INTERVAL {timeout} SECOND)
""", (locked_at,))
is_expired = cursor.fetchone()[0]
if not is_expired:
return False
# Acquire or refresh lock
instance_id = f"scheduler_{id(self)}"
if row:
# Update existing lock
cursor.execute(f"""
UPDATE distributed_locks
SET locked_at = CURRENT_TIMESTAMP,
locked_by = {placeholder}
WHERE lock_name = {placeholder}
""", (instance_id, job_name))
else:
# Insert new lock
cursor.execute(f"""
INSERT INTO distributed_locks (lock_name, locked_at, locked_by)
VALUES ({placeholder}, CURRENT_TIMESTAMP, {placeholder})
""", (job_name, instance_id))
conn.commit()
return True
except Exception as e:
logger.error(f"Error acquiring lock for {job_name}: {e}")
return False
async def _release_lock(self, job_name: str):
"""
Release distributed lock for job.
Args:
job_name: Name of the job
"""
try:
with self.db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
cursor.execute(f"""
DELETE FROM distributed_locks
WHERE lock_name = {placeholder}
""", (job_name,))
conn.commit()
except Exception as e:
logger.error(f"Error releasing lock for {job_name}: {e}")
# Job handlers
async def _run_blockchain_monitor(self):
"""Run blockchain monitoring job"""
logger.info("Running blockchain monitor job")
try:
await self.payment_service.blockchain_monitor.check_crypto_payments()
except Exception as e:
logger.error(f"Blockchain monitor job failed: {e}", exc_info=True)
async def _run_subscription_renewal(self):
"""Run subscription renewal job"""
logger.info("Running subscription renewal job")
try:
await self.payment_service.renewal_processor.process_renewals()
except Exception as e:
logger.error(f"Subscription renewal job failed: {e}", exc_info=True)
async def _run_payment_retry(self):
"""Run payment retry job"""
logger.info("Running payment retry job")
try:
from aisbf.payments.subscription.retry import PaymentRetryProcessor
retry_processor = PaymentRetryProcessor(
self.db,
self.payment_service.subscription_manager
)
await retry_processor.process_retries()
except Exception as e:
logger.error(f"Payment retry job failed: {e}", exc_info=True)
async def _run_wallet_consolidation(self):
"""Run wallet consolidation job"""
logger.info("Running wallet consolidation job")
try:
from aisbf.payments.crypto.consolidation import WalletConsolidator
consolidator = WalletConsolidator(
self.db,
self.payment_service.wallet_manager
)
await consolidator.consolidate_wallets()
await consolidator.process_consolidation_queue()
except Exception as e:
logger.error(f"Wallet consolidation job failed: {e}", exc_info=True)
async def _run_price_update(self):
"""Run price update job"""
logger.info("Running price update job")
try:
await self.payment_service.price_service.update_all_prices()
except Exception as e:
logger.error(f"Price update job failed: {e}", exc_info=True)
async def _run_notification_queue(self):
"""Run notification queue processing job"""
logger.info("Running notification queue job")
try:
from aisbf.payments.notifications.email import EmailNotificationService
email_service = EmailNotificationService(self.db)
await email_service.process_notification_queue()
except Exception as e:
logger.error(f"Notification queue job failed: {e}", exc_info=True)
def get_job_status(self) -> Dict:
"""
Get status of all scheduled jobs.
Returns:
Dict with job statuses
"""
status = {
'running': self.running,
'jobs': []
}
with self.db._get_connection() as conn:
cursor = conn.cursor()
for job_name, interval, _ in self.jobs:
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
cursor.execute(f"""
SELECT locked_at, locked_by
FROM distributed_locks
WHERE lock_name = {placeholder}
""", (job_name,))
row = cursor.fetchone()
job_status = {
'name': job_name,
'interval': interval,
'locked': bool(row),
'locked_at': row[0] if row else None,
'locked_by': row[1] if row else None
}
status['jobs'].append(job_status)
return status
async def run_job_now(self, job_name: str):
"""
Manually trigger a job to run immediately.
Args:
job_name: Name of the job to run
"""
# Find job handler
handler = None
for name, _, h in self.jobs:
if name == job_name:
handler = h
break
if not handler:
raise ValueError(f"Unknown job: {job_name}")
logger.info(f"Manually running job: {job_name}")
# Try to acquire lock
if await self._acquire_lock(job_name):
try:
await handler()
logger.info(f"Job {job_name} completed successfully")
finally:
await self._release_lock(job_name)
else:
raise RuntimeError(f"Job {job_name} is already running")
main.py
View file @ 3412387c
......@@ -6195,6 +6195,327 @@ async def api_save_payment_gateways(request: Request):
return JSONResponse({"error": str(e)}, status_code=500)
# Admin configuration API endpoints
@app.get("/api/admin/config/price-sources")
async def get_price_sources(request: Request):
"""Get crypto price source configuration"""
auth_check = require_admin(request)
if auth_check:
return auth_check
db = DatabaseRegistry.get_config_database()
with db._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT crypto_type, price_source, api_key, update_interval_seconds, is_enabled
FROM crypto_price_sources
""")
rows = cursor.fetchall()
sources = [
{
'crypto_type': row[0],
'price_source': row[1],
'api_key': row[2],
'update_interval': row[3],
'enabled': bool(row[4])
}
for row in rows
]
return JSONResponse({'price_sources': sources})
@app.post("/api/admin/config/price-sources")
async def update_price_sources(request: Request):
"""Update crypto price source configuration"""
auth_check = require_admin(request)
if auth_check:
return auth_check
try:
body = await request.json()
db = DatabaseRegistry.get_config_database()
with db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if db.db_type == 'sqlite' else '%s'
for source in body.get('price_sources', []):
cursor.execute(f"""
UPDATE crypto_price_sources
SET price_source = {placeholder},
api_key = {placeholder},
update_interval_seconds = {placeholder},
is_enabled = {placeholder}
WHERE crypto_type = {placeholder}
""", (
source['price_source'],
source.get('api_key'),
source['update_interval'],
source['enabled'],
source['crypto_type']
))
conn.commit()
return JSONResponse({'success': True, 'message': 'Price sources updated'})
except Exception as e:
logger.error(f"Error updating price sources: {e}")
return JSONResponse({'error': str(e)}, status_code=500)
@app.get("/api/admin/config/consolidation")
async def get_consolidation_config(request: Request):
"""Get wallet consolidation configuration"""
auth_check = require_admin(request)
if auth_check:
return auth_check
db = DatabaseRegistry.get_config_database()
with db._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT crypto_type, threshold_amount, admin_address, is_enabled
FROM crypto_consolidation_settings
""")
rows = cursor.fetchall()
settings = [
{
'crypto_type': row[0],
'threshold': float(row[1]),
'admin_address': row[2],
'enabled': bool(row[3])
}
for row in rows
]
return JSONResponse({'consolidation_settings': settings})
@app.post("/api/admin/config/consolidation")
async def update_consolidation_config(request: Request):
"""Update wallet consolidation configuration"""
auth_check = require_admin(request)
if auth_check:
return auth_check
try:
body = await request.json()
db = DatabaseRegistry.get_config_database()
with db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if db.db_type == 'sqlite' else '%s'
for setting in body.get('consolidation_settings', []):
cursor.execute(f"""
UPDATE crypto_consolidation_settings
SET threshold_amount = {placeholder},
admin_address = {placeholder},
is_enabled = {placeholder}
WHERE crypto_type = {placeholder}
""", (
setting['threshold'],
setting['admin_address'],
setting['enabled'],
setting['crypto_type']
))
conn.commit()
return JSONResponse({'success': True, 'message': 'Consolidation settings updated'})
except Exception as e:
logger.error(f"Error updating consolidation settings: {e}")
return JSONResponse({'error': str(e)}, status_code=500)
@app.get("/api/admin/config/email")
async def get_email_config(request: Request):
"""Get email notification configuration"""
auth_check = require_admin(request)
if auth_check:
return auth_check
db = DatabaseRegistry.get_config_database()
# Get SMTP config
with db._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT smtp_host, smtp_port, smtp_username, from_email, from_name, use_tls
FROM email_config
LIMIT 1
""")
smtp_row = cursor.fetchone()
# Get notification settings
cursor.execute("""
SELECT notification_type, is_enabled, subject_template
FROM email_notification_settings
""")
notif_rows = cursor.fetchall()
smtp_config = None
if smtp_row:
smtp_config = {
'smtp_host': smtp_row[0],
'smtp_port': smtp_row[1],
'smtp_username': smtp_row[2],
'from_email': smtp_row[3],
'from_name': smtp_row[4],
'use_tls': bool(smtp_row[5])
}
notifications = [
{
'type': row[0],
'enabled': bool(row[1]),
'subject': row[2]
}
for row in notif_rows
]
return JSONResponse({
'smtp_config': smtp_config,
'notifications': notifications
})
@app.post("/api/admin/config/email")
async def update_email_config(request: Request):
"""Update email notification configuration"""
auth_check = require_admin(request)
if auth_check:
return auth_check
try:
body = await request.json()
db = DatabaseRegistry.get_config_database()
with db._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if db.db_type == 'sqlite' else '%s'
# Update SMTP config
if 'smtp_config' in body:
smtp = body['smtp_config']
# Check if config exists
cursor.execute("SELECT id FROM email_config LIMIT 1")
exists = cursor.fetchone()
if exists:
cursor.execute(f"""
UPDATE email_config
SET smtp_host = {placeholder},
smtp_port = {placeholder},
smtp_username = {placeholder},
smtp_password = {placeholder},
from_email = {placeholder},
from_name = {placeholder},
use_tls = {placeholder}
""", (
smtp['smtp_host'],
smtp['smtp_port'],
smtp.get('smtp_username'),
smtp.get('smtp_password'),
smtp['from_email'],
smtp.get('from_name'),
smtp.get('use_tls', True)
))
else:
cursor.execute(f"""
INSERT INTO email_config
(smtp_host, smtp_port, smtp_username, smtp_password, from_email, from_name, use_tls)
VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder})
""", (
smtp['smtp_host'],
smtp['smtp_port'],
smtp.get('smtp_username'),
smtp.get('smtp_password'),
smtp['from_email'],
smtp.get('from_name'),
smtp.get('use_tls', True)
))
# Update notification settings
if 'notifications' in body:
for notif in body['notifications']:
cursor.execute(f"""
UPDATE email_notification_settings
SET is_enabled = {placeholder},
subject_template = {placeholder}
WHERE notification_type = {placeholder}
""", (
notif['enabled'],
notif['subject'],
notif['type']
))
conn.commit()
return JSONResponse({'success': True, 'message': 'Email configuration updated'})
except Exception as e:
logger.error(f"Error updating email configuration: {e}")
return JSONResponse({'error': str(e)}, status_code=500)
@app.get("/api/admin/scheduler/status")
async def get_scheduler_status(request: Request):
"""Get payment scheduler status"""
auth_check = require_admin(request)
if auth_check:
return auth_check
if not payment_service:
return JSONResponse({'error': 'Payment service not initialized'}, status_code=503)
try:
from aisbf.payments.scheduler import PaymentScheduler
# Get scheduler from payment service if available
if hasattr(payment_service, 'scheduler'):
status = payment_service.scheduler.get_job_status()
return JSONResponse(status)
else:
return JSONResponse({'error': 'Scheduler not available'}, status_code=503)
except Exception as e:
logger.error(f"Error getting scheduler status: {e}")
return JSONResponse({'error': str(e)}, status_code=500)
@app.post("/api/admin/scheduler/run-job")
async def run_scheduler_job(request: Request):
"""Manually trigger a scheduler job"""
auth_check = require_admin(request)
if auth_check:
return auth_check
if not payment_service:
return JSONResponse({'error': 'Payment service not initialized'}, status_code=503)
try:
body = await request.json()
job_name = body.get('job_name')
if not job_name:
return JSONResponse({'error': 'job_name required'}, status_code=400)
if hasattr(payment_service, 'scheduler'):
await payment_service.scheduler.run_job_now(job_name)
return JSONResponse({'success': True, 'message': f'Job {job_name} triggered'})
else:
return JSONResponse({'error': 'Scheduler not available'}, status_code=503)
except ValueError as e:
return JSONResponse({'error': str(e)}, status_code=400)
except Exception as e:
logger.error(f"Error running scheduler job: {e}")
return JSONResponse({'error': str(e)}, status_code=500)
@app.get("/dashboard/pricing")
async def dashboard_pricing(request: Request):
......
tests/payments/test_integration.py 0 → 100644
View file @ 3412387c
"""
Integration tests for payment system
Tests complete payment flows including:
- Crypto payment flow
- Subscription creation and management
- Payment retries
- Wallet consolidation
- Email notifications
"""
import pytest
import asyncio
from decimal import Decimal
from datetime import datetime, timedelta
from unittest.mock import Mock, patch, AsyncMock
@pytest.fixture
def db_manager():
"""Create test database manager"""
from aisbf.database import DatabaseManager
# Use in-memory SQLite for tests
db = DatabaseManager(db_type='sqlite', db_path=':memory:')
# Run migrations
from aisbf.payments.migrations import PaymentMigrations
migrations = PaymentMigrations(db)
migrations.run_migrations()
# Create test user
with db._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO users (username, email, password_hash, role, email_verified)
VALUES ('testuser', 'test@example.com', 'hash', 'user', 1)
""")
conn.commit()
return db
@pytest.fixture
def payment_config():
"""Payment service configuration"""
return {
'encryption_key': 'test_key_32_bytes_long_exactly!!',
'currency_code': 'USD',
'btc_confirmations': 3,
'eth_confirmations': 12,
'stripe_api_key': 'test_stripe_key',
'paypal_client_id': 'test_paypal_id',
'paypal_client_secret': 'test_paypal_secret'
}
@pytest.fixture
async def payment_service(db_manager, payment_config):
"""Create payment service instance"""
from aisbf.payments.service import PaymentService
service = PaymentService(db_manager, payment_config)
await service.initialize()
return service
class TestCryptoPaymentFlow:
"""Test complete crypto payment flow"""
@pytest.mark.asyncio
async def test_create_crypto_address(self, payment_service, db_manager):
"""Test creating crypto address for user"""
# Get or create address
address = await payment_service.wallet_manager.get_or_create_user_address(1, 'btc')
assert address is not None
assert len(address) > 0
# Verify address stored in database
with db_manager._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT address FROM user_crypto_addresses
WHERE user_id = 1 AND crypto_type = 'btc'
""")
row = cursor.fetchone()
assert row is not None
assert row[0] == address
@pytest.mark.asyncio
async def test_detect_incoming_payment(self, payment_service, db_manager):
"""Test detecting incoming crypto payment"""
# Create address
address = await payment_service.wallet_manager.get_or_create_user_address(1, 'btc')
# Simulate incoming transaction
with db_manager._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO crypto_transactions
(user_id, crypto_type, tx_hash, from_address, to_address, amount_crypto, confirmations, status)
VALUES (1, 'btc', 'test_tx_hash', 'sender_addr', ?, 0.001, 3, 'pending')
""", (address,))
conn.commit()
# Process transaction (simulate blockchain monitor)
await payment_service.blockchain_monitor.process_transaction(
user_id=1,
crypto_type='btc',
tx_hash='test_tx_hash',
from_address='sender_addr',
to_address=address,
amount=Decimal('0.001'),
confirmations=3
)
# Verify wallet balance updated
with db_manager._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT balance_crypto FROM user_crypto_wallets
WHERE user_id = 1 AND crypto_type = 'btc'
""")
row = cursor.fetchone()
assert row is not None
assert Decimal(str(row[0])) > 0
class TestSubscriptionFlow:
"""Test subscription creation and management"""
@pytest.mark.asyncio
async def test_create_subscription(self, payment_service, db_manager):
"""Test creating a subscription"""
# Create tier
with db_manager._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO subscription_tiers
(name, price_monthly, price_yearly, features_json, is_active)
VALUES ('Pro', 10.00, 100.00, '{}', 1)
""")
conn.commit()
tier_id = cursor.lastrowid
# Create subscription
result = await payment_service.subscription_manager.create_subscription(
user_id=1,
tier_id=tier_id,
billing_cycle='monthly',
payment_method_id=1
)
assert result['success'] is True
assert 'subscription_id' in result
# Verify subscription in database
with db_manager._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT status FROM user_subscriptions
WHERE user_id = 1
""")
row = cursor.fetchone()
assert row is not None
assert row[0] == 'active'
@pytest.mark.asyncio
async def test_subscription_renewal(self, payment_service, db_manager):
"""Test subscription renewal process"""
# Create tier and subscription
with db_manager._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO subscription_tiers
(name, price_monthly, price_yearly, features_json, is_active)
VALUES ('Pro', 10.00, 100.00, '{}', 1)
""")
tier_id = cursor.lastrowid
# Create subscription expiring soon
cursor.execute("""
INSERT INTO user_subscriptions
(user_id, tier_id, status, billing_cycle, next_billing_date, payment_method_id)
VALUES (1, ?, 'active', 'monthly', date('now', '+1 day'), 1)
""", (tier_id,))
conn.commit()
# Process renewals
await payment_service.renewal_processor.process_renewals()
# Verify renewal was attempted
with db_manager._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT COUNT(*) FROM subscription_billing_history
WHERE user_id = 1
""")
count = cursor.fetchone()[0]
assert count > 0
class TestWalletConsolidation:
"""Test wallet consolidation"""
@pytest.mark.asyncio
async def test_consolidate_above_threshold(self, payment_service, db_manager):
"""Test consolidating wallet above threshold"""
from aisbf.payments.crypto.consolidation import WalletConsolidator
consolidator = WalletConsolidator(
db_manager,
payment_service.wallet_manager
)
# Create user wallet with balance above threshold
with db_manager._get_connection() as conn:
cursor = conn.cursor()
# Create address
cursor.execute("""
INSERT INTO user_crypto_addresses
(user_id, crypto_type, address, derivation_path, derivation_index)
VALUES (1, 'btc', 'user_btc_address', 'm/44/0/0/0/0', 0)
""")
# Create wallet with high balance
cursor.execute("""
INSERT INTO user_crypto_wallets
(user_id, crypto_type, balance_crypto, balance_fiat)
VALUES (1, 'btc', 1.5, 50000.00)
""")
conn.commit()
# Run consolidation
await consolidator.consolidate_wallets()
# Verify consolidation was queued
with db_manager._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT COUNT(*) FROM crypto_consolidation_queue
WHERE user_id = 1 AND status = 'pending'
""")
count = cursor.fetchone()[0]
assert count > 0
class TestEmailNotifications:
"""Test email notification system"""
@pytest.mark.asyncio
async def test_send_payment_success_notification(self, db_manager):
"""Test sending payment success notification"""
from aisbf.payments.notifications.email import EmailNotificationService
email_service = EmailNotificationService(db_manager)
# Configure SMTP (mock)
with db_manager._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO email_config
(smtp_host, smtp_port, smtp_username, smtp_password, from_email, from_name, use_tls)
VALUES ('smtp.test.com', 587, 'test', 'pass', 'noreply@test.com', 'Test', 1)
""")
conn.commit()
# Mock SMTP
with patch('smtplib.SMTP') as mock_smtp:
mock_server = Mock()
mock_smtp.return_value = mock_server
# Send notification
await email_service.notify_payment_success(
user_id=1,
amount=10.00,
currency='USD'
)
# Verify SMTP was called
assert mock_smtp.called
@pytest.mark.asyncio
async def test_notification_queue_retry(self, db_manager):
"""Test notification retry on failure"""
from aisbf.payments.notifications.email import EmailNotificationService
email_service = EmailNotificationService(db_manager)
# Queue a notification
email_service._queue_notification(
user_id=1,
notification_type='payment_success',
context={'amount': 10.00, 'currency': 'USD'},
error='SMTP connection failed'
)
# Verify queued
with db_manager._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT COUNT(*) FROM email_notification_queue
WHERE user_id = 1 AND status = 'pending'
""")
count = cursor.fetchone()[0]
assert count == 1
class TestPaymentScheduler:
"""Test payment scheduler"""
@pytest.mark.asyncio
async def test_scheduler_distributed_lock(self, payment_service, db_manager):
"""Test scheduler distributed locking"""
from aisbf.payments.scheduler import PaymentScheduler
scheduler = PaymentScheduler(db_manager, payment_service)
# Acquire lock
acquired = await scheduler._acquire_lock('test_job')
assert acquired is True
# Try to acquire again (should fail)
acquired2 = await scheduler._acquire_lock('test_job')
assert acquired2 is False
# Release lock
await scheduler._release_lock('test_job')
# Should be able to acquire again
acquired3 = await scheduler._acquire_lock('test_job')
assert acquired3 is True
@pytest.mark.asyncio
async def test_run_job_manually(self, payment_service, db_manager):
"""Test manually triggering a job"""
from aisbf.payments.scheduler import PaymentScheduler
scheduler = PaymentScheduler(db_manager, payment_service)
# Mock job handler
job_executed = False
async def mock_handler():
nonlocal job_executed
job_executed = True
# Replace handler
scheduler.jobs[0] = ('test_job', 60, mock_handler)
# Run job
await scheduler.run_job_now('test_job')
assert job_executed is True
class TestEndToEndFlow:
"""Test complete end-to-end payment flow"""
@pytest.mark.asyncio
async def test_complete_crypto_subscription_flow(self, payment_service, db_manager):
"""Test complete flow: crypto payment -> subscription creation -> renewal"""
# 1. Create crypto address
address = await payment_service.wallet_manager.get_or_create_user_address(1, 'btc')
assert address is not None
# 2. Simulate incoming payment
with db_manager._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO crypto_transactions
(user_id, crypto_type, tx_hash, from_address, to_address, amount_crypto, confirmations, status)
VALUES (1, 'btc', 'tx_001', 'sender', ?, 0.01, 3, 'confirmed')
""", (address,))
# Credit wallet
cursor.execute("""
INSERT INTO user_crypto_wallets
(user_id, crypto_type, balance_crypto, balance_fiat)
VALUES (1, 'btc', 0.01, 500.00)
""")
conn.commit()
# 3. Create subscription tier
with db_manager._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO subscription_tiers
(name, price_monthly, price_yearly, features_json, is_active)
VALUES ('Pro', 10.00, 100.00, '{}', 1)
""")
tier_id = cursor.lastrowid
conn.commit()
# 4. Create subscription
result = await payment_service.subscription_manager.create_subscription(
user_id=1,
tier_id=tier_id,
billing_cycle='monthly',
payment_method_id=1
)
assert result['success'] is True
# 5. Verify subscription active
with db_manager._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT status FROM user_subscriptions WHERE user_id = 1
""")
status = cursor.fetchone()[0]
assert status == 'active'
if __name__ == '__main__':
pytest.main([__file__, '-v'])
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment