"""Unit and integration tests for UNC classification, Hot Watcher routing, and document upload integration.""" from __future__ import annotations import io import os import shutil import tempfile import unittest from pathlib import Path from unittest.mock import MagicMock, patch from fastapi.testclient import TestClient from app.main import create_app from app.modules.auth.presentation.auth import get_current_user from app.modules.banking.application.unc_classifier_and_router import ( BANK_FOLDER_MAP, HOT_WATCHER_ROOT, UncClassifierAndRouter, ) from app.modules.banking.application.unc_hot_watcher import UncHotWatcherService from app.modules.core.application.document_processing_service import ( DocumentProcessingService, ) class TestUncUploadAndHotWatcher(unittest.TestCase): def setUp(self) -> None: self.temp_dir = tempfile.mkdtemp() self.mock_client = MagicMock() def tearDown(self) -> None: if os.path.exists(self.temp_dir): shutil.rmtree(self.temp_dir, ignore_errors=True) def test_unc_detection_and_bank_classification(self) -> None: """Test rule-based classification for banks and document types.""" # 1. ACB detection is_unc, bank, det = UncClassifierAndRouter.detect_unc_and_bank( filename="13456888888_SAOKE_TK_202601.xlsx" ) self.assertTrue(is_unc) self.assertEqual(bank, "ACB") # 2. Techcombank detection is_unc, bank, det = UncClassifierAndRouter.detect_unc_and_bank( filename="ACCOUNT_STATEMENT_07898888_202603.pdf" ) self.assertTrue(is_unc) self.assertEqual(bank, "TECHCOMBANK") # 3. VPBank detection is_unc, bank, det = UncClassifierAndRouter.detect_unc_and_bank( filename="AccountStmt_20260508183117.xls" ) self.assertTrue(is_unc) self.assertEqual(bank, "VPBANK") is_unc, bank, det = UncClassifierAndRouter.detect_unc_and_bank( filename="Saoketaikhoan_B2B_email_8667898888_20260401000000.xls" ) self.assertTrue(is_unc) self.assertEqual(bank, "VPBANK") # 4. Generic UNC detection is_unc, bank, det = UncClassifierAndRouter.detect_unc_and_bank( filename="UNC_Thanh_Toan_Vat_Lieu.pdf" ) self.assertTrue(is_unc) self.assertEqual(bank, "OTHER") # 5. Content-based detection is_unc, bank, det = UncClassifierAndRouter.detect_unc_and_bank( filename="ChungTu_ChuyenKhoan.pdf", text="ỦY NHIỆM CHI NGÂN HÀNG NGOẠI THƯƠNG VIỆT NAM VCB", ) self.assertTrue(is_unc) self.assertEqual(bank, "VIETCOMBANK") # 6. Non-UNC documents is_unc, bank, det = UncClassifierAndRouter.detect_unc_and_bank( filename="Hop_Dong_Thi_Cong_Co_Gioi.pdf", text="Cộng hòa Xã hội Chủ nghĩa Việt Nam. Hợp đồng kinh tế xây lắp.", ) self.assertFalse(is_unc) self.assertIsNone(bank) def test_route_to_hot_folder(self) -> None: """Test file routing into proper Hot Watcher folder with duplicate prevention.""" with patch( "app.modules.banking.application.unc_classifier_and_router.HOT_WATCHER_ROOT", self.temp_dir, ): file_bytes = b"Sample Bank Statement Content 2026" target_dir, target_file = UncClassifierAndRouter.route_to_hot_folder( file_content=file_bytes, filename="Saoke_ACB_2026.xlsx", bank_code="ACB", ) self.assertTrue(os.path.exists(target_file)) self.assertEqual(os.path.basename(target_dir), "ACB-2026") with open(target_file, "rb") as f: self.assertEqual(f.read(), file_bytes) # Test duplicate routing with same content (idempotency) target_dir2, target_file2 = UncClassifierAndRouter.route_to_hot_folder( file_content=file_bytes, filename="Saoke_ACB_2026.xlsx", bank_code="ACB", ) self.assertEqual(target_file, target_file2) @patch("app.core.postgres.erp_client.ErpDatabaseClient.check_duplicate_document") @patch("app.core.postgres.erp_client.ErpDatabaseClient.create_document") @patch("app.core.postgres.erp_client.ErpDatabaseClient.update_document") @patch("app.core.postgres.erp_client.ErpDatabaseClient.get_connection") @patch( "app.modules.core.application.document_processing_service.DocumentProcessingService.save_file" ) @patch( "app.modules.banking.application.unc_classifier_and_router.UncClassifierAndRouter.ingest_statement_file" ) def test_api_upload_unc_document_routing( self, mock_ingest: MagicMock, mock_save: MagicMock, mock_get_conn: MagicMock, mock_update_doc: MagicMock, mock_create_doc: MagicMock, mock_check_dup: MagicMock, ) -> None: """Test /v1/erp/documents/upload when an UNC is uploaded.""" mock_conn = MagicMock() mock_cur = MagicMock() mock_cur.fetchone.side_effect = [ {"id": "company-uuid-1"}, # for select company {"id": "project-uuid-1"}, # for select project ] mock_conn.cursor.return_value.__enter__.return_value = mock_cur mock_get_conn.return_value.__enter__.return_value = mock_conn mock_save.return_value = ( "/static/uploads/13456888888_SAOKE_TK_202601.xlsx", os.path.join(self.temp_dir, "13456888888_SAOKE_TK_202601.xlsx"), 2048, "xlsx", "mock_hash_acb", ) mock_check_dup.return_value = None mock_create_doc.return_value = { "id": "doc-uuid-unc-1", "document_code": "TEMP-UNC01", "document_title": "13456888888_SAOKE_TK_202601", "document_type": "UNC", "document_group": "financial", } mock_update_doc.return_value = { "id": "doc-uuid-unc-1", "status": "processing", } mock_ingest.return_value = { "status": "success", "total_parsed": 5, "newly_saved": 5, "inserted": 5, "skipped": 0, } app = create_app() app.dependency_overrides[get_current_user] = lambda: { "id": "user-1", "email": "accountant@dscons.vn", "role": "accountant", } client = TestClient(app) with patch( "app.modules.banking.application.unc_classifier_and_router.HOT_WATCHER_ROOT", self.temp_dir, ): with patch("fastapi.BackgroundTasks.add_task") as mock_add_task: response = client.post( "/v1/erp/documents/upload", files={ "file": ( "13456888888_SAOKE_TK_202601.xlsx", b"Excel binary mock data for ACB", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"project_code": "PRJ-2026"}, ) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["id"], "doc-uuid-unc-1") self.assertIn("unc_routing", data) self.assertIsNotNone(data["unc_routing"]) self.assertTrue(data["unc_routing"]["is_unc"]) self.assertEqual(data["unc_routing"]["bank_code"], "ACB") self.assertEqual(data["unc_routing"]["transactions_inserted"], 5) # Verify create_document received document_type='UNC' and document_group='financial' create_payload = mock_create_doc.call_args[0][0] self.assertEqual(create_payload["document_type"], "UNC") self.assertEqual(create_payload["document_group"], "financial") self.assertEqual(create_payload["category"], "financial") @patch("app.core.postgres.erp_client.ErpDatabaseClient.check_duplicate_document") @patch("app.core.postgres.erp_client.ErpDatabaseClient.create_document") @patch("app.core.postgres.erp_client.ErpDatabaseClient.update_document") @patch("app.core.postgres.erp_client.ErpDatabaseClient.get_connection") @patch( "app.modules.core.application.document_processing_service.DocumentProcessingService.save_file" ) def test_api_upload_non_unc_document( self, mock_save: MagicMock, mock_get_conn: MagicMock, mock_update_doc: MagicMock, mock_create_doc: MagicMock, mock_check_dup: MagicMock, ) -> None: """Test /v1/erp/documents/upload when a regular legal document is uploaded.""" mock_conn = MagicMock() mock_cur = MagicMock() mock_cur.fetchone.side_effect = [ {"id": "company-uuid-1"}, None, ] mock_conn.cursor.return_value.__enter__.return_value = mock_cur mock_get_conn.return_value.__enter__.return_value = mock_conn mock_save.return_value = ( "/static/uploads/Bien_Ban_Hien_Truong.pdf", os.path.join(self.temp_dir, "Bien_Ban_Hien_Truong.pdf"), 1024, "pdf", "mock_hash_bb", ) mock_check_dup.return_value = None mock_create_doc.return_value = {"id": "doc-uuid-regular"} mock_update_doc.return_value = { "id": "doc-uuid-regular", "status": "processing", } app = create_app() app.dependency_overrides[get_current_user] = lambda: { "id": "user-1", "email": "engineer@dscons.vn", "role": "site_engineer", } client = TestClient(app) with patch("fastapi.BackgroundTasks.add_task"): response = client.post( "/v1/erp/documents/upload", files={ "file": ( "Bien_Ban_Hien_Truong.pdf", b"PDF Content of site inspection", "application/pdf", ) }, ) self.assertEqual(response.status_code, 200) data = response.json() self.assertIsNone(data.get("unc_routing")) create_payload = mock_create_doc.call_args[0][0] self.assertEqual(create_payload["document_type"], "CV") self.assertEqual(create_payload["document_group"], "Văn Bản Hành Chính") def test_hot_watcher_status_and_scan_endpoints(self) -> None: """Test Hot Watcher status and scan API endpoints.""" app = create_app() app.dependency_overrides[get_current_user] = lambda: { "id": "user-1", "email": "superadmin@dscons.vn", "role": "superadmin", } client = TestClient(app) # 1. Test status endpoint status_res = client.get("/v1/erp/banking/hot-watcher/status") self.assertEqual(status_res.status_code, 200) data = status_res.json() self.assertIn("hot_watcher_root", data) self.assertIn("subfolders", data) self.assertIn("ACB", data["subfolders"]) self.assertIn("TECHCOMBANK", data["subfolders"]) self.assertIn("VPBANK", data["subfolders"]) # 2. Test scan endpoint with mocked service with patch.object( UncHotWatcherService, "scan_all_folders" ) as mock_scan: mock_scan.return_value = { "status": "success", "files_scanned": 12, "total_parsed": 84, "newly_saved": 0, "details": [], } scan_res = client.post("/v1/erp/banking/hot-watcher/scan") self.assertEqual(scan_res.status_code, 200) scan_data = scan_res.json() self.assertEqual(scan_data["status"], "success") self.assertEqual(scan_data["files_scanned"], 12) self.assertEqual(scan_data["total_parsed"], 84) @patch("app.modules.core.application.document_processing_service.DocumentProcessingService.extract_text") @patch("app.modules.core.application.document_processing_service.DocumentProcessingService.analyze_with_ai") @patch("app.modules.core.application.document_processing_service.DocumentProcessingService.validate_deterministic_rules") @patch("app.modules.core.application.document_processing_service.DocumentProcessingService.run_cross_agent_audit") @patch("app.core.postgres.erp_client.ErpDatabaseClient.get_document") @patch("app.core.postgres.erp_client.ErpDatabaseClient.update_document") @patch("app.modules.banking.application.unc_classifier_and_router.UncClassifierAndRouter.ingest_statement_file") def test_pipeline_fallback_unc_detection( self, mock_ingest: MagicMock, mock_update: MagicMock, mock_get_doc: MagicMock, mock_cross: MagicMock, mock_rules: MagicMock, mock_analyze: MagicMock, mock_extract: MagicMock, ) -> None: """Test background pipeline detecting UNC from text content and routing to Hot Watcher.""" import asyncio test_file = os.path.join(self.temp_dir, "scan_tai_lieu_001.pdf") with open(test_file, "wb") as f: f.write(b"PDF fake content") mock_extract.return_value = "ỦY NHIỆM CHI NGÂN HÀNG THƯƠNG MẠI CỔ PHẦN Á CHÂU ACB SỐ TIỀN 50.000.000" mock_analyze.return_value = { "document_code": "UNC-ACB-01", "document_title": "Ủy nhiệm chi ACB chuyển tiền vật liệu", "document_group": "financial", "document_type": "UNC", "signature_status": "fully_executed", } mock_rules.return_value = [] mock_cross.return_value = {"confidence_score": 90, "flags": []} mock_get_doc.return_value = {"id": "doc-scan-1", "document_code": "TEMP-001"} mock_ingest.return_value = {"inserted": 1, "status": "success"} with patch("app.modules.banking.application.unc_classifier_and_router.HOT_WATCHER_ROOT", self.temp_dir): service = DocumentProcessingService() asyncio.run(service.run_pipeline("doc-scan-1", test_file, "scan_tai_lieu_001.pdf", "pdf")) # Verify document updated with financial / UNC mock_update.assert_called_once() call_payload = mock_update.call_args[0][1] self.assertEqual(call_payload["document_group"], "financial") self.assertEqual(call_payload["document_type"], "UNC") # Verify file was routed to ACB-2026 folder acb_folder = os.path.join(self.temp_dir, "ACB-2026") self.assertTrue(os.path.exists(acb_folder)) self.assertTrue(os.path.exists(os.path.join(acb_folder, "scan_tai_lieu_001.pdf"))) if __name__ == "__main__": unittest.main()