3927441103
- Add Users table for display name resolution from AD authentication - Integrate DIFY AI service for report content generation - Create docx assembly service with image embedding from MinIO - Add REST API endpoints for report generation and download - Add WebSocket notifications for generation progress - Add frontend UI with progress modal and download functionality - Add integration tests for report generation flow Report sections (Traditional Chinese): - 事件摘要 (Summary) - 時間軸 (Timeline) - 參與人員 (Participants) - 處理過程 (Resolution Process) - 目前狀態 (Current Status) - 最終處置結果 (Final Resolution) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
345 lines
12 KiB
Python
345 lines
12 KiB
Python
"""Integration tests for AI report generation
|
||
|
||
Tests the report generation flow:
|
||
1. Data collection from room
|
||
2. DIFY prompt construction
|
||
3. Document assembly
|
||
"""
|
||
import pytest
|
||
from datetime import datetime
|
||
from unittest.mock import Mock, patch, AsyncMock
|
||
import json
|
||
|
||
# Test data fixtures
|
||
@pytest.fixture
|
||
def sample_room_data():
|
||
"""Sample room metadata for testing"""
|
||
return {
|
||
"room_id": "test-room-123",
|
||
"title": "設備故障測試",
|
||
"incident_type": "equipment_failure",
|
||
"severity": "high",
|
||
"status": "active",
|
||
"location": "A棟生產線",
|
||
"description": "測試用事件描述",
|
||
"resolution_notes": None,
|
||
"created_at": datetime(2024, 1, 15, 9, 0),
|
||
"resolved_at": None,
|
||
"created_by": "test@example.com",
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_messages():
|
||
"""Sample messages for testing"""
|
||
return [
|
||
{
|
||
"message_id": "msg-1",
|
||
"sender_id": "user1@example.com",
|
||
"sender_name": "張三",
|
||
"content": "發現設備異常",
|
||
"message_type": "text",
|
||
"created_at": datetime(2024, 1, 15, 9, 5),
|
||
"file_name": None,
|
||
},
|
||
{
|
||
"message_id": "msg-2",
|
||
"sender_id": "user2@example.com",
|
||
"sender_name": "李四",
|
||
"content": "收到，立即前往查看",
|
||
"message_type": "text",
|
||
"created_at": datetime(2024, 1, 15, 9, 10),
|
||
"file_name": None,
|
||
},
|
||
]
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_members():
|
||
"""Sample members for testing"""
|
||
return [
|
||
{
|
||
"user_id": "user1@example.com",
|
||
"display_name": "張三",
|
||
"role": "owner",
|
||
},
|
||
{
|
||
"user_id": "user2@example.com",
|
||
"display_name": "李四",
|
||
"role": "editor",
|
||
},
|
||
]
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_files():
|
||
"""Sample files for testing"""
|
||
return [
|
||
{
|
||
"file_id": "file-1",
|
||
"filename": "故障照片.jpg",
|
||
"file_type": "image",
|
||
"mime_type": "image/jpeg",
|
||
"uploaded_at": datetime(2024, 1, 15, 9, 15),
|
||
"uploader_id": "user1@example.com",
|
||
"uploader_name": "張三",
|
||
"minio_object_path": "rooms/test-room-123/file-1.jpg",
|
||
},
|
||
]
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_ai_response():
|
||
"""Sample AI response JSON for testing"""
|
||
return {
|
||
"summary": {
|
||
"content": "A棟生產線設備於2024年1月15日發生異常，維修人員已前往處理。"
|
||
},
|
||
"timeline": {
|
||
"events": [
|
||
{"time": "09:05", "description": "張三發現設備異常並通報"},
|
||
{"time": "09:10", "description": "李四收到通報前往查看"},
|
||
]
|
||
},
|
||
"participants": {
|
||
"members": [
|
||
{"name": "張三", "role": "事件發起人"},
|
||
{"name": "李四", "role": "維修負責人"},
|
||
]
|
||
},
|
||
"resolution_process": {
|
||
"content": "維修人員接獲通報後立即前往現場查看設備狀況。"
|
||
},
|
||
"current_status": {
|
||
"status": "active",
|
||
"description": "維修人員正在現場處理中"
|
||
},
|
||
"final_resolution": {
|
||
"has_resolution": False,
|
||
"content": ""
|
||
}
|
||
}
|
||
|
||
|
||
class TestPromptConstruction:
|
||
"""Tests for prompt construction"""
|
||
|
||
def test_build_report_prompt_includes_room_info(self, sample_room_data, sample_messages, sample_members, sample_files):
|
||
"""Test that prompt includes room information"""
|
||
from app.modules.report_generation.prompts import build_report_prompt
|
||
|
||
prompt = build_report_prompt(
|
||
room_data=sample_room_data,
|
||
messages=sample_messages,
|
||
members=sample_members,
|
||
files=sample_files,
|
||
)
|
||
|
||
assert "設備故障測試" in prompt
|
||
assert "設備故障" in prompt # Translated incident type
|
||
assert "高" in prompt # Translated severity
|
||
assert "A棟生產線" in prompt
|
||
|
||
def test_build_report_prompt_includes_messages(self, sample_room_data, sample_messages, sample_members, sample_files):
|
||
"""Test that prompt includes message content"""
|
||
from app.modules.report_generation.prompts import build_report_prompt
|
||
|
||
prompt = build_report_prompt(
|
||
room_data=sample_room_data,
|
||
messages=sample_messages,
|
||
members=sample_members,
|
||
files=sample_files,
|
||
)
|
||
|
||
assert "發現設備異常" in prompt
|
||
assert "張三" in prompt
|
||
assert "李四" in prompt
|
||
|
||
def test_build_report_prompt_includes_files(self, sample_room_data, sample_messages, sample_members, sample_files):
|
||
"""Test that prompt includes file information"""
|
||
from app.modules.report_generation.prompts import build_report_prompt
|
||
|
||
prompt = build_report_prompt(
|
||
room_data=sample_room_data,
|
||
messages=sample_messages,
|
||
members=sample_members,
|
||
files=sample_files,
|
||
)
|
||
|
||
assert "故障照片.jpg" in prompt
|
||
assert "圖片" in prompt # File type label
|
||
|
||
|
||
class TestDifyJsonParsing:
|
||
"""Tests for DIFY response JSON parsing"""
|
||
|
||
def test_extract_json_from_pure_json(self, sample_ai_response):
|
||
"""Test parsing pure JSON response"""
|
||
from app.modules.report_generation.services.dify_client import DifyService
|
||
|
||
service = DifyService()
|
||
text = json.dumps(sample_ai_response)
|
||
result = service._extract_json(text)
|
||
|
||
assert result["summary"]["content"] == sample_ai_response["summary"]["content"]
|
||
|
||
def test_extract_json_from_markdown_code_block(self, sample_ai_response):
|
||
"""Test parsing JSON from markdown code block"""
|
||
from app.modules.report_generation.services.dify_client import DifyService
|
||
|
||
service = DifyService()
|
||
text = f"```json\n{json.dumps(sample_ai_response)}\n```"
|
||
result = service._extract_json(text)
|
||
|
||
assert result["summary"]["content"] == sample_ai_response["summary"]["content"]
|
||
|
||
def test_extract_json_from_mixed_text(self, sample_ai_response):
|
||
"""Test parsing JSON embedded in other text"""
|
||
from app.modules.report_generation.services.dify_client import DifyService
|
||
|
||
service = DifyService()
|
||
text = f"Here is the report:\n{json.dumps(sample_ai_response)}\nEnd of report."
|
||
result = service._extract_json(text)
|
||
|
||
assert result["summary"]["content"] == sample_ai_response["summary"]["content"]
|
||
|
||
def test_validate_schema_with_valid_data(self, sample_ai_response):
|
||
"""Test schema validation with valid data"""
|
||
from app.modules.report_generation.services.dify_client import DifyService
|
||
|
||
service = DifyService()
|
||
# Should not raise any exception
|
||
service._validate_schema(sample_ai_response)
|
||
|
||
def test_validate_schema_with_missing_section(self, sample_ai_response):
|
||
"""Test schema validation with missing section"""
|
||
from app.modules.report_generation.services.dify_client import DifyService, DifyValidationError
|
||
|
||
service = DifyService()
|
||
del sample_ai_response["summary"]
|
||
|
||
with pytest.raises(DifyValidationError) as exc_info:
|
||
service._validate_schema(sample_ai_response)
|
||
|
||
assert "summary" in str(exc_info.value)
|
||
|
||
|
||
class TestDocxGeneration:
|
||
"""Tests for docx document generation"""
|
||
|
||
def test_create_report_returns_bytesio(self, sample_room_data, sample_ai_response, sample_files):
|
||
"""Test that document assembly returns BytesIO"""
|
||
from app.modules.report_generation.services.docx_service import DocxAssemblyService
|
||
import io
|
||
|
||
service = DocxAssemblyService()
|
||
result = service.create_report(
|
||
room_data=sample_room_data,
|
||
ai_content=sample_ai_response,
|
||
files=sample_files,
|
||
include_images=False, # Skip image download for test
|
||
include_file_list=True,
|
||
)
|
||
|
||
assert isinstance(result, io.BytesIO)
|
||
assert result.getvalue() # Should have content
|
||
|
||
def test_create_report_is_valid_docx(self, sample_room_data, sample_ai_response, sample_files):
|
||
"""Test that generated document is valid DOCX"""
|
||
from app.modules.report_generation.services.docx_service import DocxAssemblyService
|
||
from docx import Document
|
||
|
||
service = DocxAssemblyService()
|
||
result = service.create_report(
|
||
room_data=sample_room_data,
|
||
ai_content=sample_ai_response,
|
||
files=sample_files,
|
||
include_images=False,
|
||
include_file_list=True,
|
||
)
|
||
|
||
# Should be able to parse as DOCX
|
||
doc = Document(result)
|
||
assert len(doc.paragraphs) > 0
|
||
|
||
|
||
class TestReportDataService:
|
||
"""Tests for report data collection service"""
|
||
|
||
def test_to_prompt_dict_format(self, sample_room_data, sample_messages, sample_members, sample_files):
|
||
"""Test data conversion to prompt dictionary format"""
|
||
from app.modules.report_generation.services.report_data_service import (
|
||
ReportDataService,
|
||
RoomReportData,
|
||
MessageData,
|
||
MemberData,
|
||
FileData,
|
||
)
|
||
|
||
# Create RoomReportData manually for testing
|
||
room_data = RoomReportData(
|
||
room_id=sample_room_data["room_id"],
|
||
title=sample_room_data["title"],
|
||
incident_type=sample_room_data["incident_type"],
|
||
severity=sample_room_data["severity"],
|
||
status=sample_room_data["status"],
|
||
location=sample_room_data["location"],
|
||
description=sample_room_data["description"],
|
||
resolution_notes=sample_room_data["resolution_notes"],
|
||
created_at=sample_room_data["created_at"],
|
||
resolved_at=sample_room_data["resolved_at"],
|
||
created_by=sample_room_data["created_by"],
|
||
messages=[
|
||
MessageData(
|
||
message_id=m["message_id"],
|
||
sender_id=m["sender_id"],
|
||
sender_name=m["sender_name"],
|
||
content=m["content"],
|
||
message_type=m["message_type"],
|
||
created_at=m["created_at"],
|
||
file_name=m.get("file_name"),
|
||
)
|
||
for m in sample_messages
|
||
],
|
||
members=[
|
||
MemberData(
|
||
user_id=m["user_id"],
|
||
display_name=m["display_name"],
|
||
role=m["role"],
|
||
)
|
||
for m in sample_members
|
||
],
|
||
files=[
|
||
FileData(
|
||
file_id=f["file_id"],
|
||
filename=f["filename"],
|
||
file_type=f["file_type"],
|
||
mime_type=f["mime_type"],
|
||
uploaded_at=f["uploaded_at"],
|
||
uploader_id=f["uploader_id"],
|
||
uploader_name=f["uploader_name"],
|
||
minio_object_path=f["minio_object_path"],
|
||
)
|
||
for f in sample_files
|
||
],
|
||
)
|
||
|
||
# Test conversion using class method directly (not from db session)
|
||
# Create mock db session
|
||
mock_db = Mock()
|
||
service = ReportDataService(mock_db)
|
||
result = service.to_prompt_dict(room_data)
|
||
|
||
assert "room_data" in result
|
||
assert "messages" in result
|
||
assert "members" in result
|
||
assert "files" in result
|
||
assert result["room_data"]["title"] == sample_room_data["title"]
|
||
assert len(result["messages"]) == 2
|
||
assert len(result["members"]) == 2
|
||
assert len(result["files"]) == 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v"])
|