""" Integration & Unit Test Suite for Inverse-BIM Constraint Solver and Tri-Source Truth Grounding Engine. Real-world verification using Ben Kem box culvert (Cống hộp Bến Kem) dossier: - CAD Vector Geometry: 02. Thiet ke dong khung.dxf & 03. kc cong.pdf - BBS Structural Rebar Schedule: Page 14 of 03. kc cong.pdf - Statutory Tender BoQ / HSMT: cống hộp bến kem.21.02.2026.TĐ.xlsx & Biểu mẫu mời thầu.xlsx """ from __future__ import annotations from decimal import Decimal from pathlib import Path from fastapi.testclient import TestClient from app.main import create_app from app.modules.takeoff.application.drawing_takeoff_service import ( get_drawing_takeoff_service, ) from app.modules.takeoff.application.inverse_bim_constraint_solver import ( InverseBimConstraintSolver, ) from app.modules.takeoff.application.tri_source_truth_grounding import ( TriSourceTruthGrounding, ) client = TestClient(create_app()) AUTH_HEADERS = {"Authorization": "Bearer dev-test-token"} BEN_KEM_BASE = Path("CongTrinh/HĐ-2026/Kiến Minh-2026/Hệ thống cống hộp kênh bến Kem. ok") BEN_KEM_DXF = BEN_KEM_BASE / "BV + DT MUONG THOAT ben kem/02. Thiet ke dong khung.dxf" BEN_KEM_KC_PDF = BEN_KEM_BASE / "Đấu Thầu/File_IB2600072041/Bản vẽ/03. kc cong.pdf" BEN_KEM_TD_XLSX = BEN_KEM_BASE / "Thẩm định/cống hộp bến kem.21.02.2026.TĐ.xlsx" BEN_KEM_BM_XLSX = BEN_KEM_BASE / "Đấu Thầu/Biểu mẫu mời thầu.xlsx" def test_ben_kem_dataset_files_presence(): """Verify all real-world Ben Kem dossier files exist in the DSCons workspace.""" assert BEN_KEM_BASE.exists(), "Thư mục dự án cống hộp Bến Kem phải tồn tại" assert BEN_KEM_DXF.exists(), "File CAD DXF 02. Thiet ke dong khung.dxf phải tồn tại" assert BEN_KEM_KC_PDF.exists(), "File PDF kết cấu cống 03. kc cong.pdf phải tồn tại" assert BEN_KEM_TD_XLSX.exists(), "File Excel dự toán thẩm định phải tồn tại" assert BEN_KEM_BM_XLSX.exists(), "File Excel biểu mẫu mời thầu phải tồn tại" def test_inverse_bim_constraint_extraction_from_cad(): """Test Inverse-BIM extraction of parametric constraints from CAD vector geometry.""" solver = InverseBimConstraintSolver() model = solver.extract_parametric_constraints(BEN_KEM_DXF) assert model.structure_type == "box_culvert" assert "Bến Kem" in model.structure_name assert model.num_segments == 5 assert model.total_length_m == Decimal("215.0000") assert len(model.segments) == 5 # Verify cross section constraints cs = model.segments[0].cross_section assert cs.clear_span_m == Decimal("2.5000") assert cs.clear_height_m == Decimal("2.5000") assert cs.wall_thickness_m == Decimal("0.3000") assert cs.bottom_slab_thickness_m == Decimal("0.2500") assert cs.top_slab_thickness_m == Decimal("0.2500") assert cs.bottom_slab_width_m == Decimal("3.4000") assert cs.lean_concrete_thickness_m == Decimal("0.1000") assert cs.lean_concrete_width_m == Decimal("3.4000") assert cs.bamboo_pile_density_per_m2 == Decimal("25.0000") assert cs.bamboo_pile_length_m == Decimal("2.5000") assert cs.haunch_size_m == Decimal("0.1500") def test_inverse_bim_deterministic_solid_quantities(): """Test closed-form solid geometry calculations without AI hallucinations.""" solver = InverseBimConstraintSolver() model = solver.extract_parametric_constraints(BEN_KEM_DXF) solids = solver.compute_deterministic_solids(model) # 1. Foundation: Lean concrete 73.1 m3, Bamboo piles 456.875 (100m) assert solids.lean_concrete_volume_m3 == Decimal("73.1000") assert solids.bamboo_piles_total_100m == Decimal("456.8750") assert solids.bamboo_piles_count == 18275 assert solids.lean_concrete_formwork_100m2 == Decimal("0.4640") # 2. Bottom slab: 182.75 m3, formwork 1.16 (100m2) assert solids.bottom_slab_concrete_volume_m3 == Decimal("182.7500") assert solids.bottom_slab_formwork_100m2 == Decimal("1.1600") # 3. Walls: 280.1925 m3, formwork 21.6250 (100m2) assert solids.wall_concrete_volume_m3 == Decimal("280.1925") assert solids.wall_formwork_100m2 == Decimal("21.6250") # 4. Top slab / deck: 134.375 m3, formwork 6.5350 (100m2) assert solids.top_slab_concrete_volume_m3 == Decimal("134.3750") assert solids.top_slab_formwork_100m2 == Decimal("6.5350") # 5. Joint Waterstop: 207.55 m2 assert solids.waterstop_m2 == Decimal("207.5500") # 6. Inlet / Outlet wing walls assert solids.wing_bamboo_piles_100m == Decimal("15.7450") assert solids.wing_lean_concrete_m3 == Decimal("2.5200") assert solids.wing_bottom_slab_concrete_m3 == Decimal("12.8340") assert solids.wing_wall_concrete_m3 == Decimal("2.5938") # 7. Earthwork and Larsen sheet piles assert solids.slush_mud_suction_m3 == Decimal("1152.9000") assert solids.foundation_excavation_100m3 == Decimal("4.8853") assert solids.mass_excavation_100m3 == Decimal("16.4143") assert solids.soil_haul_100m3 == Decimal("16.4140") assert solids.soil_backfill_100m3 == Decimal("14.1032") assert solids.mountain_soil_material_m3 == Decimal("1551.3520") assert solids.larsen_pile_driving_100m == Decimal("48.0400") assert solids.larsen_pile_rental_m == Decimal("2882.4000") assert solids.water_pumping_shifts == Decimal("75.0000") assert solids.ground_mat_segments == Decimal("5.0000") def test_bbs_rebar_schedule_extraction(): """Test extraction of Bar Bending Schedule (BBS) rebar items from drawing sheet 14.""" engine = TriSourceTruthGrounding() bbs_items = engine.extract_bbs_schedule(BEN_KEM_KC_PDF) assert len(bbs_items) >= 4 bbs_by_name = {b.component_name: b for b in bbs_items} # 1. Đáy cống: 10.3291 t assert "Đáy cống" in bbs_by_name assert bbs_by_name["Đáy cống"].total_weight_ton == Decimal("10.3291") # 2. Thành cống: 22.3225 t assert "Thành cống" in bbs_by_name assert bbs_by_name["Thành cống"].total_weight_ton == Decimal("22.3225") # 3. Bản mặt cống: 9.3036 t assert "Bản mặt cống" in bbs_by_name assert bbs_by_name["Bản mặt cống"].total_weight_ton == Decimal("9.3036") # 4. Cửa cống: 0.2324 t assert "Cửa cống" in bbs_by_name assert bbs_by_name["Cửa cống"].total_weight_ton == Decimal("0.2324") # Total rebar: 42.1876 t total_rebar_ton = sum(b.total_weight_ton for b in bbs_items) assert total_rebar_ton == Decimal("42.1876") def test_hsmt_tender_items_extraction(): """Test extraction of official statutory tender BoQ items from approved estimation.""" engine = TriSourceTruthGrounding() hsmt_items = engine.extract_hsmt_tender_items(BEN_KEM_TD_XLSX) assert len(hsmt_items) == 39 assert hsmt_items[0]["stt"] == 1 assert hsmt_items[0]["unit"] == "ca" assert hsmt_items[1]["norm_code"] == "AC.22111" assert hsmt_items[1]["quantity"] == Decimal("48.0400") # Total tender bid value (after-tax G_XD = 6,759,733,141 VND) total_bid = sum(i["total_amount"] for i in hsmt_items) assert abs(total_bid - Decimal("6759733141.0")) < Decimal("100.0") def test_tri_source_reconciliation_ben_kem_ground_truth(): """Test 100% reconciliation cross-audit on Ben Kem dataset with zero discrepancies.""" service = get_drawing_takeoff_service() report = service.run_tri_source_reconciliation( cad_source=BEN_KEM_DXF, bbs_source=BEN_KEM_KC_PDF, hsmt_source=BEN_KEM_TD_XLSX, ) assert report.total_items_audited == 39 assert report.matched_items_count == 39 assert report.discrepancy_items_count == 0 assert report.overall_confidence_score == 100 assert report.confidence_level == "tri_source_grounded_100" # Verify key individual work item matchings rec_by_order = {item.item_order: item for item in report.reconciliation_items} # Item 14: Đóng cọc tre móng cống (456.875 100m) it14 = rec_by_order[14] assert it14.norm_code == "AC.11121" assert it14.source_cad_geometry_qty == Decimal("456.8750") assert it14.source_hsmt_tender_qty == Decimal("456.8750") assert it14.variance_delta == Decimal("0.0000") assert it14.status == "MATCHED_100%" # Item 17: Bê tông lót móng M100 (73.1 m3) it17 = rec_by_order[17] assert it17.source_cad_geometry_qty == Decimal("73.1000") assert it17.source_hsmt_tender_qty == Decimal("73.1000") assert it17.variance_delta == Decimal("0.0000") assert it17.status == "MATCHED_100%" # Item 18: Cốt thép móng (10.3291 t from BBS) it18 = rec_by_order[18] assert it18.source_bbs_rebar_qty == Decimal("10.3291") assert it18.source_hsmt_tender_qty == Decimal("10.3291") assert it18.variance_delta == Decimal("0.0000") assert it18.status == "MATCHED_100%" # Item 20: Bê tông móng M250 (182.75 m3) it20 = rec_by_order[20] assert it20.source_cad_geometry_qty == Decimal("182.7500") assert it20.source_hsmt_tender_qty == Decimal("182.7500") assert it20.variance_delta == Decimal("0.0000") assert it20.status == "MATCHED_100%" # Item 21: Cốt thép tường cống (22.3225 t from BBS) it21 = rec_by_order[21] assert it21.source_bbs_rebar_qty == Decimal("22.3225") assert it21.source_hsmt_tender_qty == Decimal("22.3225") assert it21.variance_delta == Decimal("0.0000") assert it21.status == "MATCHED_100%" # Item 23: Bê tông tường M250 (280.1925 m3) it23 = rec_by_order[23] assert it23.source_cad_geometry_qty == Decimal("280.1925") assert it23.source_hsmt_tender_qty == Decimal("280.1925") assert it23.variance_delta == Decimal("0.0000") assert it23.status == "MATCHED_100%" # Item 26: Cốt thép bản mặt cống (9.3036 t from BBS) it26 = rec_by_order[26] assert it26.source_bbs_rebar_qty == Decimal("9.3036") assert it26.source_hsmt_tender_qty == Decimal("9.3036") assert it26.variance_delta == Decimal("0.0000") assert it26.status == "MATCHED_100%" # Item 27: Bê tông bản mặt cống (134.375 m3) it27 = rec_by_order[27] assert it27.source_cad_geometry_qty == Decimal("134.3750") assert it27.source_hsmt_tender_qty == Decimal("134.3750") assert it27.variance_delta == Decimal("0.0000") assert it27.status == "MATCHED_100%" def test_tri_source_reconciliation_anomaly_detection(): """Negative testing: Injected discrepancies must be flagged immediately by the engine.""" engine = TriSourceTruthGrounding() solver = InverseBimConstraintSolver() model = solver.extract_parametric_constraints(BEN_KEM_DXF) solids = solver.compute_deterministic_solids(model) bbs_items = engine.extract_bbs_schedule(BEN_KEM_KC_PDF) hsmt_items = engine.extract_hsmt_tender_items(BEN_KEM_TD_XLSX) # Corrupt 2 items intentionally in HSMT: # 1. Change lean concrete from 73.1 to 95.0 m3 for itm in hsmt_items: if itm["stt"] == 17: itm["quantity"] = Decimal("95.0000") elif itm["stt"] == 21: # Corrupt wall rebar from 22.3225 to 18.0 t itm["quantity"] = Decimal("18.0000") report = engine.reconcile_tri_source( solids=solids, bbs_items=bbs_items, hsmt_items=hsmt_items, ) assert report.discrepancy_items_count == 2 assert report.matched_items_count == 37 assert report.confidence_level == "discrepancies_detected" assert report.overall_confidence_score < 100 rec_by_order = {item.item_order: item for item in report.reconciliation_items} assert rec_by_order[17].status == "DISCREPANCY_FLAGGED" assert rec_by_order[17].variance_delta == Decimal("73.1000") - Decimal("95.0000") assert rec_by_order[21].status == "DISCREPANCY_FLAGGED" assert rec_by_order[21].variance_delta == Decimal("22.3225") - Decimal("18.0000") def test_api_tri_source_endpoints(): """Test REST API presentation layer endpoints for Tri-Source reconciliation.""" # 1. GET /v1/takeoff/ben-kem/tri-source-audit resp = client.get("/v1/takeoff/ben-kem/tri-source-audit", headers=AUTH_HEADERS) assert resp.status_code == 200, resp.text data = resp.json() assert data["status"] == "success" report = data["report"] assert report["total_items_audited"] == 39 assert report["matched_items_count"] == 39 assert report["overall_confidence_score"] == 100 # 2. POST /v1/takeoff/inverse-bim/solve solve_resp = client.post( "/v1/takeoff/inverse-bim/solve", json={ "cad_source_path": str(BEN_KEM_DXF), "num_segments": 5, "segment_length_m": 43.0, "structure_name": "Cống Bến Kem API Test", }, headers=AUTH_HEADERS, ) assert solve_resp.status_code == 200, solve_resp.text solve_data = solve_resp.json() assert solve_data["status"] == "success" assert solve_data["solids"]["lean_concrete_volume_m3"] == "73.1000" assert solve_data["solids"]["bottom_slab_concrete_volume_m3"] == "182.7500" # 3. POST /v1/takeoff/tri-source/reconcile recon_resp = client.post( "/v1/takeoff/tri-source/reconcile", json={ "cad_source_path": str(BEN_KEM_DXF), "bbs_source_path": str(BEN_KEM_KC_PDF), "hsmt_source_path": str(BEN_KEM_TD_XLSX), }, headers=AUTH_HEADERS, ) assert recon_resp.status_code == 200, recon_resp.text recon_data = recon_resp.json() assert recon_data["status"] == "success" assert recon_data["report"]["matched_items_count"] == 39 def test_tri_source_reconciliation_with_procurement_tender_sheet(): """Verify 100% reconciliation against the statutory 38-item procurement tender sheet (Biểu mẫu mời thầu.xlsx).""" service = get_drawing_takeoff_service() report = service.run_tri_source_reconciliation( cad_source=BEN_KEM_DXF, bbs_source=BEN_KEM_KC_PDF, hsmt_source=BEN_KEM_BM_XLSX, ) assert report.total_items_audited == 38 assert report.matched_items_count == 38 assert report.discrepancy_items_count == 0 assert report.overall_confidence_score == 100 assert report.confidence_level == "tri_source_grounded_100" def test_parametric_constraint_scaling_non_hardcoded(): """Verify that Inverse-BIM constraint solver dynamically computes quantities without hardcoded values.""" solver = InverseBimConstraintSolver() # Baseline 215m model base_model = solver.extract_parametric_constraints(BEN_KEM_DXF) base_solids = solver.compute_deterministic_solids(base_model) # Scaled model: 3 segments of 20m = 60m total length scaled_model = solver.extract_parametric_constraints( BEN_KEM_DXF, num_segments=3, segment_length_m=20.0, ) scaled_solids = solver.compute_deterministic_solids(scaled_model) assert scaled_model.total_length_m == Decimal("60.0000") assert scaled_model.num_segments == 3 # Scaled quantities must be dynamically computed and differ from 215m baseline assert scaled_solids.lean_concrete_volume_m3 != base_solids.lean_concrete_volume_m3 assert scaled_solids.bottom_slab_concrete_volume_m3 != base_solids.bottom_slab_concrete_volume_m3 assert scaled_solids.wall_concrete_volume_m3 != base_solids.wall_concrete_volume_m3 assert scaled_solids.top_slab_concrete_volume_m3 != base_solids.top_slab_concrete_volume_m3 assert scaled_solids.larsen_pile_driving_100m != base_solids.larsen_pile_driving_100m # Exact verification of scaled lean concrete: 60m * 3.4m * 0.1m = 20.4 m3 assert scaled_solids.lean_concrete_volume_m3 == Decimal("20.4000") # Bottom slab: 60m * 3.4m * 0.25m = 51.0 m3 assert scaled_solids.bottom_slab_concrete_volume_m3 == Decimal("51.0000") # Top slab: 60m * 2.5m * 0.25m = 37.5 m3 assert scaled_solids.top_slab_concrete_volume_m3 == Decimal("37.5000") def test_unmapped_item_handling_reference_only(): """Verify that unmapped HSMT items are marked REFERENCE_ONLY and do not falsely inflate matched_count.""" engine = TriSourceTruthGrounding() solver = InverseBimConstraintSolver() model = solver.extract_parametric_constraints(BEN_KEM_DXF) solids = solver.compute_deterministic_solids(model) bbs_items = engine.extract_bbs_schedule(BEN_KEM_KC_PDF) fake_hsmt = [ { "stt": 999, "norm_code": "CUSTOM.UNKNOWN", "item_name": "Công tác thí nghiệm hiện trường không đo bóc từ CAD", "unit": "gói", "quantity": Decimal("1.0000"), "bid_rate": Decimal("50000000.00"), "total_amount": Decimal("50000000.00"), } ] report = engine.reconcile_tri_source( solids=solids, bbs_items=bbs_items, hsmt_items=fake_hsmt, ) assert report.total_items_audited == 1 assert report.matched_items_count == 0 # Crucial: Must NOT falsely count as matched! assert report.discrepancy_items_count == 0 assert report.reconciliation_items[0].status == "REFERENCE_ONLY"