#!/usr/bin/env python """ Comprehensive Integration Test for Grazuri Migration Tests the entire system end-to-end to ensure compatibility """ import os import django import sys from decimal import Decimal from datetime import datetime, timedelta os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'branch_system.settings') django.setup() from django.contrib.auth import get_user_model from django.db import connection, transaction from loans.models import LoanProduct, LoanApplication, Loan from users.models import Branch User = get_user_model() class IntegrationTester: def __init__(self): self.test_results = [] self.created_objects = { 'branches': [], 'users': [], 'applications': [], 'loans': [] } def log_result(self, test_name, passed, message=""): """Log test result""" status = "✓ PASS" if passed else "✗ FAIL" self.test_results.append((test_name, passed, message)) print(f"{status}: {test_name}") if message: print(f" {message}") return passed def test_1_check_branches(self): """Test 1: Check if branches exist and create test branches""" print("\n" + "=" * 80) print("TEST 1: Branch System") print("=" * 80) try: # Check if Branch model exists branch_count = Branch.objects.count() print(f"\nExisting branches: {branch_count}") # Create test branches if needed test_branches = [ {'name': 'Nairobi HQ', 'code': 'NRB-HQ', 'location': 'Nairobi CBD'}, {'name': 'Mombasa Branch', 'code': 'MSA-01', 'location': 'Mombasa'}, {'name': 'Kisumu Branch', 'code': 'KSM-01', 'location': 'Kisumu'}, ] for branch_data in test_branches: branch, created = Branch.objects.get_or_create( code=branch_data['code'], defaults={ 'name': branch_data['name'], 'location': branch_data['location'], 'is_active': True } ) if created: self.created_objects['branches'].append(branch) print(f" Created: {branch.name} ({branch.code})") else: print(f" Exists: {branch.name} ({branch.code})") return self.log_result("Branch System", True, f"Total branches: {Branch.objects.count()}") except Exception as e: return self.log_result("Branch System", False, f"Error: {str(e)}") def test_2_check_loan_products(self): """Test 2: Verify Grazuri loan products""" print("\n" + "=" * 80) print("TEST 2: Grazuri Loan Products") print("=" * 80) try: biashara = LoanProduct.objects.filter(product_type='biashara', is_active=True).first() logbook = LoanProduct.objects.filter(product_type='logbook', is_active=True).first() if not biashara: return self.log_result("Loan Products", False, "Biashara Loan not found") if not logbook: return self.log_result("Loan Products", False, "Log Book Loan not found") print(f"\n✓ Biashara Loan: {biashara.name}") print(f" GL Code: {biashara.gl_code}") print(f" Account Type: {biashara.grazuri_account_type}") print(f" Interest: {biashara.interest_rate}%") print(f"\n✓ Log Book Loan: {logbook.name}") print(f" GL Code: {logbook.gl_code}") print(f" Account Type: {logbook.grazuri_account_type}") print(f" Interest: {logbook.interest_rate}%") return self.log_result("Loan Products", True, "Both Grazuri products active") except Exception as e: return self.log_result("Loan Products", False, f"Error: {str(e)}") def test_3_create_test_clients(self): """Test 3: Create test clients in different branches""" print("\n" + "=" * 80) print("TEST 3: Create Test Clients") print("=" * 80) try: branches = Branch.objects.all()[:3] if not branches: return self.log_result("Create Clients", False, "No branches available") test_clients = [ { 'username': 'test_client_1', 'email': 'client1@test.com', 'first_name': 'John', 'last_name': 'Kamau', 'phone_number': '+254712345001', 'id_number': 'ID12345001', 'role': 'borrower', }, { 'username': 'test_client_2', 'email': 'client2@test.com', 'first_name': 'Mary', 'last_name': 'Wanjiku', 'phone_number': '+254712345002', 'id_number': 'ID12345002', 'role': 'borrower', }, { 'username': 'test_client_3', 'email': 'client3@test.com', 'first_name': 'Peter', 'last_name': 'Omondi', 'phone_number': '+254712345003', 'id_number': 'ID12345003', 'role': 'borrower', }, ] created_count = 0 for i, client_data in enumerate(test_clients): branch = branches[i % len(branches)] user, created = User.objects.get_or_create( username=client_data['username'], defaults={ **client_data, 'branch': branch, 'is_active': True, 'status': 'active', } ) if created: user.set_password('Test@1234') user.save() self.created_objects['users'].append(user) created_count += 1 print(f" Created: {user.get_full_name()} at {branch.name}") else: print(f" Exists: {user.get_full_name()} at {branch.name}") total_clients = User.objects.filter(role='borrower').count() return self.log_result("Create Clients", True, f"Total clients: {total_clients}") except Exception as e: return self.log_result("Create Clients", False, f"Error: {str(e)}") def test_4_create_loan_applications(self): """Test 4: Create loan applications with Grazuri products""" print("\n" + "=" * 80) print("TEST 4: Create Loan Applications") print("=" * 80) try: biashara = LoanProduct.objects.filter(product_type='biashara', is_active=True).first() logbook = LoanProduct.objects.filter(product_type='logbook', is_active=True).first() clients = User.objects.filter(role='borrower')[:3] if not clients: return self.log_result("Loan Applications", False, "No clients available") applications_data = [ { 'borrower': clients[0], 'product': biashara, 'amount': Decimal('50000.00'), 'duration': 90, # 3 months 'purpose': 'Business expansion - inventory purchase' }, { 'borrower': clients[1] if len(clients) > 1 else clients[0], 'product': logbook, 'amount': Decimal('100000.00'), 'duration': 180, # 6 months 'purpose': 'Vehicle logbook loan for business capital' }, { 'borrower': clients[2] if len(clients) > 2 else clients[0], 'product': biashara, 'amount': Decimal('25000.00'), 'duration': 60, # 2 months 'purpose': 'Working capital for shop' }, ] created_count = 0 for app_data in applications_data: try: # Calculate loan amounts months = Decimal(str(app_data['duration'])) / Decimal('30') months = max(Decimal('1'), months) interest = app_data['amount'] * (app_data['product'].interest_rate / Decimal('100')) * months fee = app_data['amount'] * (app_data['product'].processing_fee / Decimal('100')) total = app_data['amount'] + interest + fee application = LoanApplication.objects.create( borrower=app_data['borrower'], loan_product=app_data['product'], requested_amount=app_data['amount'], requested_duration=app_data['duration'], purpose=app_data['purpose'], repayment_method='monthly', interest_amount=interest, processing_fee_amount=fee, total_amount=total, status='pending' ) self.created_objects['applications'].append(application) created_count += 1 print(f"\n Created Application: {application.application_number}") print(f" Borrower: {app_data['borrower'].get_full_name()}") print(f" Product: {app_data['product'].name}") print(f" Amount: KES {app_data['amount']:,.2f}") print(f" Interest: KES {interest:,.2f}") print(f" Fee: KES {fee:,.2f}") print(f" Total: KES {total:,.2f}") except Exception as e: print(f" Error creating application: {str(e)}") return self.log_result("Loan Applications", created_count > 0, f"Created {created_count} applications") except Exception as e: return self.log_result("Loan Applications", False, f"Error: {str(e)}") def test_5_approve_and_disburse_loans(self): """Test 5: Approve and disburse test loans""" print("\n" + "=" * 80) print("TEST 5: Approve and Disburse Loans") print("=" * 80) try: # Create or get admin user admin_user, created = User.objects.get_or_create( username='test_admin', defaults={ 'email': 'admin@test.com', 'first_name': 'Admin', 'last_name': 'User', 'role': 'admin', 'is_staff': True, 'is_superuser': True, 'is_active': True, } ) if created: admin_user.set_password('Admin@1234') admin_user.save() # Approve pending applications pending_apps = LoanApplication.objects.filter(status='pending') if not pending_apps: return self.log_result("Approve Loans", False, "No pending applications") approved_count = 0 for application in pending_apps: try: loan = application.approve( approved_by=admin_user, notes="Test approval for integration testing" ) self.created_objects['loans'].append(loan) approved_count += 1 print(f"\n Approved: {loan.loan_number}") print(f" Borrower: {loan.borrower.get_full_name()}") print(f" Principal: KES {loan.principal_amount:,.2f}") print(f" Total: KES {loan.total_amount:,.2f}") print(f" Due Date: {loan.due_date.strftime('%Y-%m-%d')}") except Exception as e: print(f" Error approving application {application.application_number}: {str(e)}") return self.log_result("Approve Loans", approved_count > 0, f"Approved {approved_count} loans") except Exception as e: return self.log_result("Approve Loans", False, f"Error: {str(e)}") def test_6_check_database_schema(self): """Test 6: Verify database schema compatibility""" print("\n" + "=" * 80) print("TEST 6: Database Schema Compatibility") print("=" * 80) try: with connection.cursor() as cursor: # Check critical tables critical_tables = [ 'users', 'branches', 'loan_products', 'loan_applications', 'loans', 'repayments' ] missing_tables = [] for table in critical_tables: cursor.execute(f"SHOW TABLES LIKE '{table}'") if not cursor.fetchone(): missing_tables.append(table) print(f" ✗ Missing table: {table}") else: print(f" ✓ Table exists: {table}") if missing_tables: return self.log_result("Database Schema", False, f"Missing tables: {', '.join(missing_tables)}") # Check Grazuri-specific columns in loan_products cursor.execute("DESCRIBE loan_products") columns = [row[0] for row in cursor.fetchall()] required_columns = ['gl_code', 'grazuri_account_type'] missing_columns = [col for col in required_columns if col not in columns] if missing_columns: return self.log_result("Database Schema", False, f"Missing columns in loan_products: {', '.join(missing_columns)}") print(f"\n ✓ Grazuri columns present: {', '.join(required_columns)}") return self.log_result("Database Schema", True, "All required tables and columns present") except Exception as e: return self.log_result("Database Schema", False, f"Error: {str(e)}") def test_7_test_reports_compatibility(self): """Test 7: Test reports with new schema""" print("\n" + "=" * 80) print("TEST 7: Reports Compatibility") print("=" * 80) try: with connection.cursor() as cursor: # Test 1: Loan summary report print("\n Testing Loan Summary Report...") cursor.execute(""" SELECT COUNT(*) as total_loans, SUM(principal_amount) as total_principal, SUM(total_amount) as total_repayment, AVG(interest_amount) as avg_interest FROM loans WHERE is_deleted = 0 """) row = cursor.fetchone() if row: print(f" Total Loans: {row[0]}") print(f" Total Principal: KES {row[1]:,.2f}" if row[1] else " Total Principal: KES 0.00") print(f" Total Repayment: KES {row[2]:,.2f}" if row[2] else " Total Repayment: KES 0.00") print(f" Avg Interest: KES {row[3]:,.2f}" if row[3] else " Avg Interest: KES 0.00") # Test 2: Product performance report print("\n Testing Product Performance Report...") cursor.execute(""" SELECT lp.name, lp.product_type, lp.grazuri_account_type, COUNT(l.id) as loan_count, SUM(l.principal_amount) as total_disbursed FROM loan_products lp LEFT JOIN loan_applications la ON la.loan_product_id = lp.id LEFT JOIN loans l ON l.application_id = la.id WHERE lp.is_active = 1 GROUP BY lp.id, lp.name, lp.product_type, lp.grazuri_account_type """) products = cursor.fetchall() for name, ptype, gtype, count, total in products: print(f" {name} ({gtype}): {count} loans, KES {total:,.2f}" if total else f" {name} ({gtype}): {count} loans, KES 0.00") # Test 3: Branch performance report print("\n Testing Branch Performance Report...") cursor.execute(""" SELECT b.name, b.code, COUNT(DISTINCT u.id) as client_count, COUNT(DISTINCT l.id) as loan_count FROM branches b LEFT JOIN users u ON u.branch_id = b.id AND u.role = 'borrower' LEFT JOIN loans l ON l.borrower_id = u.id WHERE b.is_active = 1 GROUP BY b.id, b.name, b.code """) branches = cursor.fetchall() for name, code, clients, loans in branches: print(f" {name} ({code}): {clients} clients, {loans} loans") return self.log_result("Reports Compatibility", True, "All reports working") except Exception as e: return self.log_result("Reports Compatibility", False, f"Error: {str(e)}") def test_8_test_loan_calculations(self): """Test 8: Verify loan calculations are correct""" print("\n" + "=" * 80) print("TEST 8: Loan Calculations") print("=" * 80) try: loans = Loan.objects.all() if not loans: return self.log_result("Loan Calculations", True, "No loans to test (expected for fresh system)") all_correct = True for loan in loans: # Verify total = principal + interest + fee calculated_total = loan.principal_amount + loan.interest_amount + loan.processing_fee if abs(calculated_total - loan.total_amount) > Decimal('0.01'): print(f"\n ✗ Calculation error in {loan.loan_number}") print(f" Expected: {calculated_total}") print(f" Actual: {loan.total_amount}") all_correct = False else: print(f"\n ✓ {loan.loan_number}: Calculations correct") print(f" Principal: KES {loan.principal_amount:,.2f}") print(f" Interest: KES {loan.interest_amount:,.2f}") print(f" Fee: KES {loan.processing_fee:,.2f}") print(f" Total: KES {loan.total_amount:,.2f}") return self.log_result("Loan Calculations", all_correct, f"Verified {loans.count()} loan(s)") except Exception as e: return self.log_result("Loan Calculations", False, f"Error: {str(e)}") def cleanup_test_data(self): """Clean up test data""" print("\n" + "=" * 80) print("CLEANUP TEST DATA") print("=" * 80) print("\nTest data created:") print(f" Branches: {len(self.created_objects['branches'])}") print(f" Users: {len(self.created_objects['users'])}") print(f" Applications: {len(self.created_objects['applications'])}") print(f" Loans: {len(self.created_objects['loans'])}") response = input("\nDo you want to keep this test data? (y/n): ").strip().lower() if response == 'n': print("\nCleaning up test data...") # Delete in reverse order of creation for loan in self.created_objects['loans']: loan.delete() print(f" Deleted loan: {loan.loan_number}") for app in self.created_objects['applications']: app.delete() print(f" Deleted application: {app.application_number}") for user in self.created_objects['users']: user.delete() print(f" Deleted user: {user.username}") for branch in self.created_objects['branches']: branch.delete() print(f" Deleted branch: {branch.name}") print("\n✓ Test data cleaned up") else: print("\n✓ Test data kept for further testing") def generate_report(self): """Generate final test report""" print("\n" + "=" * 80) print("INTEGRATION TEST REPORT") print("=" * 80) total = len(self.test_results) passed = sum(1 for _, p, _ in self.test_results if p) failed = total - passed print(f"\nTotal Tests: {total}") print(f"Passed: {passed}") print(f"Failed: {failed}") if failed > 0: print("\nFailed Tests:") for name, passed, message in self.test_results: if not passed: print(f" ✗ {name}: {message}") print("\n" + "=" * 80) if failed == 0: print("✅ ALL INTEGRATION TESTS PASSED!") print("The system is fully compatible with Grazuri migration") else: print(f"⚠ {failed} TEST(S) FAILED") print("Please review the errors above") print("=" * 80) return failed == 0 def run_all_tests(self): """Run all integration tests""" print("\n" + "=" * 80) print("GRAZURI MIGRATION - INTEGRATION TESTING") print("Haven Grazuri Investment Limited") print("=" * 80) try: with transaction.atomic(): # Run all tests self.test_1_check_branches() self.test_2_check_loan_products() self.test_3_create_test_clients() self.test_4_create_loan_applications() self.test_5_approve_and_disburse_loans() self.test_6_check_database_schema() self.test_7_test_reports_compatibility() self.test_8_test_loan_calculations() # Generate report all_passed = self.generate_report() # Ask about cleanup self.cleanup_test_data() return 0 if all_passed else 1 except Exception as e: print(f"\n✗ CRITICAL ERROR: {str(e)}") import traceback traceback.print_exc() return 1 def main(): """Main execution""" tester = IntegrationTester() return tester.run_all_tests() if __name__ == '__main__': sys.exit(main())