"""Unit & Integration tests for Enterprise E-Invoice Processing Engine."""
from decimal import Decimal
import pytest
from fastapi.testclient import TestClient
from app.modules.auth.presentation.auth import get_current_user
from app.main import app
from app.modules.invoices.application.invoice_processing_service import InvoiceProcessingService, to_decimal
app.dependency_overrides[get_current_user] = lambda: {
"id": "admin-id",
"email": "admin@dscons.vn",
"role": "superadmin",
"full_name": "Ban Quản Trị Hệ Thống DSCons",
}
client = TestClient(app)
service = InvoiceProcessingService()
def test_to_decimal_precision():
"""Kiểm tra độ chính xác số học Decimal (NUMERIC(18, 4))."""
assert to_decimal("1500000.50") == Decimal("1500000.5000")
assert to_decimal(12.34567) == Decimal("12.3457")
assert to_decimal(None) == Decimal("0.0000")
assert to_decimal("") == Decimal("0.0000")
def test_parse_gdt_nd123_xml():
"""Kiểm tra bóc tách XML chuẩn Nghị định 123 của Tổng Cục Thuế."""
sample_gdt_xml = """
2.0.0
HÓA ĐƠN GIÁ TRỊ GIA TĂNG
1
1C26TTEST
00099999
2026-08-20
VND
1.0000
TCT-TEST-99999
CÔNG TY THÉP TIẾN LÊN HẢI PHÒNG
0209998881
Quận Ngô Quyền, Hải Phòng
CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN
0202111150
Thôn Tú Đôi 3, xã Nghi Dương, TP Hải Phòng
1
Thép hình chữ I 150x75x5x7
Cây
10.0000
2500000.0000
25000000.0000
8%
2000000.0000
25000000.0000
2000000.0000
27000000.0000
Hai mươi bảy triệu đồng.
CN=CÔNG TY THÉP TIẾN LÊN, MST=0209998881
2026-08-20T10:00:00Z
"""
parsed = service.parse_xml_content(sample_gdt_xml, source_channel="test_gdt")
assert parsed["direction"] == "input"
assert parsed["invoice_number"] == "00099999"
assert parsed["invoice_series"] == "1C26TTEST"
assert parsed["seller_tax_code"] == "0209998881"
assert parsed["buyer_tax_code"] == "0202111150"
assert parsed["subtotal_amount_vnd"] == Decimal("25000000.0000")
assert parsed["vat_amount_vnd"] == Decimal("2000000.0000")
assert parsed["total_amount_vnd"] == Decimal("27000000.0000")
assert parsed["signature_valid"] is True
assert len(parsed["items"]) == 1
assert parsed["items"][0]["item_name"] == "Thép hình chữ I 150x75x5x7"
assert parsed["items"][0]["unit"] == "Cây"
def test_parse_misa_meinvoice_xml():
"""Kiểm tra bóc tách XML từ phần mềm MISA meInvoice."""
sample_misa_xml = """
00000888
1C26TDS
1
2026-08-21
0202111150
CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN
0203334445
BAN QLDA ĐẦU TƯ XÂY DỰNG HUYỆN AN LÃO
1200000000.0000
96000000.0000
1296000000.0000
Một tỷ hai trăm chín mươi sáu triệu đồng.
-
Nghiệm thu khối lượng hoàn thành Cầu vượt sông Đa Độ
Đợt
1.0000
1200000000.0000
1200000000.0000
96000000.0000
1296000000.0000
"""
parsed = service.parse_xml_content(sample_misa_xml, source_channel="test_misa")
assert parsed["direction"] == "output" # DSCons is seller
assert parsed["invoice_number"] == "00000888"
assert parsed["seller_tax_code"] == "0202111150"
assert parsed["buyer_tax_code"] == "0203334445"
assert parsed["total_amount_vnd"] == Decimal("1296000000.0000")
assert len(parsed["items"]) == 1
def test_sha256_deduplication_and_persistence():
"""Kiểm tra lưu trữ vào PostgreSQL và cơ chế chống trùng lặp SHA-256."""
unique_xml = """
1
1C26TUNIQ
00009911
2026-08-21
CÔNG TY XI MĂNG VICEM HẢI PHÒNG
0200881122
CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN
0202111150
1
Xi măng PCB40 Vicem Hải Phòng
Tấn
50.0000
1400000.0000
70000000.0000
8%
5600000.0000
70000000.0000
5600000.0000
75600000.0000
"""
parsed = service.parse_xml_content(unique_xml, source_channel="test_dedup")
try:
res1 = service.save_invoice(parsed, user_id="Test Runner")
assert res1["status"] in ("success", "duplicate")
# Second save with exact same XML content must return duplicate
res2 = service.save_invoice(parsed, user_id="Test Runner")
assert res2["status"] == "duplicate"
assert "đã tồn tại" in res2["message"]
finally:
with service.get_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"DELETE FROM erp_invoices WHERE invoice_number = '00009911' OR source_channel = 'test_dedup';"
)
conn.commit()
def test_api_list_and_kpis():
"""Kiểm tra API lấy danh sách và tổng hợp chỉ số KPI hóa đơn."""
# Test KPIs
res_kpi = client.get("/v1/erp/invoices/summary/kpis")
assert res_kpi.status_code == 200
kpi_data = res_kpi.json()
assert kpi_data["status"] == "success"
assert "total_input_invoices" in kpi_data["kpis"]
assert "net_vat_balance_vnd" in kpi_data["kpis"]
# Test List
res_list = client.get("/v1/erp/invoices")
assert res_list.status_code == 200
assert res_list.status_code == 200, res_list.text
list_data = res_list.json()
assert list_data["status"] == "success"
assert isinstance(list_data["invoices"], list)
assert len(list_data["invoices"]) > 0
first_inv_id = list_data["invoices"][0]["id"]
res_detail = client.get(f"/v1/erp/invoices/{first_inv_id}")
assert res_detail.status_code == 200
det_data = res_detail.json()
assert det_data["invoice"]["id"] == first_inv_id
assert "items" in det_data["invoice"]
def test_sync_configuration_and_truthful_sync():
"""Kiểm tra lưu trữ cấu hình kết nối thực tế và phản hồi trung thực khi chưa cấu hình hoặc lỗi xác thực."""
try:
# 1. Test sync when unconfigured for a non-existent test company tax code - must return not_configured, NEVER fake success
res_sync_unconfigured = client.post(
"/v1/erp/invoices/sync-gdt?tax_code=0000000000"
)
assert res_sync_unconfigured.status_code == 200
sync_data = res_sync_unconfigured.json()
assert sync_data["status"] == "not_configured"
assert "Chưa cấu hình tài khoản kết nối" in sync_data["message"]
# 2. Test saving GDT config with test tax code
res_save = client.post(
"/v1/erp/invoices/sync/configs",
json={
"service_type": "gdt_portal",
"tax_code": "020TEST999",
"username": "020TEST999",
"password": "TestPasswordSecure123!",
"extra_config": {},
},
)
assert res_save.status_code == 200
assert res_save.json()["status"] == "success"
# 2.1 Test auth failures tracking in PostgreSQL for Circuit Breaker
assert service.get_auth_failures("gdt_portal", "020TEST999") == 0
f1 = service.increment_auth_failures("gdt_portal", "020TEST999")
assert f1 == 1
assert service.get_auth_failures("gdt_portal", "020TEST999") == 1
f2 = service.increment_auth_failures("gdt_portal", "020TEST999")
assert f2 == 2
assert service.get_auth_failures("gdt_portal", "020TEST999") == 2
service.reset_auth_failures("gdt_portal", "020TEST999")
assert service.get_auth_failures("gdt_portal", "020TEST999") == 0
# 3. Test sync with invalid test credentials - must return auth_error/connection_error, NEVER success
res_sync_invalid = client.post("/v1/erp/invoices/sync-gdt?tax_code=020TEST999")
assert res_sync_invalid.status_code == 200
invalid_data = res_sync_invalid.json()
assert invalid_data["status"] in ("auth_error", "connection_error", "error")
msg_lower = invalid_data["message"].lower()
assert (
"lỗi" in msg_lower
or "không thể" in msg_lower
or "cổng tổng cục thuế" in msg_lower
)
finally:
# 4. Clean up only test config
with service.get_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"DELETE FROM erp_invoice_sync_configs WHERE tax_code IN ('020TEST999', '0000000000');"
)
conn.commit()
def test_categorize_item_cost():
"""Kiểm tra nhận diện và phân nhóm chi phí tự động theo từ khóa xây dựng."""
assert (
service.categorize_item_cost("Thép cuộn D10 Hòa Phát CB300", "VT-THEP")
== "material_main"
)
assert (
service.categorize_item_cost("Bê tông thương phẩm mác 300 R28", "VT-BETONG")
== "material_main"
)
assert (
service.categorize_item_cost("Xi măng Vicem Hải Phòng PCB40", "VT-XM")
== "material_main"
)
assert (
service.categorize_item_cost("Cát vàng hạt lớn bê tông", "VT-CAT")
== "material_main"
)
assert (
service.categorize_item_cost("Thuê ca máy đào bánh xích 0.8m3", "CA-MAY")
== "machinery_labor"
)
assert (
service.categorize_item_cost("Gia công ép cọc bê tông cốt thép D400", "TP-COC")
== "subcontractor"
)
assert (
service.categorize_item_cost(
"Chi phí dựng lán trại tạm thời công trường", "CP-LAN"
)
== "general_expense"
)
assert (
service.categorize_item_cost("Văn phòng phẩm và giấy in A4", "VPP")
== "general_expense"
)
assert (
service.categorize_item_cost("Bánh kẹo trà hoa quả tiếp khách họp", "OTHER")
== "other"
)
def test_vat_tax_schedule_and_csv_export():
"""Kiểm tra tổng hợp Bảng kê thuế GTGT (01-1 & 01-2) và xuất CSV UTF-8 BOM."""
# Test service method
schedule = service.get_vat_tax_schedule()
assert "tax_breakdown" in schedule
assert "input_schedule" in schedule
assert "output_schedule" in schedule
assert "kpis" in schedule
assert "rate_8" in schedule["tax_breakdown"]["input"]
assert "rate_10" in schedule["tax_breakdown"]["input"]
# Test API endpoint
res = client.get("/v1/erp/invoices/reports/vat-schedule")
assert res.status_code == 200
data = res.json()
assert data["status"] == "success"
assert "schedule" in data
# Test CSV export endpoint
res_csv = client.get("/v1/erp/invoices/reports/vat-schedule/export?direction=all")
assert res_csv.status_code == 200
assert res_csv.headers["content-type"].startswith("text/csv")
csv_text = res_csv.text
assert "BẢNG KÊ" in csv_text
assert "Số HĐ" in csv_text
def test_material_and_cost_analytics():
"""Kiểm tra phân tích bóc tách chi phí vật tư và đơn giá bình quân gia quyền."""
# Test service method
analytics = service.get_material_and_cost_analytics()
assert "total_spend_vnd" in analytics
assert "categories_breakdown" in analytics
assert "top_materials" in analytics
assert isinstance(analytics["categories_breakdown"], list)
assert len(analytics["categories_breakdown"]) >= 1
# Test API endpoint
res = client.get("/v1/erp/invoices/analytics/cost-breakdown")
assert res.status_code == 200
data = res.json()
assert data["status"] == "success"
assert "analytics" in data
def test_fidic_ipc_output_invoice_creation_and_audit():
"""Kiểm tra lập hóa đơn nghiệm thu IPC (FIDIC) với khấu trừ tạm ứng & giữ lại bảo hành, kèm rà soát rủi ro thuế."""
# 1. Get an existing project
res_proj = client.get("/v1/erp/projects")
assert res_proj.status_code == 200
proj_resp = res_proj.json()
projects = (
proj_resp if isinstance(proj_resp, list) else proj_resp.get("projects", [])
)
if not projects:
pytest.skip("Chưa có dự án trong DB để test lập hóa đơn IPC")
target_project = projects[0]
proj_id = target_project["id"]
# 2. Call API to create FIDIC IPC Invoice
req_payload = {
"project_id": proj_id,
"wbs_items": [
{
"item_name": "Nghiệm thu khối lượng hoàn thành Cột dầm sàn Tầng 2",
"item_code": "WBS-COL-02",
"unit": "Đợt",
"quantity": 1,
"unit_price_vnd": 600000000,
},
{
"item_name": "Thi công cốt thép dầm sàn Tầng 2",
"item_code": "WBS-STEEL-02",
"unit": "Tấn",
"quantity": 10,
"unit_price_vnd": 20000000,
},
],
"advance_deduction_pct": 10.0,
"retention_pct": 5.0,
"vat_rate_pct": 8.0,
"buyer_tax_code": "0209991122",
"buyer_name": "BAN QUẢN LÝ DỰ ÁN KIẾM THỬ TỰ ĐỘNG",
"buyer_address": "Số 1 Đường Hùng Vương, TP Hải Phòng",
"invoice_series": "1C26TTEST",
"notes": "Kiểm thử tự động hóa đơn IPC",
}
res_ipc = client.post("/v1/erp/invoices/output/ipc-invoice", json=req_payload)
assert res_ipc.status_code == 200
ipc_data = res_ipc.json()
assert ipc_data["status"] == "success"
assert "invoice_id" in ipc_data
created_id = ipc_data["invoice_id"]
try:
# Gross = 600m + 200m = 800m
# Advance deduction 10% = 80m
# Retention 5% = 40m
# Net subtotal = 800m - 80m - 40m = 680m
# VAT 8% = 54.4m
# Total payable = 734.4m
deductions = ipc_data["ipc_deductions"]
assert deductions["gross_milestone_amount_vnd"] == 800000000.0
assert deductions["advance_deduction_amount_vnd"] == 80000000.0
assert deductions["retention_amount_vnd"] == 40000000.0
assert deductions["net_subtotal_vnd"] == 680000000.0
assert deductions["vat_amount_vnd"] == 54400000.0
assert deductions["total_payable_vnd"] == 734400000.0
# 3. Test Tax Risk Audit on created invoice
res_audit = client.get(f"/v1/erp/invoices/{created_id}/tax-risk-audit")
assert res_audit.status_code == 200
audit_data = res_audit.json()
assert audit_data["status"] == "success"
report = audit_data["audit_report"]
assert "risk_score" in report
assert "risk_level" in report
assert "risk_flags" in report
assert isinstance(report["risk_flags"], list)
assert len(report["risk_flags"]) > 0
finally:
# Clean up test invoice
with service.get_connection() as conn, conn.cursor() as cur:
cur.execute("DELETE FROM erp_invoices WHERE id = %s;", (created_id,))
conn.commit()
def test_cash_register_mtt_recognition_and_metadata():
"""Kiểm tra nhận diện Hóa đơn máy tính tiền MTT (K26TXN, K23TXN, 1M26...) và cấp nhãn chuẩn."""
assert InvoiceProcessingService.is_cash_register_invoice("K26TXN", "1") is True
assert InvoiceProcessingService.is_cash_register_invoice("K23TXN", "1") is True
assert InvoiceProcessingService.is_cash_register_invoice("1M26TDS", "1") is True
assert InvoiceProcessingService.is_cash_register_invoice("M26POS", "2") is True
assert InvoiceProcessingService.is_cash_register_invoice("1C26TDS", "1") is False
assert InvoiceProcessingService.is_cash_register_invoice("C24TAA", "1") is False
petrol_meta = InvoiceProcessingService.get_invoice_type_metadata(
"K26TXN", "1", "CÔNG TY TNHH MTV PETROLIMEX HẢI PHÒNG"
)
assert petrol_meta["is_cash_register"] is True
assert petrol_meta["type_category"] == "cash_register_petrol"
assert "XĂNG DẦU" in petrol_meta["type_label"]
assert "Điều 11" in petrol_meta["legal_basis"]
sales_mtt_meta = InvoiceProcessingService.get_invoice_type_metadata(
"M26ABC", "2", "Siêu Thị Vật Tư"
)
assert sales_mtt_meta["is_cash_register"] is True
assert sales_mtt_meta["type_category"] == "cash_register_sales"
std_meta = InvoiceProcessingService.get_invoice_type_metadata(
"1C26TDS", "1", "Công Ty Xi Măng"
)
assert std_meta["is_cash_register"] is False
assert std_meta["type_category"] == "standard_vat"
def test_mtt_tax_risk_audit_signature_exemption():
"""Kiểm tra miễn trừ chữ ký số cho HĐĐT máy tính tiền theo Khoản 3 Điều 11 NĐ 123/2020/NĐ-CP."""
sample_mtt_xml = """
2.0.0
HÓA ĐƠN GTGT KHỞI TẠO TỪ MÁY TÍNH TIỀN
1
K26TXNTEST
00888888
2026-08-21
VND
1.0000
MTT-CQT-00888888
CÔNG TY XĂNG DẦU KIỂM THỬ MTT
0209997771
Hải Phòng
CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN
0202111150
Hải Phòng
1
Xăng RON 95-III (Bơm từ máy tính tiền)
Lít
50.0000
24000.0000
1200000.0000
8%
96000.0000
1200000.0000
96000.0000
1296000.0000
Một triệu hai trăm chín mươi sáu nghìn đồng.
"""
parsed = service.parse_xml_content(sample_mtt_xml, source_channel="manual_xml")
assert parsed["invoice_series"] == "K26TXNTEST"
assert parsed["signature_valid"] is True
assert "Mã CQT" in parsed["signed_by"] or "MTT" in parsed["signed_by"]
save_res = service.save_invoice(parsed, user_id="Test MTT Runner")
assert save_res["status"] in ("success", "duplicate")
inv_id = save_res["invoice_id"]
try:
audit_res = service.audit_invoice_tax_risk(inv_id)
assert "risk_score" in audit_res
# Must have safe risk score 0 and exemption flag
assert audit_res["risk_score"] == 0
flag_types = [f["type"] for f in audit_res["risk_flags"]]
assert "SIGNATURE_EXEMPT_MTT" in flag_types
# Test list API with MTT filter
res_list = client.get(
"/v1/erp/invoices?invoice_type_filter=cash_register&search=K26TXNTEST"
)
assert res_list.status_code == 200
invs = res_list.json()["invoices"]
assert len(invs) >= 1
assert invs[0]["is_cash_register"] is True
finally:
with service.get_connection() as conn, conn.cursor() as cur:
cur.execute("DELETE FROM erp_invoices WHERE id = %s;", (inv_id,))
conn.commit()
def test_clean_captcha_text():
"""Kiểm tra độ chính xác và làm sạch ký tự Captcha từ AI Vision (Sanitization)."""
from app.modules.invoices.application.invoice_processing.gdt_captcha import clean_captcha_text
assert clean_captcha_text("ABCDE") == "ABCDE"
assert clean_captcha_text("abcde") == "ABCDE"
assert clean_captcha_text("```\nXY789\n```") == "XY789"
assert clean_captcha_text("Captcha: 6H789K") == "6H789K"
assert clean_captcha_text("Result: **7K8P9**") == "7K8P9"
assert clean_captcha_text("a 1 b 2 c 3") == "A1B2C3"
assert clean_captcha_text("") == ""
assert clean_captcha_text(None) == ""
def test_solve_gdt_captcha_fallback_cascade(monkeypatch):
"""Kiểm tra cơ chế Multi-Tier Fallback Cascade giải mã Captcha."""
from app.modules.invoices.application.invoice_processing.gdt_captcha import InvoiceGdtCaptchaMixin
solver = InvoiceGdtCaptchaMixin()
# Mock fetch SVG
mock_svg = ""
mock_key = "test_key_123"
monkeypatch.setattr(solver, "_render_svg_to_png", lambda svg: b"dummy_png_bytes")
# 1. Tier A/B fails, Tier C succeeds
monkeypatch.setattr(solver, "_solve_with_openrouter", lambda uri, key: (None, None))
monkeypatch.setattr(
solver,
"_solve_with_gemini_direct",
lambda uri, key, url: ("CASC78", "Direct Gemini [gemini-3.6-flash]"),
)
class DummyResponse:
def __enter__(self):
return self
def __exit__(self, *args):
pass
def read(self):
return b'{"key": "test_key_123", "content": ""}'
import urllib.request
monkeypatch.setattr(
urllib.request, "urlopen", lambda req, *args, **kwargs: DummyResponse()
)
ckey, cval = solver._solve_gdt_captcha()
assert ckey == "test_key_123"
assert cval == "CASC78"
def test_gdt_test_connection_api_endpoint(monkeypatch):
"""Kiểm tra API /sync/test-connection hoạt động chuẩn xác."""
from app.modules.invoices.presentation.invoice_modules import invoices_sync
# Mock _authenticate_gdt_portal to return success
mock_auth = lambda tax_code, username, password, extra_config=None: {
"status": "success",
"token": "mock_jwt_token_123",
"solved_captcha": "TEST88",
"message": "Xác thực và vượt Captcha Cổng Thuế thành công.",
}
monkeypatch.setattr(service, "_authenticate_gdt_portal", mock_auth)
monkeypatch.setattr(
invoices_sync.invoice_service, "_authenticate_gdt_portal", mock_auth
)
resp = client.post(
"/v1/erp/invoices/sync/test-connection",
json={
"service_type": "gdt_portal",
"tax_code": "0202111150",
"username": "0202111150",
"password": "valid_pwd",
"extra_config": {},
},
)
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "success"
assert "Đã vượt Captcha tự động" in data["message"]
def test_persist_gdt_raw_invoice_success_and_deduplication():
"""Kiểm tra parse và lưu trữ bản ghi hóa đơn từ API Cổng Tổng Cục Thuế vào CSDL."""
raw_inv = {
"nbmst": "0201805660",
"nbten": "CÔNG TY TNHH THƯƠNG MẠI TRUNG KIÊN",
"nbdchi": "Hải Phòng",
"nmmst": "0202111150",
"nmten": "CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN",
"nmdchi": "Thôn Tú Đôi 3, xã Nghi Dương, TP Hải Phòng",
"khhdon": "1C26TDS",
"shdon": "00088899",
"khmshdon": "1",
"tdlap": "2026-08-20T10:00:00+07:00",
"tgtttbso": 50000000.0,
"tgtthue": 4000000.0,
"tgtttbthue": 46000000.0,
"tthai": 1,
"hdhhdvu": [
{
"stt": 1,
"ten": "Cát vàng san lấp",
"mhhdvu": "CAT-01",
"dvtinh": "m3",
"sluong": 200,
"dgia": 230000,
"thtien": 46000000,
"ltsuat": "8%",
"tthue": 3680000,
}
],
}
res1 = service._persist_gdt_raw_invoice(
raw_inv, fallback_direction="input", my_tax_code="0202111150"
)
assert res1 == "new"
# Persist duplicate
res2 = service._persist_gdt_raw_invoice(
raw_inv, fallback_direction="input", my_tax_code="0202111150"
)
assert res2 == "dup"
# Cleanup
with service.get_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"DELETE FROM erp_invoices WHERE invoice_number = '00088899' AND invoice_series = '1C26TDS';"
)
conn.commit()
def test_clean_captcha_text_resilience():
"""Kiểm tra làm sạch mã Captcha chống lỗi các dạng phản hồi hội thoại và suy luận từ LLM."""
from app.modules.invoices.application.invoice_processing.gdt_captcha import clean_captcha_text
cases = {
"B5QHX8": "B5QHX8",
"Captcha: B5QHX8": "B5QHX8",
"The characters are: B5QHX8": "B5QHX8",
"The characters in this captcha are B5QHX8.": "B5QHX8",
"```\nB5QHX8\n```": "B5QHX8",
"B5QHX8\n": "B5QHX8",
"Result is B5QHX8": "B5QHX8",
"The code in the image appears to be 67WEPB.": "67WEPB",
"Thinking process...\nFinal Answer: 34G7FM": "34G7FM",
"MG9": "MG9",
"Captcha: MG9": "MG9",
}
for raw_input, expected in cases.items():
cleaned = clean_captcha_text(raw_input)
assert cleaned == expected, f"Failed for {raw_input!r}: got {cleaned!r} != {expected!r}"
assert 3 <= len(cleaned) <= 8