#!/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()