from datetime import datetime from unittest.mock import AsyncMock, MagicMock, patch import pytest from fastapi.testclient import TestClient from app.main import create_app from app.modules.auth.application.auth_service import AuthService from app.modules.invoices.application.invoice_processing.auto_sync_scheduler import ( InvoiceAutoSyncSchedulerService, ) @pytest.fixture def auth_headers(): auth = AuthService() token = auth.create_jwt_token( { "id": "acc-test-id", "sub": "acc-test-id", "email": "accountant_test@dinhsonconstruction.com", "full_name": "Kế Toán Trưởng Định Sơn", "role": "accountant", "company_id": "0202111150", } ) return {"Authorization": f"Bearer {token}"} class TestInvoiceAutoSyncScheduler: def test_calculate_next_run_all_slots(self): service = InvoiceAutoSyncSchedulerService() assert service.scheduled_hours == [1, 13] assert service.jitter_minutes == 30 assert service.enable_jitter is True # Khi không dùng jitter (chế độ kiểm thử xác định) # 00:30 -> 01:00 cùng ngày t0 = datetime(2026, 8, 29, 0, 30, 0) next_run, delay = service.calculate_next_run(t0, with_jitter=False) assert next_run == datetime(2026, 8, 29, 1, 0, 0) assert delay == 1800.0 # 01:00:01 -> 13:00 cùng ngày t1 = datetime(2026, 8, 29, 1, 0, 1) next_run, delay = service.calculate_next_run(t1, with_jitter=False) assert next_run == datetime(2026, 8, 29, 13, 0, 0) assert delay == (datetime(2026, 8, 29, 13, 0, 0) - t1).total_seconds() # 13:00:01 -> 01:00 ngày hôm sau t2 = datetime(2026, 8, 29, 13, 0, 1) next_run, delay = service.calculate_next_run(t2, with_jitter=False) assert next_run == datetime(2026, 8, 30, 1, 0, 0) assert delay == (datetime(2026, 8, 30, 1, 0, 0) - t2).total_seconds() def test_calculate_next_run_with_jitter_range(self): service = InvoiceAutoSyncSchedulerService(jitter_minutes=30) # Thời điểm trước cửa sổ 00:30: t = 00:00 t0 = datetime(2026, 8, 29, 0, 0, 0) runs = [] for _ in range(50): target, delay = service.calculate_next_run(t0, with_jitter=True) runs.append(target) # Phải nằm trong khoảng 01:00 ± 30 phút: từ 00:30 đến 01:30 assert datetime(2026, 8, 29, 0, 30, 0) <= target <= datetime(2026, 8, 29, 1, 30, 0) assert delay > 0 # Chứng minh tính ngẫu nhiên: min và max khác nhau assert min(runs) < max(runs) def test_calculate_next_run_inside_window(self): service = InvoiceAutoSyncSchedulerService(jitter_minutes=30) # Thời điểm đã nằm trong cửa sổ: t = 00:45 (cửa sổ 00:30 - 01:30) t_inside = datetime(2026, 8, 29, 0, 45, 0) target, delay = service.calculate_next_run(t_inside, with_jitter=True) # Target phải nằm từ sau t_inside đến 01:30 assert t_inside < target <= datetime(2026, 8, 29, 1, 30, 0) assert delay > 0 def test_calculate_next_run_advances_slots_with_update_state(self): service = InvoiceAutoSyncSchedulerService(jitter_minutes=30) # Lập lịch chạy cho slot 01:00 t1 = datetime(2026, 8, 29, 0, 0, 0) target1, delay1 = service.calculate_next_run(t1, with_jitter=True, update_state=True) assert datetime(2026, 8, 29, 0, 30, 0) <= target1 <= datetime(2026, 8, 29, 1, 30, 0) # Sau khi chạy xong lúc 01:10 (vẫn trong khoảng trước 01:30), slot kế tiếp phải là 13:00 t2 = datetime(2026, 8, 29, 1, 10, 0) target2, delay2 = service.calculate_next_run(t2, with_jitter=True, update_state=True) assert datetime(2026, 8, 29, 12, 30, 0) <= target2 <= datetime(2026, 8, 29, 13, 30, 0) # Sau khi chạy slot 13:00 lúc 13:15, slot kế tiếp phải là ngày hôm sau 01:00 t3 = datetime(2026, 8, 29, 13, 15, 0) target3, delay3 = service.calculate_next_run(t3, with_jitter=True, update_state=True) assert datetime(2026, 8, 30, 0, 30, 0) <= target3 <= datetime(2026, 8, 30, 1, 30, 0) @pytest.mark.anyio async def test_execute_scheduled_sync_success_no_alert(self): mock_invoice_svc = MagicMock() mock_invoice_svc.run_gdt_tax_sync.return_value = { "status": "success", "message": "Đã kết nối Tổng Cục Thuế và nạp 8 hóa đơn mới (12 đã tồn tại).", "total_new": 8, "total_duplicates": 12, "connectors": [ { "service": "gdt_portal", "status": "success", "message": "OK", "new_imported": 8, "duplicates": 12, } ], } service = InvoiceAutoSyncSchedulerService( invoice_service=mock_invoice_svc, alert_email="contact@dinhsonconstruction.com", ) with patch( "app.modules.core.application.alert_service.BrevoEmailService.send_email", new_callable=AsyncMock, ) as mock_send: res = await service.execute_scheduled_sync(trigger_slot="09:00") assert res["status"] == "success" assert res["new_imported"] == 8 assert res["duplicates"] == 12 assert res["alert_sent"] is False # Không gửi email khi đồng bộ thành công mock_send.assert_not_called() @pytest.mark.anyio async def test_execute_scheduled_sync_failure_sends_email_alert(self): mock_invoice_svc = MagicMock() mock_invoice_svc.run_gdt_tax_sync.return_value = { "status": "auth_error", "message": "Tài khoản Cổng HĐĐT hoadondientu.gdt.gov.vn bị sai mật khẩu.", "connectors": [ { "service": "gdt_portal", "status": "auth_error", "message": "Mật khẩu không đúng hoặc tài khoản bị khóa.", } ], } service = InvoiceAutoSyncSchedulerService( invoice_service=mock_invoice_svc, alert_email="contact@dinhsonconstruction.com", ) with patch( "app.modules.core.application.alert_service.BrevoEmailService.send_email", new_callable=AsyncMock, ) as mock_send: mock_send.return_value = { "success": True, "message_id": "", } res = await service.execute_scheduled_sync(trigger_slot="05:00") assert res["status"] == "auth_error" assert res["alert_sent"] is True assert ( res["error_details"] == "Tài khoản Cổng HĐĐT hoadondientu.gdt.gov.vn bị sai mật khẩu." ) # Kiểm tra BrevoEmailService được gọi đúng với địa chỉ email mock_send.assert_called_once() call_kwargs = mock_send.call_args.kwargs assert call_kwargs["recipient_email"] == "contact@dinhsonconstruction.com" assert "🚨 [DSCons ERP Alert]" in call_kwargs["subject"] assert "05:00" in call_kwargs["subject"] assert "0202111150" in call_kwargs["html_content"] assert "Mật khẩu không đúng" in call_kwargs["html_content"] @pytest.mark.anyio async def test_execute_scheduled_sync_exception_sends_alert(self): mock_invoice_svc = MagicMock() mock_invoice_svc.run_gdt_tax_sync.side_effect = ConnectionError( "Cổng Tổng Cục Thuế không phản hồi (Connection timed out)" ) service = InvoiceAutoSyncSchedulerService( invoice_service=mock_invoice_svc, alert_email="contact@dinhsonconstruction.com", ) with patch( "app.modules.core.application.alert_service.BrevoEmailService.send_email", new_callable=AsyncMock, ) as mock_send: mock_send.return_value = { "success": True, "message_id": "", } res = await service.execute_scheduled_sync(trigger_slot="13:00") assert res["status"] == "exception" assert res["alert_sent"] is True assert "Connection timed out" in res["error_details"] mock_send.assert_called_once() call_kwargs = mock_send.call_args.kwargs assert call_kwargs["recipient_email"] == "contact@dinhsonconstruction.com" assert "13:00" in call_kwargs["subject"] assert "Connection timed out" in call_kwargs["html_content"] def test_get_scheduler_status(self): service = InvoiceAutoSyncSchedulerService( alert_email="contact@dinhsonconstruction.com" ) status = service.get_status() assert status["scheduled_hours"] == [1, 13] assert status["scheduled_slots"] == [ "01:00", "13:00", ] assert status["jitter_minutes"] == 30 assert status["enable_jitter"] is True assert status["alert_recipient_email"] == "contact@dinhsonconstruction.com" assert "next_run_at" in status assert "seconds_until_next_run" in status assert "is_paused" in status assert "pause_reason" in status @pytest.mark.anyio async def test_execute_scheduled_sync_when_paused_skips_run_and_alert(self): mock_invoice_svc = MagicMock() service = InvoiceAutoSyncSchedulerService( invoice_service=mock_invoice_svc, alert_email="contact@dinhsonconstruction.com", is_paused=True, pause_reason="Cục thuế tạm khóa tài khoản", ) with patch( "app.modules.core.application.alert_service.BrevoEmailService.send_email", new_callable=AsyncMock, ) as mock_send: res = await service.execute_scheduled_sync(trigger_slot="09:00", force=False) assert res["status"] == "paused" assert res["alert_sent"] is False assert "Cục thuế tạm khóa" in res["error_details"] mock_invoice_svc.run_gdt_tax_sync.assert_not_called() mock_send.assert_not_called() @pytest.mark.anyio async def test_execute_scheduled_sync_when_paused_allows_force_override(self): mock_invoice_svc = MagicMock() mock_invoice_svc.run_gdt_tax_sync.return_value = { "status": "success", "total_new": 1, "total_duplicates": 2, "connectors": [{"service": "gdt_portal", "status": "success"}], } service = InvoiceAutoSyncSchedulerService( invoice_service=mock_invoice_svc, alert_email="contact@dinhsonconstruction.com", is_paused=True, ) res = await service.execute_scheduled_sync(trigger_slot="MANUAL-TRIGGER", force=True) assert res["status"] == "success" assert res["new_imported"] == 1 mock_invoice_svc.run_gdt_tax_sync.assert_called_once() def test_pause_and_resume_methods(self): service = InvoiceAutoSyncSchedulerService(is_paused=False) assert service.is_paused is False assert service.pause_reason == "" service.pause("Bảo trì hệ thống thuế") assert service.is_paused is True assert service.pause_reason == "Bảo trì hệ thống thuế" service.resume() assert service.is_paused is False assert service.pause_reason == "" def test_api_scheduler_routes(self, auth_headers): app = create_app() app.dependency_overrides = {} client = TestClient(app) # 1. GET /v1/erp/invoices/sync/scheduler-status resp = client.get( "/v1/erp/invoices/sync/scheduler-status", headers=auth_headers ) assert resp.status_code == 200 data = resp.json() assert data["status"] == "success" sched = data["scheduler"] assert sched["scheduled_hours"] == [1, 13] assert sched["alert_recipient_email"] == "contact@dinhsonconstruction.com" assert "is_paused" in sched # 2. POST /v1/erp/invoices/sync/scheduler/pause pause_resp = client.post( "/v1/erp/invoices/sync/scheduler/pause", json={"reason": "Cục thuế khóa tài khoản kiểm tra"}, headers=auth_headers, ) assert pause_resp.status_code == 200 pause_data = pause_resp.json() assert pause_data["status"] == "success" assert pause_data["scheduler"]["is_paused"] is True assert "Cục thuế khóa" in pause_data["scheduler"]["pause_reason"] # 3. POST /v1/erp/invoices/sync/scheduler/resume resume_resp = client.post( "/v1/erp/invoices/sync/scheduler/resume", headers=auth_headers, ) assert resume_resp.status_code == 200 resume_data = resume_resp.json() assert resume_data["status"] == "success" assert resume_data["scheduler"]["is_paused"] is False # 4. POST /v1/erp/invoices/sync/trigger-scheduled-sync with patch.object( InvoiceAutoSyncSchedulerService, "execute_scheduled_sync", new_callable=AsyncMock, ) as mock_exec: mock_exec.return_value = { "status": "success", "trigger_slot": "MANUAL-TRIGGER", "new_imported": 3, "duplicates": 10, "alert_sent": False, } post_resp = client.post( "/v1/erp/invoices/sync/trigger-scheduled-sync", headers=auth_headers ) assert post_resp.status_code == 200 post_data = post_resp.json() assert post_data["status"] == "success" assert post_data["result"]["new_imported"] == 3 @pytest.mark.anyio async def test_circuit_breaker_first_auth_failure_does_not_pause(self): mock_invoice_svc = MagicMock() mock_invoice_svc.run_gdt_tax_sync.return_value = { "status": "auth_error", "message": "Cổng Tổng Cục Thuế phản hồi: Tên đăng nhập hoặc mật khẩu không đúng.", "connectors": [ { "service": "gdt_portal", "tax_code": "0202111150", "status": "auth_error", "message": "Tên đăng nhập hoặc mật khẩu không đúng.", } ], } mock_invoice_svc.increment_auth_failures.return_value = 1 service = InvoiceAutoSyncSchedulerService( invoice_service=mock_invoice_svc, alert_email="contact@dinhsonconstruction.com", max_consecutive_auth_failures=2, ) with patch( "app.modules.core.application.alert_service.BrevoEmailService.send_email", new_callable=AsyncMock, ) as mock_send: mock_send.return_value = {"success": True, "message_id": ""} res = await service.execute_scheduled_sync(trigger_slot="01:00") assert res["status"] == "auth_error" assert service.is_paused is False assert res.get("circuit_breaker_tripped") is not True mock_invoice_svc.increment_auth_failures.assert_called_once_with( "gdt_portal", "0202111150" ) mock_invoice_svc.deactivate_sync_config.assert_not_called() mock_send.assert_called_once() assert "01:00" in mock_send.call_args.kwargs["subject"] @pytest.mark.anyio async def test_circuit_breaker_second_auth_failure_trips_and_pauses(self): mock_invoice_svc = MagicMock() mock_invoice_svc.run_gdt_tax_sync.return_value = { "status": "auth_error", "message": "Cổng Tổng Cục Thuế phản hồi: Tên đăng nhập hoặc mật khẩu không đúng.", "connectors": [ { "service": "gdt_portal", "tax_code": "0202111150", "status": "auth_error", "message": "Tên đăng nhập hoặc mật khẩu không đúng.", } ], } # Lần thứ 2 thất bại liên tiếp -> đạt ngưỡng 2 mock_invoice_svc.increment_auth_failures.return_value = 2 service = InvoiceAutoSyncSchedulerService( invoice_service=mock_invoice_svc, alert_email="contact@dinhsonconstruction.com", max_consecutive_auth_failures=2, ) with patch( "app.modules.core.application.alert_service.BrevoEmailService.send_email", new_callable=AsyncMock, ) as mock_send: mock_send.return_value = {"success": True, "message_id": ""} res = await service.execute_scheduled_sync(trigger_slot="05:00") assert res["status"] == "circuit_breaker_suspended" assert res["circuit_breaker_tripped"] is True assert service.is_paused is True assert "ngắt mạch" in service.pause_reason.lower() mock_invoice_svc.increment_auth_failures.assert_called_once_with( "gdt_portal", "0202111150" ) mock_invoice_svc.deactivate_sync_config.assert_called_once() deact_kwargs = mock_invoice_svc.deactivate_sync_config.call_args.kwargs assert deact_kwargs["service_type"] == "gdt_portal" assert deact_kwargs["tax_code"] == "0202111150" assert deact_kwargs["status"] == "CIRCUIT_BREAKER_SUSPENDED" # Email khẩn cấp được gửi mock_send.assert_called_once() send_kwargs = mock_send.call_args.kwargs assert "KHẨN CẤP" in send_kwargs["subject"] assert "Ngắt Mạch Đồng Bộ" in send_kwargs["subject"] assert "0202111150" in send_kwargs["html_content"] @pytest.mark.anyio async def test_sync_success_resets_auth_failures(self): mock_invoice_svc = MagicMock() mock_invoice_svc.run_gdt_tax_sync.return_value = { "status": "success", "total_new": 5, "total_duplicates": 10, "connectors": [ { "service": "gdt_portal", "tax_code": "0202111150", "status": "success", "message": "OK", "new_imported": 5, "duplicates": 10, } ], } service = InvoiceAutoSyncSchedulerService( invoice_service=mock_invoice_svc, alert_email="contact@dinhsonconstruction.com", ) res = await service.execute_scheduled_sync(trigger_slot="09:00") assert res["status"] == "success" mock_invoice_svc.reset_auth_failures.assert_any_call("gdt_portal", "0202111150") @pytest.mark.anyio async def test_execute_scheduled_sync_cooldown_active_no_alert(self): mock_invoice_svc = MagicMock() mock_invoice_svc.run_gdt_tax_sync.return_value = { "status": "cooldown_active", "message": "Tài khoản Cổng Thuế đang trong thời gian giãn cách 3 giờ đến 2026-09-22T10:00:00Z.", "connectors": [ { "service": "gdt_portal", "tax_code": "0202111150", "status": "cooldown_active", "message": "Cooldown 3h", } ], } service = InvoiceAutoSyncSchedulerService( invoice_service=mock_invoice_svc, alert_email="contact@dinhsonconstruction.com", ) with patch( "app.modules.core.application.alert_service.BrevoEmailService.send_email", new_callable=AsyncMock, ) as mock_send: res = await service.execute_scheduled_sync(trigger_slot="09:00") assert res["status"] == "cooldown_active" assert res["new_imported"] == 0 assert res["duplicates"] == 0 assert "giãn cách" in res["error_details"] # Không được gửi email cảnh báo khi đang cooldown mock_send.assert_not_called() # Không được tăng số lần lỗi hoặc kích hoạt ngắt mạch mock_invoice_svc.increment_auth_failures.assert_not_called() assert service.is_paused is False def test_resume_resets_auth_failures_and_clears_cooldown(self): mock_invoice_svc = MagicMock() service = InvoiceAutoSyncSchedulerService( invoice_service=mock_invoice_svc, is_paused=True, pause_reason="Test pause", ) assert service.is_paused is True service.resume() assert service.is_paused is False assert service.pause_reason == "" mock_invoice_svc.reset_auth_failures.assert_called_once_with( "gdt_portal", "0202111150" ) mock_invoice_svc.clear_auth_cooldown.assert_called_once_with( "gdt_portal", "0202111150" )