""" Unit & Integration Tests for AIHawk Stealth Anti-Bot and 3-Hour Auth Cooldown. Ensures zero-risk of GDT account lockout: 1. AIHawk stealth scripts and Chromium flags properly configured. 2. MAX_ATTEMPTS is strictly capped at 1. 3. Cooldown logic in storage_sync_config correctly calculates and checks expiry. 4. GDT Fetcher aborts connection attempt if cooldown is active. 5. GDT Fetcher triggers 3h cooldown on authentication failure. 6. GDT Sync aggregates cooldown status cleanly. """ from datetime import datetime, timedelta, timezone from unittest.mock import MagicMock, patch import uuid import pytest from app.modules.invoices.application.invoice_processing import InvoiceProcessingService from app.modules.invoices.application.invoice_processing.gdt_auth import ( AIHAWK_STEALTH_JS, GDT_MAX_AUTH_ATTEMPTS, ) class TestGdtStealthConfiguration: def test_stealth_script_masks_automation(self): """Kiểm tra AIHawk Stealth script loại bỏ hoàn toàn các dấu vết tự động hóa.""" assert "webdriver" in AIHAWK_STEALTH_JS assert "Object.defineProperty(navigator, 'webdriver'" in AIHAWK_STEALTH_JS assert "window.chrome" in AIHAWK_STEALTH_JS assert "'platform'" in AIHAWK_STEALTH_JS assert "Win32" in AIHAWK_STEALTH_JS assert "'plugins'" in AIHAWK_STEALTH_JS assert "'languages'" in AIHAWK_STEALTH_JS assert "permissions.query" in AIHAWK_STEALTH_JS def test_max_attempts_strictly_capped_at_one(self): """Quy định an toàn tuyệt đối: Không được tự động retry đăng nhập trong 1 phiên (MAX_ATTEMPTS = 1).""" assert GDT_MAX_AUTH_ATTEMPTS == 1 class TestGdtAuthCooldownStorage: def test_cooldown_expiry_check_logic(self): """Kiểm tra tính toán cooldown_until khi check_auth_cooldown.""" svc = InvoiceProcessingService() now = datetime.now(timezone.utc) future_iso = (now + timedelta(hours=2)).isoformat() past_iso = (now - timedelta(minutes=10)).isoformat() mock_conn = MagicMock() mock_conn.__enter__.return_value = mock_conn mock_cur = MagicMock() mock_conn.cursor.return_value.__enter__.return_value = mock_cur with patch.object(svc, "get_connection", return_value=mock_conn): # 1. Cooldown còn hiệu lực (> now) mock_cur.fetchone.return_value = {"cd_until": future_iso} is_active, until_str = svc.check_auth_cooldown("gdt_portal", "0202111150") assert is_active is True assert until_str == future_iso # 2. Cooldown đã hết hạn (< now) mock_cur.fetchone.return_value = {"cd_until": past_iso} is_active, until_str = svc.check_auth_cooldown("gdt_portal", "0202111150") assert is_active is False assert until_str is None # 3. Không có cooldown trong cấu hình mock_cur.fetchone.return_value = {"cd_until": None} is_active, until_str = svc.check_auth_cooldown("gdt_portal", "0202111150") assert is_active is False assert until_str is None class TestGdtFetcherCooldownEnforcement: def test_fetcher_aborts_when_in_cooldown(self): """Fetcher phải lập tức dừng phiên đồng bộ nếu tài khoản đang trong thời gian giãn cách.""" svc = InvoiceProcessingService() future_str = (datetime.now(timezone.utc) + timedelta(hours=3)).isoformat() with patch.object( svc, "check_auth_cooldown", return_value=(True, future_str) ) as mock_check: res = svc._execute_real_gdt_connection( config_id=uuid.uuid4(), tax_code="0202111150", username="0202111150", password="encrypted_password", extra={}, from_date="2026-08-01", to_date="2026-08-30", ) assert res["status"] == "cooldown_active" assert "chờ an toàn 3 giờ" in res["message"] assert res["cooldown_until"] == future_str mock_check.assert_called_once_with("gdt_portal", "0202111150") def test_fetcher_activates_cooldown_on_auth_failure(self): """Fetcher phải kích hoạt cooldown 3 giờ ngay lập tức khi đăng nhập thất bại.""" svc = InvoiceProcessingService() with patch.object( svc, "check_auth_cooldown", return_value=(False, None) ), patch.object( svc, "_authenticate_gdt_portal", return_value={ "status": "auth_error", "message": "Mật khẩu không chính xác hoặc tài khoản đang bị tạm khóa", }, ), patch.object( svc, "set_auth_cooldown", return_value="2026-09-22T10:00:00Z" ) as mock_set_cooldown, patch.object( svc, "_update_sync_status" ): res = svc._execute_real_gdt_connection( config_id=uuid.uuid4(), tax_code="0202111150", username="0202111150", password="wrong_password", extra={}, from_date="2026-08-01", to_date="2026-08-30", ) assert res["status"] == "auth_error" # Phải kích hoạt cooldown 3 giờ mock_set_cooldown.assert_called_once_with("gdt_portal", "0202111150", hours=3.0) assert "3 giờ" in res["message"] class TestGdtSyncCooldownAggregation: def test_gdt_sync_returns_cooldown_active_status(self): """run_gdt_tax_sync phải trả về status='cooldown_active' khi connector bị cooldown.""" svc = InvoiceProcessingService() # Giả lập cursor query DB trả về 1 config GDT mock_conn = MagicMock() mock_conn.__enter__.return_value = mock_conn mock_cur = MagicMock() mock_conn.cursor.return_value.__enter__.return_value = mock_cur mock_cur.fetchall.return_value = [ { "id": uuid.uuid4(), "service_type": "gdt_portal", "tax_code": "0202111150", "username": "0202111150", "password_hash": "encrypted_pw", "extra_config": {}, "is_active": True, } ] with patch.object( svc, "get_connection", return_value=mock_conn ), patch.object( svc, "_execute_real_gdt_connection", return_value={ "status": "cooldown_active", "message": "Đang trong thời gian giãn cách bảo vệ tài khoản đến 10:00", "service": "gdt_portal", "tax_code": "0202111150", }, ): res = svc.run_gdt_tax_sync( tax_code="0202111150", from_date="2026-08-01", to_date="2026-08-30", sync_type="all", ) assert res["status"] == "cooldown_active" assert "giãn cách" in res["message"] assert len(res["connectors"]) == 1 assert res["connectors"][0]["status"] == "cooldown_active"