from __future__ import annotations """Unit and Integration tests for Flexible Warehouse Dispatch and Material Wastage Analytics. Adheres to DSCons Rule 1 (Zero Synthetic Data in prod, clean test teardown), Rule 2 (Regression Gate 100% Passed), and Decimal(18, 4) financial precision. """ from decimal import Decimal import uuid import pytest from app.core.postgres.base_pkg.base_client import BasePostgresClient from app.modules.inventory.application.warehouse_dispatch_service import WarehouseDispatchService from app.modules.inventory.domain.dispatch_entities import ( NORM_WASTAGE_PERCENTAGES, VarianceAlertStatus, ) @pytest.fixture def db_client(): return BasePostgresClient() @pytest.fixture def dispatch_service(db_client): return WarehouseDispatchService(postgres_client=db_client) def test_norm_wastage_percentages_construction_rules(): """Kiểm tra bảng định mức hao hụt chuẩn theo Thông tư 12/2021 & Định mức 2026.""" assert NORM_WASTAGE_PERCENTAGES["STEEL"] == Decimal("2.5") assert NORM_WASTAGE_PERCENTAGES["CONCRETE"] == Decimal("2.0") assert NORM_WASTAGE_PERCENTAGES["AGGREGATE"] == Decimal("4.0") assert NORM_WASTAGE_PERCENTAGES["CEMENT"] == Decimal("2.5") assert NORM_WASTAGE_PERCENTAGES["BRICK"] == Decimal("3.0") def test_dispatch_validation_zero_or_negative(dispatch_service): """Kiểm tra chặn xuất kho với số lượng <= 0.""" with pytest.raises(ValueError, match="lớn hơn 0"): dispatch_service.create_dispatch( inventory_item_id=str(uuid.uuid4()), quantity_dispatched=0, ) with pytest.raises(ValueError, match="lớn hơn 0"): dispatch_service.create_dispatch( inventory_item_id=str(uuid.uuid4()), quantity_dispatched=-5.5, ) def test_dispatch_non_existent_item(dispatch_service): """Kiểm tra xử lý khi ID lô vật tư không tồn tại trong kho.""" random_id = str(uuid.uuid4()) with pytest.raises(ValueError, match="Không tìm thấy lô vật tư"): dispatch_service.create_dispatch( inventory_item_id=random_id, quantity_dispatched=10.0, ) def test_flexible_dispatch_lifecycle(db_client, dispatch_service): """Kiểm tra toàn bộ vòng đời xuất kho linh hoạt: 1. Tạo lô vật tư test tạm thời. 2. Xuất kho một phần (không cần WBS hay Dự án - tính linh hoạt). 3. Kiểm tra số dư còn lại (quantity_remaining) và trạng thái ĐÃ_XUẤT_MỘT_PHẦN. 4. Thử xuất vượt quá tồn kho còn lại -> Bị chặn (chống xuất âm). 5. Xuất hết phần còn lại -> Trạng thái ĐÃ_XUẤT_HẾT. 6. Dọn dẹp sạch sẽ dữ liệu test (Teardown). """ test_item_id = str(uuid.uuid4()) company_id = None with db_client.get_connection() as conn: with conn.cursor() as cur: # Lấy company_id thực tế đầu tiên cur.execute("SELECT id FROM companies LIMIT 1;") c_row = cur.fetchone() if c_row: company_id = c_row["id"] if not company_id: pytest.skip("Chưa có company trong database để chạy integration test.") # 1. Setup: Tạo 1 lô thép 20.0 tấn with db_client.get_connection() as conn: with conn.cursor() as cur: cur.execute(""" INSERT INTO erp_warehouse_inventory ( id, company_id, item_code, item_name, material_group, unit, quantity_received, quantity_remaining, unit_price_vnd, total_value_vnd, status ) VALUES ( %s, %s, 'TEST-THEP-CB400', 'Thép vằn D16 Hòa Phát CB400 (Test)', 'STEEL', 'Tấn', 20.0000, 20.0000, 15500000.0000, 310000000.0000, 'CHỜ_XUẤT_CÔNG_TRƯỜNG' ); """, (test_item_id, company_id)) conn.commit() try: # 2. Xuất 6.5 tấn ra xe tải chở ra công trường (không truyền WBS) disp1 = dispatch_service.create_dispatch( inventory_item_id=test_item_id, quantity_dispatched=6.5, recipient_name="Nguyễn Văn Hùng (Chỉ huy trưởng)", transport_vehicle="15C-123.45", purpose="Gia công cốt thép dầm sàn đợt 1", notes="Xuất cấp bách phục vụ đổ bê tông ca đêm", ) assert disp1["quantity_dispatched"] == 6.5 assert disp1["quantity_remaining_after"] == 13.5 assert disp1["status"] == "ĐÃ_XUẤT_MỘT_PHẦN" assert disp1["dispatch_code"].startswith("XK-") assert disp1["total_value_vnd"] == 6.5 * 15500000.0 # 3. Thử xuất 15.0 tấn trong khi kho chỉ còn 13.5 tấn -> Bắt buộc báo lỗi with pytest.raises(ValueError, match="không đủ để xuất"): dispatch_service.create_dispatch( inventory_item_id=test_item_id, quantity_dispatched=15.0, ) # 4. Xuất nốt 13.5 tấn còn lại disp2 = dispatch_service.create_dispatch( inventory_item_id=test_item_id, quantity_dispatched=13.5, recipient_name="Trần Văn Nam", purpose="Gia công móng đợt 2", ) assert disp2["quantity_dispatched"] == 13.5 assert disp2["quantity_remaining_after"] == 0.0 assert disp2["status"] == "ĐÃ_XUẤT_HẾT" # 5. Kiểm tra danh sách lịch sử xuất kho dispatches, total = dispatch_service.list_dispatches(inventory_item_id=test_item_id) assert total == 2 assert len(dispatches) == 2 assert dispatches[0]["quantity_dispatched"] == 13.5 assert dispatches[1]["quantity_dispatched"] == 6.5 finally: # 6. Strict Teardown: Xóa toàn bộ phiếu xuất và lô test tạm thời with db_client.get_connection() as conn: with conn.cursor() as cur: cur.execute("DELETE FROM erp_warehouse_dispatches WHERE inventory_item_id = %s;", (test_item_id,)) cur.execute("DELETE FROM erp_warehouse_inventory WHERE id = %s;", (test_item_id,)) conn.commit() def test_variance_report_calculation_logic(dispatch_service): """Kiểm tra nghiệp vụ phân tích và cảnh báo hao hụt 3 chiều.""" report = dispatch_service.calculate_material_variance_report(project_id=None) assert "total_materials_monitored" in report assert "total_dispatched_value_vnd" in report assert "total_wastage_cost_vnd" in report assert "average_wastage_pct" in report assert "items" in report assert isinstance(report["items"], list) # Kiểm tra cấu trúc từng item trong bảng báo cáo hao hụt for itm in report["items"]: assert "material_group" in itm assert "estimated_quantity" in itm assert "settled_quantity" in itm assert "dispatched_quantity" in itm assert "norm_wastage_pct" in itm assert "actual_wastage_quantity" in itm assert "actual_wastage_pct" in itm assert itm["variance_status"] in [ VarianceAlertStatus.SAFE.value, VarianceAlertStatus.WARNING.value, VarianceAlertStatus.EXCEEDED.value, VarianceAlertStatus.NO_ESTIMATE.value, ] def test_api_endpoints_contracts(): """Kiểm tra hợp đồng API HTTP: GET /dispatches và GET /variance-report.""" from fastapi import FastAPI from fastapi.testclient import TestClient from app.modules.inventory.presentation.inventory_truth_routes import router as inventory_router from app.modules.auth.presentation.auth import get_current_user test_app = FastAPI() test_app.include_router(inventory_router) test_app.dependency_overrides[get_current_user] = lambda: { "id": "admin-test-id", "email": "admin@dscons.vn", "role": "superadmin", } client = TestClient(test_app) # 1. Test GET /v1/erp/inventory-truth/dispatches resp_disp = client.get("/v1/erp/inventory-truth/dispatches") assert resp_disp.status_code == 200 data_disp = resp_disp.json() assert data_disp["status"] == "success" assert "items" in data_disp assert "total_count" in data_disp # 2. Test GET /v1/erp/inventory-truth/variance-report resp_var = client.get("/v1/erp/inventory-truth/variance-report") assert resp_var.status_code == 200 data_var = resp_var.json() assert data_var["status"] == "success" assert "report" in data_var assert "items" in data_var["report"] # 3. Test POST /v1/erp/inventory-truth/dispatch với dữ liệu sai -> báo lỗi 400 resp_bad = client.post( "/v1/erp/inventory-truth/dispatch", json={"inventory_item_id": str(uuid.uuid4()), "quantity_dispatched": 0} ) assert resp_bad.status_code == 422 # Pydantic gt=0 validation error