"""Automated test suite for Enterprise Project CRUD, WBS Hierarchy, and AI WBS Generator.""" import unittest from fastapi.testclient import TestClient from app.modules.auth.presentation.auth import get_current_user from app.main import create_app from app.modules.core.application.ai_wbs_generator_service import AIWbsGeneratorService class TestProjectCrudAndWbsService(unittest.TestCase): def setUp(self): self.app = create_app() self.client = TestClient(self.app) self.ai_service = AIWbsGeneratorService() def tearDown(self): self.app.dependency_overrides = {} def test_unauthorized_access(self): """Khi chưa đăng nhập, các endpoint trả về 401.""" res = self.client.get("/v1/erp/projects") self.assertEqual(res.status_code, 401) def test_ai_wbs_generator_service(self): """Kiểm thử nhân viên AI phân tích biểu giá hợp đồng và lập timeline CPM 4 giai đoạn.""" result = self.ai_service.generate_timeline_and_wbs( project_name="Cụm Công Trình Cấp Nước Xã Kiến Minh", project_type="Hạ tầng cấp nước", contract_value=3500000000.0, start_date="2026-09-01", duration_days=180, contract_schedule_text="1. Đào đất đặt ống HDPE D110: 4500 md\n2. Xây bể chứa nước sạch 500m3: 1 cụm\n3. Trạm bơm cấp nước: 2 máy", ) self.assertIn("wbs_items", result) self.assertIn("milestones", result) self.assertEqual(result["total_duration_days"], 180) self.assertEqual(result["generated_by_agent"], "phuc") wbs = result["wbs_items"] phases = [item for item in wbs if item["task_type"] == "phase"] self.assertEqual(len(phases), 4) # Kiểm tra mã WBS wbs_codes = [item["wbs_code"] for item in wbs] self.assertIn("1.0", wbs_codes) self.assertIn("2.0", wbs_codes) self.assertIn("3.0", wbs_codes) self.assertIn("4.0", wbs_codes) self.assertIn("2.1", wbs_codes) # Kiểm tra milestones self.assertGreaterEqual(len(result["milestones"]), 4) def test_full_project_crud_lifecycle(self): """Kiểm thử vòng đời CRUD dự án và quản lý WBS tasks.""" self.app.dependency_overrides[get_current_user] = lambda: { "id": "admin-id", "email": "admin@dscons.vn", "role": "superadmin", } # 1. List projects res = self.client.get("/v1/erp/projects") self.assertEqual(res.status_code, 200) initial_projects = res.json() self.assertIsInstance(initial_projects, list) # 2. AI Generate WBS preview ai_res = self.client.post( "/v1/erp/projects/ai-generate-wbs", json={ "project_name": "Nhà điều hành KCN Sông Lô", "project_type": "Dân dụng & Công nghiệp", "contract_value": 4500000000.0, "start_date": "2026-10-01", "duration_days": 120, "contract_schedule_text": "Hạng mục kết cấu móng và hoàn thiện nhà 3 tầng", }, ) self.assertEqual(ai_res.status_code, 200) ai_data = ai_res.json() generated_wbs = ai_data["wbs_items"] # 3. Create new project with AI generated WBS create_res = self.client.post( "/v1/erp/projects", json={ "project_code": "DA-SONGLO-2026", "project_name": "Nhà điều hành KCN Sông Lô", "client_name": "Ban QLDA KCN Sông Lô", "location": "Vĩnh Phúc", "project_type": "Dân dụng & Công nghiệp", "budget_amount": 4200000000.0, "contract_value": 4500000000.0, "start_date": "2026-10-01", "contract_number": "09.1/2026/HĐ-SL", "contract_duration_days": 120, "status": "planning", "wbs_items": generated_wbs, }, ) self.assertEqual(create_res.status_code, 201) created = create_res.json(); import json; open("dump.json", "w", encoding="utf-8").write(json.dumps(created, ensure_ascii=False)) project_id = created["id"] self.assertEqual(created["project_code"], "DA-SONGLO-2026") self.assertGreaterEqual(len(created["wbs_items"]), 10) # 4. Get detail detail_res = self.client.get(f"/v1/erp/projects/{project_id}") self.assertEqual(detail_res.status_code, 200) detail = detail_res.json() self.assertEqual(detail["project_name"], "Nhà điều hành KCN Sông Lô") self.assertIn("wbs_tree", detail) self.assertIn("milestones", detail) # 5. Update Project update_res = self.client.put( f"/v1/erp/projects/{project_id}", json={ "status": "in_progress", "notes": "Đã khởi công sau khi phê duyệt KHLCNT", }, ) self.assertEqual(update_res.status_code, 200) updated = update_res.json() self.assertEqual(updated["status"], "in_progress") # 6. Add a custom WBS subtask wbs_add_res = self.client.post( f"/v1/erp/projects/{project_id}/wbs", json={ "wbs_code": "2.5", "task_name": "Thí nghiệm kiểm tra độ nén tĩnh cọc khoan nhồi", "task_type": "task", "unit": "Điểm", "quantity": 3, "unit_price_vnd": 15000000.0, "total_amount_vnd": 45000000.0, "duration_days": 4, "status": "in_progress", "progress_percent": 50, "assigned_role": "Đơn vị thí nghiệm độc lập", }, ) self.assertEqual(wbs_add_res.status_code, 201) new_wbs = wbs_add_res.json() wbs_id = new_wbs["id"] # 7. Update WBS subtask progress to completed wbs_up_res = self.client.put( f"/v1/erp/projects/wbs/{wbs_id}", json={ "status": "completed", "progress_percent": 100, "notes": "Đạt cường độ thiết kế R28", }, ) self.assertEqual(wbs_up_res.status_code, 200) self.assertEqual(wbs_up_res.json()["progress_percent"], 100) # 8. Delete WBS item wbs_del_res = self.client.delete(f"/v1/erp/projects/wbs/{wbs_id}") self.assertEqual(wbs_del_res.status_code, 200) # 9. Clean up / Delete Project del_res = self.client.delete(f"/v1/erp/projects/{project_id}") self.assertEqual(del_res.status_code, 200) if __name__ == "__main__": unittest.main()