import unittest import os import shutil import tempfile import uuid from decimal import Decimal from datetime import datetime # Adjust Python path to load modules correctly import sys sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) import ledger_core as core import registry import exports class TestLedgerSystem(unittest.TestCase): def setUp(self): # Create a temporary directory for isolated test databases self.test_dir = tempfile.mkdtemp() self.db_path = os.path.join(self.test_dir, "test_ledger.db") self.registry_db_path = os.path.join(self.test_dir, "test_workspace.db") # Initialize thread-local database path core.set_active_db(self.db_path) core.init_db() # Initialize registry database registry.DB_PATH = self.registry_db_path registry.init_registry_db() def tearDown(self): # Cleanup temporary files shutil.rmtree(self.test_dir) core.set_active_db(None) # ---------------------------------------------------- # 1. Date Normalization & CSV Validation # ---------------------------------------------------- def test_date_normalization(self): self.assertEqual(core.parse_and_normalize_date("2026-05-31"), "2026-05-31") self.assertEqual(core.parse_and_normalize_date("31/05/2026"), "2026-05-31") self.assertEqual(core.parse_and_normalize_date("31-05-2026"), "2026-05-31") self.assertEqual(core.parse_and_normalize_date("05/31/2026"), "2026-05-31") self.assertEqual(core.parse_and_normalize_date("2026/05/31"), "2026-05-31") with self.assertRaises(ValueError): core.parse_and_normalize_date("not-a-date") def test_validate_entry_csv(self): # Balanced non-contra rows = [ {"date": "2026-01-01", "account": "1000", "amount": "100.00"}, {"date": "2026-01-01", "account": "2000", "amount": "-100.00"} ] ok, err, parsed, warnings = core.validate_entry_csv(rows) self.assertTrue(ok) self.assertEqual(len(parsed), 2) self.assertEqual(len(warnings), 0) # Unbalanced non-contra rows_bad = [ {"date": "2026-01-01", "account": "1000", "amount": "100.00"}, {"date": "2026-01-01", "account": "2000", "amount": "-90.00"} ] ok, err, parsed, warnings = core.validate_entry_csv(rows_bad) self.assertFalse(ok) self.assertIn("do not net to zero", err) # P&L warning check rows_pl = [ {"date": "2026-01-01", "account": "4000", "amount": "100.00"}, {"date": "2026-01-01", "account": "2000", "amount": "-100.00"} ] ok, err, parsed, warnings = core.validate_entry_csv(rows_pl, source_type='OPENING') self.assertTrue(ok) self.assertEqual(len(warnings), 1) self.assertIn("temporary P&L account", warnings[0]) # ---------------------------------------------------- # 2. Registry & Password Hashing # ---------------------------------------------------- def test_registry_user_and_engagement(self): # Create user ok, msg = registry.register_user("admin", "password123") self.assertTrue(ok) # Verify user self.assertTrue(registry.verify_user("admin", "password123")) self.assertFalse(registry.verify_user("admin", "wrongpassword")) self.assertFalse(registry.verify_user("nonexistent", "password123")) # Create/Unlock Engagement ok, msg = registry.register_engagement("CompanyA", "passphraseA") self.assertTrue(ok) self.assertTrue(registry.verify_engagement("CompanyA", "passphraseA")) self.assertFalse(registry.verify_engagement("CompanyA", "wrongpassphrase")) # ---------------------------------------------------- # 3. Period Locking & Guards # ---------------------------------------------------- def test_period_locking(self): # Configure company with lock date core.save_company("CO1", "Test Company", "123", "456", "Addr", "USD", "31 Dec", "monthly", "2026-01-15") # Helper to upload entries rows = [ {"date": "2026-01-10", "account": "1000", "amount": "50.00", "contra": "", "reference": "Locked"}, {"date": "2026-01-10", "account": "2000", "amount": "-50.00", "contra": "", "reference": "Locked"} ] headers = ["date", "account", "amount", "contra", "reference"] uid = core.upload_entry_batch("test.csv", "", headers, rows) # Check lock guard blocks post/unpost/delete locked, msg = core.is_batch_locked(uid) self.assertTrue(locked) self.assertIn("locked on or before", msg) # Upload a batch after lock date rows_ok = [ {"date": "2026-01-20", "account": "1000", "amount": "50.00", "contra": "", "reference": "Unlocked"}, {"date": "2026-01-20", "account": "2000", "amount": "-50.00", "contra": "", "reference": "Unlocked"} ] uid_ok = core.upload_entry_batch("test_ok.csv", "", headers, rows_ok) locked_ok, msg_ok = core.is_batch_locked(uid_ok) self.assertFalse(locked_ok) # ---------------------------------------------------- # 4. Year-End Rollover & Anti-Double-Count # ---------------------------------------------------- def test_year_end_rollover(self): # Setup Chart of Accounts coa_uid = str(uuid.uuid4()) conn = core.get_db() c = conn.cursor() c.execute("INSERT INTO uploads (upload_id, type, filename, status) VALUES (?, 'COA', 'coa.csv', 'POSTED')", (coa_uid,)) c.execute("INSERT INTO coa_rows (upload_id, account_no, description) VALUES (?, '1000', 'Cash')", (coa_uid,)) c.execute("INSERT INTO coa_rows (upload_id, account_no, description) VALUES (?, '3000', 'Retained Earnings')", (coa_uid,)) c.execute("INSERT INTO coa_rows (upload_id, account_no, description) VALUES (?, '4000', 'Revenue')", (coa_uid,)) # Setup transactional entries for current year entry_uid = str(uuid.uuid4()) c.execute("INSERT INTO uploads (upload_id, type, filename, status) VALUES (?, 'ENTRY', 'entries.csv', 'POSTED')", (entry_uid,)) # Cash increase c.execute("INSERT INTO entry_rows (upload_id, date, account, amount) VALUES (?, '2025-06-30', '1000', 500.00)", (entry_uid,)) # Revenue earned (P&L account) c.execute("INSERT INTO entry_rows (upload_id, date, account, amount) VALUES (?, '2025-06-30', '4000', -500.00)", (entry_uid,)) conn.commit() conn.close() # Run year-end rollover success, rollover_uid = core.generate_rollover_batch( year_end_date="2025-12-31", opening_date="2026-01-01", retained_earnings_acct="3000", pl_starts_from="4000" ) self.assertTrue(success) # Verify the generated opening entries conn = core.get_db() c = conn.cursor() c.execute("SELECT account, amount, reference FROM entry_rows WHERE upload_id = ? ORDER BY account", (rollover_uid,)) rows = c.fetchall() conn.close() # We expect two opening entries: # 1. Cash (Asset - BS) rolled forward: 500.00 # 2. Retained Earnings (Equity - BS) absorbed P&L: -500.00 # 3. Revenue (P&L) should be cleared (no opening entry of 4000) self.assertEqual(len(rows), 2) self.assertEqual(rows[0][0], "1000") self.assertEqual(Decimal(str(rows[0][1])), Decimal("500.00")) self.assertEqual(rows[1][0], "3000") self.assertEqual(Decimal(str(rows[1][1])), Decimal("-500.00")) # Verify duplicate rollover block works success2, err2 = core.generate_rollover_batch( year_end_date="2025-12-31", opening_date="2026-01-01", retained_earnings_acct="3000", pl_starts_from="4000" ) self.assertFalse(success2) self.assertIn("already exists", err2) # ---------------------------------------------------- # 5. Backups & Restore Validation # ---------------------------------------------------- def test_backup_and_restore(self): # Configure company details core.save_company("CO2", "Backup Corp", "999", "888", "Nowhere", "EUR", "31 Dec", "monthly", None) # Perform backup backup_zip = os.path.join(self.test_dir, "backup.zip") manifest_data = core.get_backup_manifest_stats() # Let's use the core backup wrapper (needs sqlite online backup first) tmp_db = os.path.join(self.test_dir, "temp_backup.db") core.backup_db(tmp_db) # Write zip backup file import zipfile import json with zipfile.ZipFile(backup_zip, 'w') as zf: zf.write(tmp_db, "ledger.db") zf.writestr("manifest.json", json.dumps(manifest_data)) # Check manifest stats self.assertEqual(manifest_data["company_name"], "Backup Corp") # Test restore check validation # Create a fresh database target to restore into restore_target_db = os.path.join(self.test_dir, "restored.db") # Run restore check logic with zipfile.ZipFile(backup_zip, 'r') as zf: # Check structure self.assertIn("ledger.db", zf.namelist()) self.assertIn("manifest.json", zf.namelist()) # Extract to temp temp_extracted = os.path.join(self.test_dir, "extracted.db") zf.extract("ledger.db", self.test_dir) os.rename(os.path.join(self.test_dir, "ledger.db"), temp_extracted) # Verify schema conn = sqlite3_conn = sqlite3_connect(temp_extracted) c = conn.cursor() c.execute("SELECT name FROM sqlite_master WHERE type='table'") tables = [r[0] for r in c.fetchall()] conn.close() self.assertIn("company", tables) self.assertIn("uploads", tables) self.assertIn("coa_rows", tables) self.assertIn("entry_rows", tables) def sqlite3_connect(path): import sqlite3 return sqlite3.connect(path) if __name__ == "__main__": unittest.main()