from __future__ import annotations """Unit & Integration Tests for BoQ Unit Rate Breakdown (VL+NC+M), Triple-Column Variance, and Material Requisitions.""" import uuid import pytest from fastapi.testclient import TestClient from app.core.postgres.erp_client import ErpDatabaseClient from app.main import app from app.modules.takeoff.application.boq_rate_breakdown_service import ( BENCHMARK_PRICES, BoqRateBreakdownEngine, ) AUTH_HEADERS = {"Authorization": "Bearer dev-test-token"} @pytest.fixture def test_takeoff_with_items(): client = ErpDatabaseClient() takeoff_id = str(uuid.uuid4()) item_id_1 = str(uuid.uuid4()) item_id_2 = str(uuid.uuid4()) item_id_3 = str(uuid.uuid4()) # Create parent takeoff client.create_drawing_takeoff( { "id": takeoff_id, "drawing_code": "TEST-DWG-001", "drawing_title": "Bản vẽ Cống Hộp Thí Nghiệm", "drawing_type": "HYDRAULIC_CULVERT", "total_estimated_cost_vnd": 45000000.0, } ) # Create 3 items: Concrete, Rebar, Formwork client.create_drawing_takeoff_item( { "id": item_id_1, "takeoff_id": takeoff_id, "item_order": 1, "wbs_code": "1.1", "norm_code": "AF.11110", "item_name": "Bê tông lót móng cống M150 đá 1x2", "category": "concrete", "dimension_formula": "2.5 * 10.0 * 0.1", "unit": "m3", "quantity": 2.5, "unit_price_vnd": 1250000.0, "total_amount_vnd": 3125000.0, } ) client.create_drawing_takeoff_item( { "id": item_id_2, "takeoff_id": takeoff_id, "item_order": 2, "wbs_code": "1.2", "norm_code": "AF.61110", "item_name": "Cốt thép cống hộp CB400", "category": "rebar", "dimension_formula": "1.2 * 1.05", "unit": "tấn", "quantity": 1.26, "unit_price_vnd": 19500000.0, "total_amount_vnd": 24570000.0, } ) client.create_drawing_takeoff_item( { "id": item_id_3, "takeoff_id": takeoff_id, "item_order": 3, "wbs_code": "1.3", "norm_code": "AF.81110", "item_name": "Ván khuôn thành cống phủ phim", "category": "formwork", "dimension_formula": "2 * (2.0 + 1.5) * 10.0", "unit": "m2", "quantity": 70.0, "unit_price_vnd": 245000.0, "total_amount_vnd": 17150000.0, } ) yield { "takeoff_id": takeoff_id, "item_ids": [item_id_1, item_id_2, item_id_3], } # Strict Rule 12 Teardown try: conn = client.get_connection() with conn.cursor() as cur: cur.execute( "DELETE FROM erp_drawing_takeoff_items WHERE takeoff_id = %s;", (takeoff_id,), ) cur.execute( "DELETE FROM erp_drawing_takeoffs WHERE id = %s;", (takeoff_id,) ) conn.commit() conn.close() except Exception as e: print(f"Teardown error: {e}") def test_boq_engine_rate_breakdown_concrete(): """Kiểm tra phân rã đơn giá bê tông tuân thủ Cát vàng Hải Phòng 710k (Rule 11) và đá 1x2, xi măng.""" res = BoqRateBreakdownEngine.breakdown_item_rate( item_name="Bê tông đài móng M250", category="concrete", unit="m3", total_unit_price=1350000.0, ) assert res["material_cost_vnd"] > 0 assert res["labor_cost_vnd"] > 0 assert res["equipment_cost_vnd"] > 0 assert len(res["material_breakdown"]) >= 3 # Verify sand material benchmark price in breakdown sand_mat = next( (m for m in res["material_breakdown"] if "CAT" in m["material_code"]), None ) assert sand_mat is not None assert sand_mat["unit_price_vnd"] == BENCHMARK_PRICES["CAT_VANG_BETONG"] assert sand_mat["unit_price_vnd"] == 710000.0 def test_boq_engine_rate_breakdown_rebar(): """Kiểm tra phân rã đơn giá cốt thép tuân thủ Thép CB400.""" res = BoqRateBreakdownEngine.breakdown_item_rate( item_name="Cốt thép dầm móng CB400", category="rebar", unit="tấn", total_unit_price=19500000.0, ) assert res["material_cost_vnd"] > 0 thep_mat = next( (m for m in res["material_breakdown"] if "THEP" in m["material_code"]), None ) assert thep_mat is not None assert thep_mat["unit_price_vnd"] == 15500000.0 def test_boq_engine_triple_variance_calculation(): """Kiểm tra tính toán sai lệch 3 cột và phân loại Variation Order (VO).""" # 1. Normal (<= 5%) res_normal = BoqRateBreakdownEngine.compute_triple_variance( design_quantity=100.0, hsmt_quantity=98.0 ) assert res_normal["variation_status"] == "normal" assert res_normal["variance_quantity"] == 2.0 assert res_normal["variance_percentage"] == round((2.0 / 98.0) * 100, 2) # 2. VO Increase (> 5%) res_increase = BoqRateBreakdownEngine.compute_triple_variance( design_quantity=120.0, hsmt_quantity=100.0 ) assert res_increase["variation_status"] == "vo_increase" assert res_increase["variance_percentage"] == 20.0 # 3. VO Decrease (< -5%) res_decrease = BoqRateBreakdownEngine.compute_triple_variance( design_quantity=80.0, hsmt_quantity=100.0 ) assert res_decrease["variation_status"] == "vo_decrease" assert res_decrease["variance_percentage"] == -20.0 def test_api_get_rate_breakdown(test_takeoff_with_items): """Kiểm tra endpoint GET /v1/takeoff/{id}/rate-breakdown.""" client = TestClient(app) takeoff_id = test_takeoff_with_items["takeoff_id"] res = client.get(f"/v1/takeoff/{takeoff_id}/rate-breakdown", headers=AUTH_HEADERS) assert res.status_code == 200 data = res.json() assert data["status"] == "success" assert data["count"] == 3 assert "totals" in data assert data["totals"]["total_material_cost_vnd"] > 0 assert data["totals"]["total_labor_cost_vnd"] > 0 assert data["totals"]["total_equipment_cost_vnd"] > 0 def test_api_get_triple_variance_and_update(test_takeoff_with_items): """Kiểm tra endpoint GET /v1/takeoff/{id}/triple-column-variance và PUT /v1/takeoff/items/{id}/quantities.""" client = TestClient(app) takeoff_id = test_takeoff_with_items["takeoff_id"] item_id = test_takeoff_with_items["item_ids"][0] # Update item HSMT quantity to create a VO increase (> 5%) # Design qty is 2.5, HSMT is 2.0 -> +25% put_res = client.put( f"/v1/takeoff/items/{item_id}/quantities", headers=AUTH_HEADERS, json={"hsmt_quantity": 2.0, "actual_quantity": 2.6}, ) assert put_res.status_code == 200 updated_item = put_res.json()["item"] assert float(updated_item["hsmt_quantity"]) == 2.0 assert updated_item["variation_status"] == "vo_increase" # Get triple variance summary get_res = client.get( f"/v1/takeoff/{takeoff_id}/triple-column-variance", headers=AUTH_HEADERS ) assert get_res.status_code == 200 v_data = get_res.json() assert v_data["status"] == "success" assert v_data["summary"]["vo_increase_items_count"] >= 1 assert v_data["summary"]["has_variation_orders"] is True def test_api_generate_material_request(test_takeoff_with_items): """Kiểm tra endpoint POST /v1/takeoff/{id}/generate-material-request.""" client = TestClient(app) takeoff_id = test_takeoff_with_items["takeoff_id"] res = client.post( f"/v1/takeoff/{takeoff_id}/generate-material-request", headers=AUTH_HEADERS ) assert res.status_code == 200 data = res.json() assert data["status"] == "success" assert data["materials_count"] > 0 assert data["total_material_budget_vnd"] > 0 # Ensure required sand and steel exist in schedule schedule = data["material_schedule"] materials = [m["material_name"] for m in schedule] assert any( "Cát vàng" in name or "CAT" in m["material_code"] for m, name in zip(schedule, materials) ) assert any( "Thép" in name or "THEP" in m["material_code"] for m, name in zip(schedule, materials) )