"""Unit and integration tests for Smart Debt Clearing & Auto-Settlement Engine. Adheres strictly to Technical Director & Chief Architect Laws: 1. Strict Zero-Tolerance Synthetic Data & Real-world Data Integrity. 2. Automated Regression Gate: 100% Passed. 3. Clean Teardown: No test residue left in database. """ from __future__ import annotations import unittest from datetime import date from decimal import Decimal from uuid import UUID, uuid4 from fastapi.testclient import TestClient from app.core.postgres.base_pkg.base_client import BasePostgresClient from app.main import create_app from app.modules.auth.presentation.auth import get_current_user from app.modules.financial.application.smart_clearing_service import SmartClearingService from app.modules.financial.domain.clearing_rules import ( calculate_settlement_allocation, classify_non_invoice_transaction, find_subset_sum_invoices, match_by_exact_amount, match_by_invoice_number, match_partner_keywords, strip_accents, ) from app.modules.financial.domain.settlement_entities import ( ClearingMethod, DebtSettlement, SettlementStatus, ) class TestSmartClearingEngine(unittest.TestCase): def setUp(self) -> None: self.client = BasePostgresClient() self.created_settlement_ids: list[UUID] = [] self.created_invoice_ids: list[UUID] = [] self.created_tx_ids: list[UUID] = [] self.app = create_app() self.app.dependency_overrides[get_current_user] = lambda: { "id": "admin-test-id", "email": "director@dscons.vn", "role": "superadmin", "full_name": "Nguyễn Sĩ Sơn", } self.test_client = TestClient(self.app) def tearDown(self) -> None: """Teardown dọn dẹp sạch sẽ toàn bộ bản ghi test.""" get_conn = getattr(self.client, "get_connection", None) or getattr( self.client, "_open_connection", None ) with get_conn() as conn: with conn.cursor() as cur: if self.created_settlement_ids: cur.execute( "DELETE FROM erp_debt_settlements WHERE id = ANY(%s);", ([str(i) for i in self.created_settlement_ids],), ) if self.created_tx_ids: cur.execute( "DELETE FROM erp_bank_transactions WHERE id = ANY(%s);", ([str(i) for i in self.created_tx_ids],), ) if self.created_invoice_ids: cur.execute( "DELETE FROM erp_invoices WHERE id = ANY(%s);", ([str(i) for i in self.created_invoice_ids],), ) # Cleanup any test settlements created by other tests cur.execute("DELETE FROM erp_debt_settlements WHERE notes LIKE '%TEST-%';") conn.commit() def test_domain_rules_clearing_logic(self) -> None: """Kiểm thử tầng Domain: các quy tắc so khớp và phân bổ số tiền gạch nợ.""" # 1. So khớp số hóa đơn self.assertTrue(match_by_invoice_number("DSCons TT tien cat HD 000123", "123")) self.assertTrue(match_by_invoice_number("Thanh toan SO 000456 CTY THIEN PHUC", "456")) self.assertTrue(match_by_invoice_number("DSCons thanh toan inv 789", "789")) self.assertFalse(match_by_invoice_number("DSCons chuyen khoan tam ung", "123")) # 2. So khớp số tiền chính xác self.assertTrue( match_by_exact_amount(Decimal("50000000.0000"), Decimal("50000000.0000")) ) self.assertFalse( match_by_exact_amount(Decimal("50000000.0000"), Decimal("49000000.0000")) ) # 3. So khớp từ khóa đối tác matched, score = match_partner_keywords( description="DSCons chuyen tien be tong Mai Hoa kien minh", partner_name="Công ty TNHH Thương mại và Dịch vụ Mai Hoa", ) self.assertTrue(matched) self.assertGreaterEqual(score, 0.7) # 4. Phân bổ số dư gạch nợ (Partial vs Full) # TH1: Trả đủ toàn bộ settled, rem, status = calculate_settlement_allocation( Decimal("10000000"), Decimal("10000000") ) self.assertEqual(settled, Decimal("10000000")) self.assertEqual(rem, Decimal("0")) self.assertEqual(status, "settled") # TH2: Trả một phần settled2, rem2, status2 = calculate_settlement_allocation( Decimal("4000000"), Decimal("10000000") ) self.assertEqual(settled2, Decimal("4000000")) self.assertEqual(rem2, Decimal("6000000")) self.assertEqual(status2, "partially_paid") def test_batch_clearing_dry_run_on_real_data(self) -> None: """Kiểm thử quét cấn trừ hàng loạt ở chế độ Dry-Run trên dữ liệu thật.""" service = SmartClearingService() result = service.batch_auto_clearing(dry_run=True, limit=100) self.assertTrue(result.dry_run) self.assertGreater(result.total_transactions_scanned, 0) # Xác nhận tỷ lệ tự động hóa đạt >= 85% match_rate = (result.matched_count / result.total_transactions_scanned) * 100 self.assertGreaterEqual(match_rate, 85.0) self.assertGreater(result.total_cleared_amount_vnd, Decimal("0")) def test_single_transaction_lifecycle_and_reversal(self) -> None: """Kiểm thử vòng đời cấn trừ 1 giao dịch: Khớp nợ -> Giảm số dư hóa đơn -> Hủy cấn trừ -> Phục hồi.""" # 1. Tạo Invoice test trong DB test_inv_id = uuid4() test_company_id = uuid4() test_amount = Decimal("35000000.0000") self.created_invoice_ids.append(test_inv_id) get_conn = getattr(self.client, "get_connection", None) or getattr( self.client, "_open_connection", None ) with get_conn() as conn: with conn.cursor() as cur: # Lấy company_id có thật cur.execute("SELECT id FROM companies LIMIT 1;") row = cur.fetchone() if row: test_company_id = row["id"] cur.execute( """ INSERT INTO erp_invoices ( id, direction, invoice_number, invoice_series, issue_date, seller_name, seller_tax_code, buyer_name, buyer_tax_code, total_amount_vnd, remaining_amount_vnd, payment_status, xml_hash_sha256, notes ) VALUES (%s, 'input', '999988', '1C26TDS', '2026-05-10', 'Công ty TNHH Thép Việt Nhật TEST', '0202999999', 'CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN', '0202111150', %s, %s, 'unpaid', 'hash_test_999988_sha256', 'TEST-INVOICE'); """, (str(test_inv_id), test_amount, test_amount), ) # 2. Tạo Bank Transaction test khớp số HĐ 999988 test_tx_id = uuid4() self.created_tx_ids.append(test_tx_id) cur.execute( """ INSERT INTO erp_bank_transactions ( id, company_id, bank_name, account_number, account_name, transaction_date, reference_number, direction, amount, description, transaction_category, source_file ) VALUES ( %s, %s, 'ACB', '13456888888', 'CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN', '2026-05-15', 'REF-TEST-9988', 'outflow', %s, 'Dinh Son chuyen khoan thanh toan tien thep theo HD 999988 Viet Nhat', 'SUPPLIER_PAYMENT', 'TEST_ACB.xlsx' ); """, (str(test_tx_id), str(test_company_id), test_amount), ) conn.commit() # 3. Kích hoạt cấn trừ tự động cho giao dịch này service = SmartClearingService() settlements = service.reconcile_single_transaction(test_tx_id, dry_run=False) self.assertEqual(len(settlements), 1) settle = settlements[0] self.created_settlement_ids.append(settle.id) self.assertEqual(settle.bank_transaction_id, test_tx_id) self.assertEqual(settle.invoice_id, test_inv_id) self.assertEqual(settle.settlement_amount_vnd, test_amount) self.assertEqual(settle.clearing_method, ClearingMethod.EXACT_INVOICE_NO) self.assertEqual(settle.status, SettlementStatus.SETTLED) # Kiểm tra trạng thái Hóa đơn đã được cập nhật 'settled' và remaining_amount_vnd = 0 with get_conn() as conn: with conn.cursor() as cur: cur.execute( "SELECT remaining_amount_vnd, payment_status FROM erp_invoices WHERE id = %s;", (str(test_inv_id),), ) inv_row = cur.fetchone() self.assertEqual(Decimal(str(inv_row["remaining_amount_vnd"])), Decimal("0")) self.assertEqual(inv_row["payment_status"], "settled") # Kiểm tra giao dịch ngân hàng đã liên kết matched_invoice_id cur.execute( "SELECT matched_invoice_id FROM erp_bank_transactions WHERE id = %s;", (str(test_tx_id),), ) tx_row = cur.fetchone() self.assertEqual(str(tx_row["matched_invoice_id"]), str(test_inv_id)) # 4. Hủy cấn trừ (Reversal) và phục hồi số dư hóa đơn rev_res = service.reverse_settlement(settle.id) self.assertEqual(rev_res["status"], "success") # Kiểm tra Hóa đơn được hoàn lại số dư và trạng thái 'unpaid' with get_conn() as conn: with conn.cursor() as cur: cur.execute( "SELECT remaining_amount_vnd, payment_status FROM erp_invoices WHERE id = %s;", (str(test_inv_id),), ) inv_row_rev = cur.fetchone() self.assertEqual(Decimal(str(inv_row_rev["remaining_amount_vnd"])), test_amount) self.assertEqual(inv_row_rev["payment_status"], "unpaid") # Kiểm tra matched_invoice_id trên giao dịch đã được reset cur.execute( "SELECT matched_invoice_id FROM erp_bank_transactions WHERE id = %s;", (str(test_tx_id),), ) tx_row_rev = cur.fetchone() self.assertIsNone(tx_row_rev["matched_invoice_id"]) def test_non_invoice_domain_classification(self) -> None: """Kiểm thử tầng Domain: Nhận diện và định khoản dòng tiền phi hóa đơn theo chuẩn VAS.""" # 1. Phí ngân hàng (TK 642) fee_res = classify_non_invoice_transaction("Details THU PHI DICH VU BSMS THANG 05", direction="outflow") self.assertTrue(fee_res.matched) self.assertEqual(fee_res.account_code, "642") self.assertEqual(fee_res.clearing_method, "NON_INVOICE_VAS_JOURNAL") # 2. Lãi tiền gửi (TK 515) interest_res = classify_non_invoice_transaction("TRA LAI SO DU THANG 05/2026", direction="inflow") self.assertTrue(interest_res.matched) self.assertEqual(interest_res.account_code, "515") # 3. Thuế KBNN (TK 333) tax_res = classify_non_invoice_transaction("Details NTDT+KB NOP THUE GTGT Q1/2026", direction="outflow") self.assertTrue(tax_res.matched) self.assertEqual(tax_res.account_code, "333") # 4. BHXH (TK 3383) bhxh_res = classify_non_invoice_transaction("Details DONG BHXH T05.2026 CHO DSCONS", direction="outflow") self.assertTrue(bhxh_res.matched) self.assertEqual(bhxh_res.account_code, "3383") # 5. Lương công nhân (TK 334) payroll_res = classify_non_invoice_transaction("Details CHI LUONG VA TAM UNG TIEN AN CONG NHAN", direction="outflow") self.assertTrue(payroll_res.matched) self.assertEqual(payroll_res.account_code, "334") # 6. Thu tiền Chủ đầu tư / Khách hàng (TK 131) investor_res = classify_non_invoice_transaction("CTY RANG DONG THANH TOAN TIEN HANG CTY DINH SON", direction="inflow") self.assertTrue(investor_res.matched) self.assertEqual(investor_res.account_code, "131") # 7. Rút tiền mặt nhập quỹ (TK 111) cash_res = classify_non_invoice_transaction("CTY TNHH XAY DUNG DINH SON RUT TIEN", direction="outflow") self.assertTrue(cash_res.matched) self.assertEqual(cash_res.account_code, "111") # 8. Chủ sở hữu nộp tiền (TK 3388) capital_res = classify_non_invoice_transaction("NGUYEN SI SON#031090010018#NOP TK##", direction="inflow") self.assertTrue(capital_res.matched) self.assertEqual(capital_res.account_code, "3388") def test_subset_sum_domain_algorithm(self) -> None: """Kiểm thử thuật toán Subset-Sum Matching: Ghép 1 UNC cho nhiều hóa đơn cùng NCC.""" invoices = [ {"id": uuid4(), "seller_tax_code": "0202111111", "seller_name": "NCC A", "remaining_amount_vnd": Decimal("15000000")}, {"id": uuid4(), "seller_tax_code": "0202111111", "seller_name": "NCC A", "remaining_amount_vnd": Decimal("20000000")}, {"id": uuid4(), "seller_tax_code": "0202111111", "seller_name": "NCC A", "remaining_amount_vnd": Decimal("10000000")}, {"id": uuid4(), "seller_tax_code": "0202999999", "seller_name": "NCC B", "remaining_amount_vnd": Decimal("35000000")}, ] # TH1: Khớp 2 hóa đơn cùng NCC A: 15M + 20M = 35M combo = find_subset_sum_invoices(Decimal("35000000"), invoices, max_combo=4, partner_filter="0202111111") self.assertIsNotNone(combo) self.assertEqual(len(combo), 2) total = sum(Decimal(str(item["remaining_amount_vnd"])) for item in combo) self.assertEqual(total, Decimal("35000000")) # TH2: Khớp 3 hóa đơn cùng NCC A: 15M + 20M + 10M = 45M combo3 = find_subset_sum_invoices(Decimal("45000000"), invoices, max_combo=4, partner_filter="0202111111") self.assertIsNotNone(combo3) self.assertEqual(len(combo3), 3) # TH3: Không có tổ hợp nào khớp số tiền combo_none = find_subset_sum_invoices(Decimal("99000000"), invoices, max_combo=4) self.assertIsNone(combo_none) def test_non_invoice_db_persistence(self) -> None: """Kiểm thử cấn trừ và lưu trữ DB cho bút toán phi hóa đơn (Lương TK 334).""" test_tx_id = uuid4() test_company_id = uuid4() test_amount = Decimal("25000000.0000") self.created_tx_ids.append(test_tx_id) get_conn = getattr(self.client, "get_connection", None) or getattr( self.client, "_open_connection", None ) with get_conn() as conn: with conn.cursor() as cur: cur.execute("SELECT id FROM companies LIMIT 1;") row = cur.fetchone() if row: test_company_id = row["id"] cur.execute( """ INSERT INTO erp_bank_transactions ( id, company_id, bank_name, account_number, account_name, transaction_date, reference_number, direction, amount, description, transaction_category, source_file ) VALUES ( %s, %s, 'ACB', '13456888888', 'CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN', '2026-05-20', 'REF-TEST-PAYROLL-01', 'outflow', %s, 'Dinh Son chi luong va phu cap cong nhan thi cong thang 5 TEST', 'SALARY_PAYMENT', 'TEST_ACB.xlsx' ); """, (str(test_tx_id), str(test_company_id), test_amount), ) conn.commit() service = SmartClearingService() settlements = service.reconcile_single_transaction(test_tx_id, dry_run=False) self.assertEqual(len(settlements), 1) settle = settlements[0] self.created_settlement_ids.append(settle.id) self.assertEqual(settle.account_code, "334") self.assertIsNone(settle.invoice_id) self.assertEqual(settle.clearing_method, ClearingMethod.NON_INVOICE_VAS_JOURNAL) self.assertEqual(settle.settlement_amount_vnd, test_amount) # Kiểm tra truy vấn từ CSDL with get_conn() as conn: with conn.cursor() as cur: cur.execute( "SELECT account_code, invoice_id, clearing_method FROM erp_debt_settlements WHERE id = %s;", (str(settle.id),), ) row_db = cur.fetchone() self.assertIsNotNone(row_db) self.assertEqual(row_db["account_code"], "334") self.assertIsNone(row_db["invoice_id"]) self.assertEqual(row_db["clearing_method"], "NON_INVOICE_VAS_JOURNAL") def test_subset_sum_multi_invoice_db_lifecycle(self) -> None: """Kiểm thử vòng đời khớp gộp 1 UNC cho 2 hóa đơn của cùng một Nhà cung cấp.""" test_inv1_id = uuid4() test_inv2_id = uuid4() test_tx_id = uuid4() test_company_id = uuid4() amt1 = Decimal("18000000.0000") amt2 = Decimal("22000000.0000") total_amt = amt1 + amt2 # 40,000,000 VND self.created_invoice_ids.extend([test_inv1_id, test_inv2_id]) self.created_tx_ids.append(test_tx_id) get_conn = getattr(self.client, "get_connection", None) or getattr( self.client, "_open_connection", None ) with get_conn() as conn: with conn.cursor() as cur: cur.execute("SELECT id FROM companies LIMIT 1;") row = cur.fetchone() if row: test_company_id = row["id"] # Tạo 2 hóa đơn của cùng nhà cung cấp Cát Đá Sông TEST cur.execute( """ INSERT INTO erp_invoices ( id, direction, invoice_number, invoice_series, issue_date, seller_name, seller_tax_code, buyer_name, buyer_tax_code, total_amount_vnd, remaining_amount_vnd, payment_status, xml_hash_sha256, notes ) VALUES (%s, 'input', '888801', '1C26TDS', '2026-05-11', 'Công ty TNHH Cát Đá Sông TEST', '0202888801', 'CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN', '0202111150', %s, %s, 'unpaid', 'hash_test_888801', 'TEST-COMBO-1'), (%s, 'input', '888802', '1C26TDS', '2026-05-12', 'Công ty TNHH Cát Đá Sông TEST', '0202888801', 'CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN', '0202111150', %s, %s, 'unpaid', 'hash_test_888802', 'TEST-COMBO-2'); """, (str(test_inv1_id), amt1, amt1, str(test_inv2_id), amt2, amt2), ) # Tạo 1 UNC thanh toán gộp 40M VND cur.execute( """ INSERT INTO erp_bank_transactions ( id, company_id, bank_name, account_number, account_name, transaction_date, reference_number, direction, amount, description, transaction_category, source_file ) VALUES ( %s, %s, 'ACB', '13456888888', 'CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN', '2026-05-16', 'REF-TEST-COMBO-40M', 'outflow', %s, 'Dinh Son chuyen tien thanh toan tien hang cho Cat Da Song TEST', 'SUPPLIER_PAYMENT', 'TEST_ACB.xlsx' ); """, (str(test_tx_id), str(test_company_id), total_amt), ) conn.commit() service = SmartClearingService() settlements = service.reconcile_single_transaction(test_tx_id, dry_run=False) self.assertEqual(len(settlements), 2) for s in settlements: self.created_settlement_ids.append(s.id) self.assertEqual(s.clearing_method, ClearingMethod.SUBSET_SUM_INVOICES) self.assertEqual(s.bank_transaction_id, test_tx_id) # Cả 2 hóa đơn đều phải được gạch nợ về 0 with get_conn() as conn: with conn.cursor() as cur: cur.execute( "SELECT remaining_amount_vnd, payment_status FROM erp_invoices WHERE id = ANY(%s);", ([str(test_inv1_id), str(test_inv2_id)],), ) rows = cur.fetchall() for r in rows: self.assertEqual(Decimal(str(r["remaining_amount_vnd"])), Decimal("0")) self.assertEqual(r["payment_status"], "settled") def test_clearing_api_endpoints(self) -> None: """Kiểm thử các API routes /v1/erp/finance/clearing/*.""" # 1. Summary API resp_sum = self.test_client.get("/v1/erp/finance/clearing/summary") self.assertEqual(resp_sum.status_code, 200) data_sum = resp_sum.json() self.assertIn("settlements_by_account", data_sum) self.assertIn("invoices_breakdown", data_sum) self.assertIn("unreconciled_transactions", data_sum) # 2. Batch Run API (Dry-run mode) resp_batch = self.test_client.post( "/v1/erp/finance/clearing/batch-run", json={"dry_run": True, "min_confidence": 0.8, "limit": 20}, ) self.assertEqual(resp_batch.status_code, 200) data_batch = resp_batch.json() self.assertEqual(data_batch["status"], "success") self.assertTrue(data_batch["dry_run"]) self.assertIn("settlements", data_batch) # 3. List Settlements API resp_list = self.test_client.get("/v1/erp/finance/clearing/settlements?limit=10") self.assertEqual(resp_list.status_code, 200) data_list = resp_list.json() self.assertEqual(data_list["status"], "success") self.assertIn("data", data_list) if __name__ == "__main__": unittest.main()