"""Automated test suite for AI Quỳnh Takeoff Training, Experience Memory Bank, and Full State Price Catalog.""" from __future__ import annotations import io import openpyxl from fastapi.testclient import TestClient from app.core.postgres.erp_client import ErpDatabaseClient from app.main import create_app from app.modules.takeoff.application.drawing_takeoff_service import DrawingTakeoffService from app.modules.inventory.application.state_price_sync_service import StatePriceSyncService client = TestClient(create_app()) AUTH_HEADERS = {"Authorization": "Bearer dev-test-token"} erp_pg_client = ErpDatabaseClient() def test_state_price_full_catalog_and_seed(): """Kiểm thử kho dữ liệu Bảng giá công bố Sở Xây dựng Hải Phòng đầy đủ 100+ danh mục.""" svc = StatePriceSyncService() count = svc.seed_official_state_prices() assert count > 0 # Lấy toàn bộ danh mục kỳ 2026-08 Khu vực II catalog = svc.get_full_state_price_catalog( publish_period="2026-08", region_code="HAI_PHONG_KHU_VUC_2", limit=500 ) assert catalog["total_items"] >= 70 assert len(catalog["items"]) >= 70 assert len(catalog["group_stats"]) >= 10 # Kiểm tra sự hiện diện của các nhóm vật tư chính groups_found = {g["material_group"] for g in catalog["group_stats"]} assert "AGGREGATE" in groups_found assert "STEEL" in groups_found assert "EQUIPMENT_SHIFT" in groups_found assert "LABOR" in groups_found assert "CEMENT" in groups_found def test_export_full_state_prices_to_excel(): """Kiểm thử tính năng xuất Bảng giá Sở Xây dựng Hải Phòng ra file Excel (.xlsx) chuẩn.""" svc = StatePriceSyncService() excel_bytes = svc.export_full_state_prices_to_excel( "2026-08", "HAI_PHONG_KHU_VUC_2" ) assert excel_bytes is not None assert len(excel_bytes) > 2000 # Kiểm tra tính hợp lệ của file Excel wb = openpyxl.load_workbook(io.BytesIO(excel_bytes)) ws = wb.active assert ws.title == "Gia_SXD_2026-08" assert "CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN" in str(ws["A1"].value) assert "BẢNG ĐƠN GIÁ VẬT LIỆU XÂY DỰNG & CA MÁY" in str(ws["A2"].value) assert ws.max_row >= 50 def test_api_material_prices_full_catalog_and_export(): """Kiểm thử các API endpoints của Bảng giá công bố Sở Xây Dựng.""" # Test GET full catalog res = client.get( "/v1/material-prices/full-catalog?period=2026-08®ion=HAI_PHONG_KHU_VUC_2", headers=AUTH_HEADERS, ) assert res.status_code == 200 json_data = res.json() assert json_data["status"] == "success" assert json_data["data"]["total_items"] >= 50 # Test GET export excel res_excel = client.get( "/v1/material-prices/export-full-excel?period=2026-08®ion=HAI_PHONG_KHU_VUC_2", headers=AUTH_HEADERS, ) assert res_excel.status_code == 200 assert ( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" in res_excel.headers["content-type"] ) assert len(res_excel.content) > 2000 # Test POST import appendix file try: files = { "file": ( "phu_luc_558_sxd_test.xlsx", res_excel.content, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) } data = { "period": "2026-08", "region": "HAI_PHONG_KHU_VUC_2", "document_reference": "Thông báo số 558/TB-SXD ngày 07/08/2026 của Sở Xây dựng Hải Phòng", } res_import = client.post( "/v1/material-prices/import-appendix-file", files=files, data=data, headers=AUTH_HEADERS, ) assert res_import.status_code == 200 import_data = res_import.json() assert import_data["status"] == "success" assert import_data["total_imported"] >= 50 finally: # Rule 12 cleanup: purge test items if any dummy were created with erp_pg_client.get_connection() as conn: with conn.cursor() as cur: cur.execute( "DELETE FROM erp_state_published_prices WHERE material_code LIKE 'TBG-IMP-TEST%';" ) conn.commit() # Test POST sync from Price_Ref res_sync = client.post( "/v1/material-prices/sync-from-price-ref", headers=AUTH_HEADERS ) assert res_sync.status_code == 200 sync_data = res_sync.json() assert sync_data["status"] == "success" assert sync_data["total_items"] >= 500 def test_ai_quynh_learned_rules_crud(): """Kiểm thử CRUD cho Kho tri thức quy tắc rút kinh nghiệm (AI QS Memory Bank).""" # 1. Create rule rule_payload = { "rule_title": "Test: Bắt buộc trừ đầu cọc BTCT trong đài móng", "drawing_type": "STRUCTURAL_CONCRETE", "rule_category": "CONCRETE_VOLUME", "trigger_pattern": "Bóc bê tông đài móng chưa trừ thể tích 0.1m cọc cắm vào đài", "correction_directive": "Lấy V_đài - (Số_cọc * Tiết_diện_cọc * 0.1m)", "created_by_role": "Kỹ Sư Trưởng QS (Test)", } saved_rule = erp_pg_client.save_takeoff_learned_rule(rule_payload) assert saved_rule is not None rule_id = str(saved_rule["id"]) # 2. List rules rules = erp_pg_client.list_takeoff_learned_rules(drawing_type="STRUCTURAL_CONCRETE") assert any(str(r["id"]) == rule_id for r in rules) # 3. Test API list & create res_list = client.get("/v1/takeoff/learned-rules/list", headers=AUTH_HEADERS) assert res_list.status_code == 200 assert any(str(r["id"]) == rule_id for r in res_list.json()["data"]) # 4. Delete rule deleted = erp_pg_client.delete_takeoff_learned_rule(rule_id) assert deleted is True # Verify deleted rules_after = erp_pg_client.list_takeoff_learned_rules() assert not any(str(r["id"]) == rule_id for r in rules_after) def test_ai_quynh_chat_train_and_apply_corrections(): """Kiểm thử luồng chat huấn luyện AI Quỳnh, lưu tin nhắn và áp dụng khối lượng hiệu chỉnh.""" # 1. Tạo bản vẽ takeoff giả lập trong DB with erp_pg_client.get_connection() as conn: with conn.cursor() as cur: cur.execute(""" INSERT INTO erp_drawing_takeoffs ( id, drawing_code, drawing_title, drawing_type, file_url, takeoff_status, total_estimated_cost_vnd, total_concrete_volume_m3, total_formwork_area_m2, total_rebar_weight_tons, total_earthwork_volume_m3, confidence_score, created_at, updated_at ) VALUES ( gen_random_uuid(), 'TEST-DWG-001', 'Bản vẽ móng thử nghiệm', 'STRUCTURAL_CONCRETE', '/static/uploads/drawings/test.dwg', 'completed', 100000000.0, 50.0, 120.0, 4.5, 80.0, 0.95, NOW(), NOW() ) RETURNING id; """) takeoff_id = str(cur.fetchone()["id"]) # Thêm 1 item mẫu cur.execute( """ INSERT INTO erp_drawing_takeoff_items ( id, takeoff_id, item_order, wbs_code, norm_code, item_name, category, dimension_formula, unit, quantity, unit_price_vnd, total_amount_vnd, created_at ) VALUES ( gen_random_uuid(), %s, 1, '1.1', 'AF.11110', 'Bê tông đài móng M300', 'concrete', '10*5*1', 'm3', 50.0, 1500000.0, 75000000.0, NOW() ); """, (takeoff_id,), ) conn.commit() try: # 2. Gọi Chat Train API chat_req = { "feedback_message": "Khối lượng bê tông đài móng bị thừa, phải là 45 m3 chứ không phải 50 m3. Ngoài ra bổ sung ván khuôn đài móng 110 m2.", "auto_apply_corrections": True, "sender_role": "Kỹ Sư Trưởng QS", } res_chat = client.post( f"/v1/takeoff/{takeoff_id}/chat-train", json=chat_req, headers=AUTH_HEADERS ) assert res_chat.status_code == 200 chat_data = res_chat.json()["data"] assert "Quỳnh" in chat_data["ai_response"] assert len(chat_data["suggested_boq_items"]) > 0 assert chat_data["applied_to_takeoff"] is True # 3. Kiểm tra lịch sử chat res_hist = client.get( f"/v1/takeoff/{takeoff_id}/chat-history", headers=AUTH_HEADERS ) assert res_hist.status_code == 200 messages = res_hist.json()["data"] assert len(messages) >= 2 # user + ai_quynh # 4. Kiểm tra Takeoff Header KPI đã được tự động cập nhật takeoff_record = erp_pg_client.get_drawing_takeoff(takeoff_id) assert takeoff_record is not None assert float(takeoff_record["total_concrete_volume_m3"]) > 0 assert len(messages) >= 2 # user + ai_quynh # 4. Kiểm tra Takeoff Header KPI đã được tự động cập nhật takeoff_record = erp_pg_client.get_drawing_takeoff(takeoff_id) assert takeoff_record is not None assert float(takeoff_record["total_concrete_volume_m3"]) > 0 finally: # Cleanup (Rule 12 compliant) with erp_pg_client.get_connection() as conn: with conn.cursor() as cur: cur.execute( "DELETE FROM erp_ai_takeoff_learned_rules WHERE takeoff_id = %s OR error_pattern LIKE %s;", (takeoff_id, "%45 m3%"), ) cur.execute( "DELETE FROM erp_ai_takeoff_chat_messages WHERE takeoff_id = %s;", (takeoff_id,), ) 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() def test_ai_quynh_learned_rules_deduplication_and_upsert(): """Kiểm thử cơ chế chống trùng lặp và gộp tri thức của AI Quỳnh QS.""" rule_payload = { "rule_title": "Quy tắc kiểm tra thép đài móng D22", "drawing_type": "STRUCTURAL_CONCRETE", "rule_category": "REBAR", "trigger_pattern": "Bóc thiếu thép đai tăng cường cột đài móng D22 a150", "correction_directive": "Bắt buộc bổ sung thép đai tăng cường D22 a150 cho mọi đài móng", "created_by_role": "Kỹ Sư Trưởng QS (Deduplication Test)", } try: # Lần 1: Lưu quy tắc -> apply_count = 1 rule1 = erp_pg_client.save_takeoff_learned_rule(rule_payload) assert rule1 is not None assert int(rule1.get("apply_count") or 1) >= 1 r1_id = str(rule1["id"]) # Lần 2: Lưu cùng quy tắc -> Không tạo row mới, chỉ tăng apply_count rule2 = erp_pg_client.save_takeoff_learned_rule(rule_payload) assert str(rule2["id"]) == r1_id assert int(rule2.get("apply_count") or 1) >= 2 # Lần 3: Gọi API Deduplicate res_dedup = client.post( "/v1/takeoff/learned-rules/deduplicate", headers=AUTH_HEADERS ) assert res_dedup.status_code == 200 assert res_dedup.json()["status"] == "success" finally: # Dọn dẹp sạch theo Rule 12 with erp_pg_client.get_connection() as conn: with conn.cursor() as cur: cur.execute( "DELETE FROM erp_ai_takeoff_learned_rules WHERE error_pattern LIKE %s;", ("%D22 a150%",), ) conn.commit() def test_ai_quynh_unit_scaling_and_sxd_price_sync(): """Kiểm thử tự động khớp nối đơn giá Sở Xây Dựng 2026 và tỷ lệ đơn vị 100m3 (Triệt tiêu sai số làm tròn 100x).""" svc = DrawingTakeoffService() # Raw items from takeoff (như ảnh người dùng phản ánh: dòng 2.6 và 2.7) raw_items = [ { "item_order": 1, "wbs_code": "2.6", "norm_code": "AB.1111", "item_name": "Đào vận chuyển đất cấp I", "unit": "100m3", "quantity": 12.555, "unit_price_vnd": 0, "total_amount_vnd": 0, }, { "item_order": 2, "wbs_code": "2.7", "norm_code": "AB.2222", "item_name": "Vận chuyển bùn, đất cấp I", "unit": "100m3", "quantity": 12.555, "unit_price_vnd": 0, "total_amount_vnd": 0, }, { "item_order": 3, "wbs_code": "3.1", "norm_code": "AF.1111", "item_name": "Bê tông lót móng M100", "unit": "m3", "quantity": 15.0, "unit_price_vnd": 0, "total_amount_vnd": 0, }, ] # Step 1: match and apply state prices processed = svc.apply_learned_rules_and_state_prices( raw_items, drawing_type="hydraulic_culvert" ) assert len(processed) >= 3 # Check item 1 (Đào vận chuyển đất 100m3) item_dao = processed[0] assert item_dao["unit"] == "100m3" assert ( item_dao["unit_price_vnd"] >= 4800000.0 ) # Phải là đơn giá 100m3, không thể là 48.000 hoặc 75.000 assert item_dao["total_amount_vnd"] >= 60000000.0 # 12.555 * 4.8tr >= 60tr # Check item 2 (Vận chuyển bùn đất 100m3) item_vc = processed[1] assert item_vc["unit"] == "100m3" assert item_vc["unit_price_vnd"] == 7500000.0 # 7.500.000 VNĐ / 100m3 assert item_vc["total_amount_vnd"] == round(12.555 * 7500000.0) # 94.162.500 VNĐ # Check item 3 (Bê tông lót móng M100) item_bt = processed[2] assert item_bt["unit_price_vnd"] == 1050000.0 assert item_bt["total_amount_vnd"] == 15750000.0 # Step 2: Test 1-click API endpoint reapply-rules-and-prices with erp_pg_client.get_connection() as conn: with conn.cursor() as cur: cur.execute(""" INSERT INTO erp_drawing_takeoffs ( id, drawing_code, drawing_title, drawing_type, file_url, takeoff_status, total_estimated_cost_vnd, total_concrete_volume_m3, total_formwork_area_m2, total_rebar_weight_tons, total_earthwork_volume_m3, confidence_score, created_at, updated_at ) VALUES ( gen_random_uuid(), 'TEST-DWG-SCALE', 'Bản vẽ kiểm thử scale 100m3', 'hydraulic_culvert', '/static/uploads/drawings/test_scale.pdf', 'completed', 0.0, 0.0, 0.0, 0.0, 0.0, 0.95, NOW(), NOW() ) RETURNING id; """) tk_id = str(cur.fetchone()["id"]) cur.execute( """ INSERT INTO erp_drawing_takeoff_items ( id, takeoff_id, item_order, wbs_code, norm_code, item_name, category, dimension_formula, unit, quantity, unit_price_vnd, total_amount_vnd, created_at ) VALUES ( gen_random_uuid(), %s, 1, '2.6', 'AB.1111', 'Vận chuyển bùn, đất cấp I', 'earthwork', '12.555 100m3', '100m3', 12.555, 75000.0, 941625.0, NOW() ); """, (tk_id,), ) conn.commit() try: res_reapply = client.post( f"/v1/takeoff/{tk_id}/reapply-rules-and-prices", headers=AUTH_HEADERS ) assert res_reapply.status_code == 200 res_data = res_reapply.json() assert res_data["status"] == "success" assert res_data["total_estimated_cost_vnd"] == 94162500.0 # Verify items in database updated_db_items = erp_pg_client.list_drawing_takeoff_items(tk_id) assert len(updated_db_items) == 1 assert float(updated_db_items[0]["unit_price_vnd"]) == 7500000.0 assert float(updated_db_items[0]["total_amount_vnd"]) == 94162500.0 finally: # Cleanup (Rule 12 compliant) with erp_pg_client.get_connection() as conn: with conn.cursor() as cur: cur.execute( "DELETE FROM erp_drawing_takeoff_items WHERE takeoff_id = %s;", (tk_id,), ) cur.execute("DELETE FROM erp_drawing_takeoffs WHERE id = %s;", (tk_id,)) conn.commit()