151 lines
4.8 KiB
Python
151 lines
4.8 KiB
Python
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from flask import Flask
|
|
|
|
from exam_pkg.api import exam_bp
|
|
from exam_pkg.manager import check_file_indexed
|
|
|
|
|
|
class ExamFileStatusApiTest(unittest.TestCase):
|
|
def setUp(self):
|
|
app = Flask(__name__)
|
|
app.config["TESTING"] = True
|
|
app.register_blueprint(exam_bp, url_prefix="/exam")
|
|
self.client = app.test_client()
|
|
|
|
@patch("exam_pkg.api.generate_questions_from_file")
|
|
def test_generate_returns_409_when_file_is_not_indexed(self, generate):
|
|
generate.return_value = {
|
|
"success": False,
|
|
"request_id": "req-1",
|
|
"error_code": "FILE_NOT_INDEXED",
|
|
"message": "文件未向量化",
|
|
"file_status": "not_found",
|
|
"chunk_count": 0,
|
|
"questions": [],
|
|
"total": 0,
|
|
"source_chunks_used": 0,
|
|
}
|
|
|
|
response = self.client.post(
|
|
"/exam/generate",
|
|
json={
|
|
"file_path": "missing.pdf",
|
|
"collection": "public_kb",
|
|
"question_types": {"single_choice": 1},
|
|
"request_id": "req-1",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 409)
|
|
body = response.get_json()
|
|
self.assertFalse(body["success"])
|
|
self.assertEqual(body["error_code"], "FILE_NOT_INDEXED")
|
|
self.assertEqual(body["status_code"], 4016)
|
|
self.assertEqual(body["data"]["request_id"], "req-1")
|
|
|
|
@patch("exam_pkg.manager.analyze_file_for_exam")
|
|
@patch("exam_pkg.api.check_file_indexed")
|
|
def test_generate_smart_stops_before_ai_analysis(self, check_status, analyze):
|
|
check_status.return_value = {
|
|
"indexed": False,
|
|
"chunk_count": 0,
|
|
"status": "not_found",
|
|
"message": "文件未找到",
|
|
}
|
|
|
|
response = self.client.post(
|
|
"/exam/generate-smart",
|
|
json={
|
|
"file_path": "missing.pdf",
|
|
"collection": "public_kb",
|
|
"request_id": "req-smart",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 409)
|
|
self.assertEqual(response.get_json()["status_code"], 4016)
|
|
analyze.assert_not_called()
|
|
|
|
@patch("exam_pkg.api.generate_questions_from_file")
|
|
def test_generate_success_response_is_unchanged(self, generate):
|
|
generate.return_value = {
|
|
"success": True,
|
|
"request_id": "req-ok",
|
|
"questions": [{"question_type": "single_choice"}],
|
|
"total": 1,
|
|
"source_chunks_used": 1,
|
|
}
|
|
|
|
response = self.client.post(
|
|
"/exam/generate",
|
|
json={
|
|
"file_path": "ready.pdf",
|
|
"collection": "public_kb",
|
|
"question_types": {"single_choice": 1},
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.get_json()
|
|
self.assertTrue(body["success"])
|
|
self.assertEqual(body["status_code"], 2020)
|
|
self.assertTrue(body["data"]["success"])
|
|
|
|
@patch("exam_pkg.api.generate_questions_from_file")
|
|
def test_status_check_error_uses_existing_exam_error(self, generate):
|
|
generate.return_value = {
|
|
"success": False,
|
|
"request_id": "req-error",
|
|
"error_code": "STATUS_CHECK_ERROR",
|
|
"message": "检查状态异常",
|
|
"file_status": "error",
|
|
"chunk_count": 0,
|
|
}
|
|
|
|
response = self.client.post(
|
|
"/exam/generate",
|
|
json={
|
|
"file_path": "document.pdf",
|
|
"collection": "public_kb",
|
|
"question_types": {"single_choice": 1},
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 500)
|
|
body = response.get_json()
|
|
self.assertEqual(body["error_code"], "EXAM_ERROR")
|
|
self.assertEqual(body["status_code"], 5020)
|
|
|
|
|
|
class FilePathNormalizationTest(unittest.TestCase):
|
|
@patch("core.engine.get_engine")
|
|
def test_windows_path_is_matched_by_filename(self, get_engine):
|
|
collection = MagicMock()
|
|
collection.get.return_value = {"ids": ["chunk-1"]}
|
|
engine = MagicMock()
|
|
engine.search_knowledge.return_value = {
|
|
"documents": [["content"]],
|
|
}
|
|
engine.kb_manager.get_collection.return_value = collection
|
|
get_engine.return_value = engine
|
|
|
|
result = check_file_indexed(
|
|
r"public_kb\\folder\\ready.pdf",
|
|
"public_kb",
|
|
)
|
|
|
|
self.assertTrue(result["indexed"])
|
|
engine.search_knowledge.assert_called_once_with(
|
|
query="document",
|
|
collections=["public_kb"],
|
|
source_filter="ready.pdf",
|
|
top_k=1,
|
|
)
|
|
collection.get.assert_called_once_with(where={"source": "ready.pdf"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|