247 lines
9.0 KiB
Python
247 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Tests for financial operations in maritime_db.py
|
|
Covers: charge_user, update_user_balance, charge_and_log, edge cases.
|
|
Uses SQLite in-memory (no DATABASE_URL → SQLite mode).
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import sqlite3
|
|
import unittest
|
|
|
|
# Force SQLite mode (no PostgreSQL)
|
|
os.environ.pop('DATABASE_URL', None)
|
|
os.environ['SECRET_KEY'] = 'test-secret-key-for-tests'
|
|
|
|
# Add parent dir to path
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
import maritime_db as db
|
|
|
|
|
|
class _NoCloseConnection:
|
|
"""Wrapper that delegates everything to real sqlite3.Connection but ignores close()."""
|
|
|
|
def __init__(self, conn):
|
|
self._conn = conn
|
|
|
|
def close(self):
|
|
pass # Don't actually close — we reuse this connection
|
|
|
|
def real_close(self):
|
|
self._conn.close()
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self._conn, name)
|
|
|
|
|
|
class TestFinancialOperations(unittest.TestCase):
|
|
"""Test suite for wallet / charging / logging operations."""
|
|
|
|
def setUp(self):
|
|
"""Create a fresh in-memory DB and test user for each test."""
|
|
db.USE_POSTGRES = False
|
|
|
|
# Create in-memory connection with no-close wrapper
|
|
raw_conn = sqlite3.connect(":memory:")
|
|
raw_conn.row_factory = sqlite3.Row
|
|
self._wrapper = _NoCloseConnection(raw_conn)
|
|
self._raw_conn = raw_conn
|
|
|
|
# Monkey-patch get_connection to return our wrapper
|
|
self._original_get_connection = db.get_connection
|
|
db.get_connection = lambda: self._wrapper
|
|
|
|
# Init schema
|
|
db.init_db()
|
|
|
|
# Create test user with $100 balance
|
|
cursor = raw_conn.cursor()
|
|
cursor.execute(
|
|
"INSERT INTO users (email, password_hash, name, lang, balance) VALUES (?, ?, ?, ?, ?)",
|
|
('test@example.com', 'hash123', 'Test User', 'en', 100.00)
|
|
)
|
|
raw_conn.commit()
|
|
self.user_id = cursor.lastrowid
|
|
|
|
def tearDown(self):
|
|
"""Restore original functions and close connection."""
|
|
db.get_connection = self._original_get_connection
|
|
self._raw_conn.close()
|
|
|
|
def _get_balance(self):
|
|
cursor = self._raw_conn.cursor()
|
|
cursor.execute("SELECT balance FROM users WHERE id = ?", (self.user_id,))
|
|
row = cursor.fetchone()
|
|
return row[0] if row else None
|
|
|
|
# --- charge_user tests ---
|
|
|
|
def test_charge_user_success(self):
|
|
"""Charge $10 from $100 balance → $90 remaining."""
|
|
result = db.charge_user(self.user_id, 10.0)
|
|
self.assertTrue(result)
|
|
self.assertAlmostEqual(self._get_balance(), 90.0, places=2)
|
|
|
|
def test_charge_user_exact_balance(self):
|
|
"""Charge exact balance → $0 remaining."""
|
|
result = db.charge_user(self.user_id, 100.0)
|
|
self.assertTrue(result)
|
|
self.assertAlmostEqual(self._get_balance(), 0.0, places=2)
|
|
|
|
def test_charge_user_insufficient_funds(self):
|
|
"""Charge more than balance → fails, balance unchanged."""
|
|
result = db.charge_user(self.user_id, 150.0)
|
|
self.assertFalse(result)
|
|
self.assertAlmostEqual(self._get_balance(), 100.0, places=2)
|
|
|
|
def test_charge_user_zero_amount(self):
|
|
"""Charge $0 → rejected (amount must be > 0)."""
|
|
result = db.charge_user(self.user_id, 0.0)
|
|
self.assertFalse(result)
|
|
self.assertAlmostEqual(self._get_balance(), 100.0, places=2)
|
|
|
|
def test_charge_user_negative_amount(self):
|
|
"""Charge negative → rejected."""
|
|
result = db.charge_user(self.user_id, -10.0)
|
|
self.assertFalse(result)
|
|
self.assertAlmostEqual(self._get_balance(), 100.0, places=2)
|
|
|
|
def test_charge_user_rounding(self):
|
|
"""Charge $10.005 → rounded to $10.01."""
|
|
result = db.charge_user(self.user_id, 10.005)
|
|
self.assertTrue(result)
|
|
self.assertAlmostEqual(self._get_balance(), 89.99, places=2)
|
|
|
|
def test_charge_user_nonexistent_user(self):
|
|
"""Charge non-existent user → False."""
|
|
result = db.charge_user(99999, 10.0)
|
|
self.assertFalse(result)
|
|
|
|
# --- update_user_balance tests ---
|
|
|
|
def test_update_balance_add(self):
|
|
"""Add $50 to $100 → $150."""
|
|
db.update_user_balance(self.user_id, 50.0)
|
|
self.assertAlmostEqual(self._get_balance(), 150.0, places=2)
|
|
|
|
def test_update_balance_zero_rejected(self):
|
|
"""Add $0 → ValueError."""
|
|
with self.assertRaises(ValueError):
|
|
db.update_user_balance(self.user_id, 0.0)
|
|
|
|
def test_update_balance_negative_rejected(self):
|
|
"""Add negative → ValueError."""
|
|
with self.assertRaises(ValueError):
|
|
db.update_user_balance(self.user_id, -10.0)
|
|
|
|
# --- charge_and_log tests ---
|
|
|
|
def test_charge_and_log_success(self):
|
|
"""Atomic charge+log: balance decreases, service_charges has entry."""
|
|
result = db.charge_and_log(self.user_id, 25.0, 'unlock_contacts', 'Test details')
|
|
self.assertTrue(result)
|
|
self.assertAlmostEqual(self._get_balance(), 75.0, places=2)
|
|
|
|
# Verify service charge was logged
|
|
cursor = self._raw_conn.cursor()
|
|
cursor.execute("SELECT * FROM service_charges WHERE user_id = ?", (self.user_id,))
|
|
charges = cursor.fetchall()
|
|
self.assertEqual(len(charges), 1)
|
|
|
|
def test_charge_and_log_with_contacts(self):
|
|
"""Atomic charge+log+contacts: all three tables updated."""
|
|
contacts = [{'name': 'John', 'email': 'john@test.com'}]
|
|
result = db.charge_and_log(
|
|
self.user_id, 10.0, 'unlock_contacts',
|
|
contacts=contacts, query='MMSI 123456', contact_type='owner'
|
|
)
|
|
self.assertTrue(result)
|
|
self.assertAlmostEqual(self._get_balance(), 90.0, places=2)
|
|
|
|
# Check purchased_contacts
|
|
cursor = self._raw_conn.cursor()
|
|
cursor.execute("SELECT * FROM purchased_contacts WHERE user_id = ?", (self.user_id,))
|
|
purchased = cursor.fetchall()
|
|
self.assertEqual(len(purchased), 1)
|
|
|
|
def test_charge_and_log_insufficient_funds(self):
|
|
"""Atomic charge with insufficient funds → no log, no contacts."""
|
|
result = db.charge_and_log(self.user_id, 200.0, 'unlock_contacts')
|
|
self.assertFalse(result)
|
|
self.assertAlmostEqual(self._get_balance(), 100.0, places=2)
|
|
|
|
# No service charge should be logged
|
|
cursor = self._raw_conn.cursor()
|
|
cursor.execute("SELECT * FROM service_charges WHERE user_id = ?", (self.user_id,))
|
|
self.assertEqual(len(cursor.fetchall()), 0)
|
|
|
|
def test_charge_and_log_zero_rejected(self):
|
|
"""Atomic charge $0 → rejected."""
|
|
result = db.charge_and_log(self.user_id, 0.0, 'test')
|
|
self.assertFalse(result)
|
|
|
|
def test_charge_and_log_negative_rejected(self):
|
|
"""Atomic charge negative → rejected."""
|
|
result = db.charge_and_log(self.user_id, -5.0, 'test')
|
|
self.assertFalse(result)
|
|
|
|
# --- Sequential charges (race condition proxy) ---
|
|
|
|
def test_sequential_charges_drain(self):
|
|
"""Multiple charges draining balance to zero."""
|
|
for i in range(10):
|
|
db.charge_user(self.user_id, 10.0)
|
|
self.assertAlmostEqual(self._get_balance(), 0.0, places=2)
|
|
|
|
# 11th charge should fail
|
|
result = db.charge_user(self.user_id, 10.0)
|
|
self.assertFalse(result)
|
|
|
|
def test_multiple_charge_and_log(self):
|
|
"""Multiple atomic operations in sequence."""
|
|
db.charge_and_log(self.user_id, 30.0, 'service_a')
|
|
db.charge_and_log(self.user_id, 20.0, 'service_b')
|
|
self.assertAlmostEqual(self._get_balance(), 50.0, places=2)
|
|
|
|
cursor = self._raw_conn.cursor()
|
|
cursor.execute("SELECT COUNT(*) FROM service_charges WHERE user_id = ?", (self.user_id,))
|
|
self.assertEqual(cursor.fetchone()[0], 2)
|
|
|
|
# --- add_service_charge tests ---
|
|
|
|
def test_add_service_charge(self):
|
|
"""Service charge logged correctly."""
|
|
db.add_service_charge(self.user_id, 'test_service', 15.0, 'Test detail')
|
|
cursor = self._raw_conn.cursor()
|
|
cursor.execute("SELECT service, amount, details FROM service_charges WHERE user_id = ?", (self.user_id,))
|
|
row = cursor.fetchone()
|
|
self.assertEqual(row[0], 'test_service')
|
|
self.assertAlmostEqual(row[1], 15.0, places=2)
|
|
self.assertEqual(row[2], 'Test detail')
|
|
|
|
# --- has_purchased_contact tests ---
|
|
|
|
def test_has_purchased_contact_empty(self):
|
|
"""No purchased contacts → empty list."""
|
|
result = db.has_purchased_contact(self.user_id, 'MMSI 123')
|
|
self.assertEqual(result, [])
|
|
|
|
def test_has_purchased_contact_found(self):
|
|
"""After purchase, contact is found."""
|
|
cursor = self._raw_conn.cursor()
|
|
cursor.execute(
|
|
"INSERT INTO purchased_contacts (user_id, contact_data, query, contact_type, amount_paid) VALUES (?, ?, ?, ?, ?)",
|
|
(self.user_id, json.dumps({'name': 'Test'}), 'MMSI 123', 'owner', 10.0)
|
|
)
|
|
self._raw_conn.commit()
|
|
|
|
result = db.has_purchased_contact(self.user_id, 'MMSI 123')
|
|
self.assertTrue(len(result) > 0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|