from __future__ import annotations import unittest from decimal import Decimal from pathlib import Path from fastapi.testclient import TestClient from app.modules.auth.presentation.auth import get_current_user from app.core.postgres.erp_client import ErpDatabaseClient from app.main import create_app from app.modules.takeoff.application.geometric_breakdown_engine import ( GeometricBreakdownEngine, ) from app.modules.takeoff.application.takeoff_training import ( GroundTruthLoader, GroundTruthRow, QSDrillEvaluator, QSRuleSynthesizer, ) class TestAIQuynhTrainingAcademy(unittest.TestCase): """Kiểm thử hệ thống bài tập tự trị và kiểm chuẩn AI Quỳnh QS (HĐ-2026 Benchmark).""" def setUp(self) -> None: self.app = create_app() self.client = TestClient(self.app) self.db_client = ErpDatabaseClient() self.test_excel_path = Path( "HĐ-2026/NVH Đại Thắng-2026/TK-DT/NVH Thôn Đại Thắng_TD2_8.7.xls" ) def tearDown(self) -> None: """Rule 12 Teardown: Thu dọn sạch sẽ toàn bộ dữ liệu kiểm thử trong DB.""" self.app.dependency_overrides = {} try: with self.db_client.get_connection() as conn: with conn.cursor() as cur: cur.execute( "DELETE FROM erp_learned_rules WHERE rule_title LIKE 'Tự Học [TEST-%' OR rule_title LIKE 'Tự Học [DA-NVH-%';" ) conn.commit() except Exception: pass def test_ground_truth_loader_extracts_real_excel_data(self) -> None: """Kiểm thử nạp dữ liệu dự toán thật từ hồ sơ HĐ-2026.""" if not self.test_excel_path.exists(): self.skipTest(f"Excel file {self.test_excel_path} not found") items = GroundTruthLoader.load_ground_truth_from_excel(self.test_excel_path) self.assertGreater(len(items), 50) # Check first item details first = items[0] self.assertEqual(first.stt, "1") self.assertEqual(first.unit, "m3") self.assertGreater(first.quantity, Decimal("0.0")) self.assertIsNotNone(first.formula_breakdown) self.assertIn("=", first.formula_breakdown) def test_geometric_breakdown_engine_formulas(self) -> None: """Kiểm thử động cơ giải toán hình học và sinh công thức phân rã chi tiết.""" # 1. Pad footing with pile deduction footing = GeometricBreakdownEngine.calculate_pad_footing( length=2.0, width=2.0, base_height=0.5, pile_count_per_footing=4, pile_side_or_dia=0.25, pile_embed_depth=0.1, count=2, ) self.assertEqual(footing["norm_code"], "AF.11111") self.assertIn("cọc", footing["formula_breakdown"]) self.assertGreater(footing["quantity"], Decimal("0.0")) # 2. Box culvert culvert = GeometricBreakdownEngine.calculate_box_culvert( length=10.0, width_inner=1.0, height_inner=1.0, wall_thick=0.2, bottom_thick=0.25, top_thick=0.2, ) self.assertEqual(culvert["norm_code"], "AF.12311") self.assertIn("m3", culvert["unit"]) # 3. Rebar weight with hooks and laps rebar = GeometricBreakdownEngine.calculate_rebar_weight( bar_diameter_mm=12.0, total_length_m=100.0, hooks_count=10, lap_count=2 ) self.assertEqual(rebar["norm_code"], "AF.61121") self.assertEqual(rebar["unit"], "tấn") # 4. Trench excavation trench = GeometricBreakdownEngine.calculate_trench_excavation( bottom_width=1.0, depth=1.5, length=20.0, slope_m=0.5 ) self.assertEqual(trench["norm_code"], "AB.11311") self.assertGreater(trench["quantity"], Decimal("0.0")) def test_drill_evaluator_and_scoring(self) -> None: """Kiểm thử bộ chấm điểm sai số (Scorecard) giữa AI Takeoff và Ground Truth.""" gt_rows = [ GroundTruthRow( stt="1", norm_code="AF.11111", task_name="Bê tông đài móng M250", unit="m3", quantity=Decimal("10.0"), ), GroundTruthRow( stt="2", norm_code="AB.11311", task_name="Đào đất hố móng", unit="m3", quantity=Decimal("50.0"), ), ] ai_items = [ { "task_name": "Bê tông đài móng M250", "norm_code": "AF.11111", "unit": "m3", "quantity": 9.9, "formula_breakdown": "2 * (2*2*0.5 - cọc) = 9.9 m3", }, { "task_name": "Đào đất hố móng", "norm_code": "AB.11311", "unit": "m3", "quantity": 48.5, "formula_breakdown": "20 * 1.5 * 1.6 = 48.5 m3", }, ] scorecard = QSDrillEvaluator.evaluate_takeoff( drill_id="DRILL-TEST-01", project_name="Dự án Test Đào Tạo", project_code="TEST-DA-01", ground_truth_items=gt_rows, ai_takeoff_items=ai_items, ) self.assertEqual(scorecard.total_ground_truth_items, 2) self.assertEqual(scorecard.matched_items, 2) self.assertEqual(scorecard.recall_rate_pct, Decimal("100.0")) self.assertLess(scorecard.volume_mae_pct, Decimal("3.5")) self.assertIn(scorecard.overall_grade, ("S", "A")) def test_rule_synthesizer_and_persistence(self) -> None: """Kiểm thử tự động sinh Learned Rules từ các sai số bóc tách.""" gt_rows = [ GroundTruthRow( stt="1", task_name="Ván khuôn móng đài cọc", unit="m2", quantity=Decimal("100.0"), ), ] scorecard = QSDrillEvaluator.evaluate_takeoff( drill_id="DRILL-TEST-RULE", project_name="Dự án Test Rule", project_code="TEST-RULE-01", ground_truth_items=gt_rows, ai_takeoff_items=[], ) self.assertEqual(scorecard.evaluated_items[0].match_status, "MISSED") synthesized = QSRuleSynthesizer.synthesize_and_save_rules( scorecard, self.db_client ) self.assertGreaterEqual(len(synthesized), 1) self.assertIn("FORMWORK", synthesized[0]["rule_category"]) def test_training_academy_api_endpoints(self) -> None: """Kiểm thử toàn bộ các API Endpoints của AI Quỳnh Training Academy.""" self.app.dependency_overrides[get_current_user] = lambda: { "id": "admin-test-id", "email": "admin@dscons.vn", "role": "superadmin", } # 1. GET /api/v1/takeoff/training/drills res_drills = self.client.get("/v1/takeoff/training/drills") self.assertEqual(res_drills.status_code, 200) drills_data = res_drills.json() self.assertGreaterEqual(len(drills_data), 1) # 2. POST /api/v1/takeoff/training/run-drill res_run = self.client.post( "/v1/takeoff/training/run-drill", json={"drill_id": "DRILL-NVH-DAITHANG-2026"}, ) self.assertEqual(res_run.status_code, 200) scorecard_data = res_run.json() self.assertEqual(scorecard_data["drill_id"], "DRILL-NVH-DAITHANG-2026") self.assertGreater(scorecard_data["total_ground_truth_items"], 0) # 3. GET /api/v1/takeoff/training/scorecard res_sc = self.client.get( "/v1/takeoff/training/scorecard?drill_id=DRILL-NVH-DAITHANG-2026" ) self.assertEqual(res_sc.status_code, 200) self.assertEqual(res_sc.json()["drill_id"], "DRILL-NVH-DAITHANG-2026") if __name__ == "__main__": unittest.main()