xiner/Edu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ai): gRPC clients 扩展 + server interceptors + proto_gen + 测试 + nextstep 文档

Browse Source
This commit is contained in:
SpecialX
2026-07-14 15:59:41 +08:00
parent fb23c5234e
commit 7b790f1276
21 changed files with 3540 additions and 345 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
services/ai/tests/test_clients.py
View File

@@ -19,7 +19,7 @@ class TestContentClientMock:
"""查询学习路径."""
client = ContentClientMock()
result = await client.get_learning_path("s-1", "math")
assert len(result) == 3
assert len(result) == 2
async def test_create_questions(self) -> None:
"""批量创建题目."""

173
services/ai/tests/test_coverage_gaps.py
View File

@@ -419,34 +419,55 @@ async def test_content_grpc_connect_close() -> None:
async def test_content_grpc_methods_not_connected() -> None:
"""未连接时调用方法应抛 AIError（不降级到 mock）."""
client = ContentClientGrpc()
assert client.is_available() is False
pre = await client.get_prerequisites("kp-1")
assert len(pre) == 1
path = await client.get_learning_path("s-1", "math")
assert len(path) == 3
questions = [
QuestionInput(
question="q1",
answer="a1",
explanation="e1",
question_type="short_answer",
difficulty="easy",
knowledge_point_ids=["kp-1"],
),
]
result = await client.create_questions(questions, user_id="u-1")
assert len(result) == 1
with pytest.raises(AIError, match="not connected"):
await client.get_prerequisites("kp-1")
with pytest.raises(AIError, match="not connected"):
await client.get_learning_path("s-1", "math")
with pytest.raises(AIError, match="not connected"):
await client.create_questions([], user_id="u-1")
async def test_content_grpc_methods_connected() -> None:
"""连接后通过 gRPC stub 发起真实调用（使用 mock stub 验证）."""
from src.ai.proto_gen import content_pb2
client = ContentClientGrpc()
client._channel = MagicMock()
# 构造 mock stub 返回 proto 响应
mock_kg_stub = MagicMock()
mock_kg_stub.GetPrerequisites = AsyncMock(
return_value=content_pb2.KnowledgePointsResponse(
points=[
content_pb2.KnowledgePoint(id="kp_pre", title="前置知识点"),
],
),
)
mock_kg_stub.GetLearningPath = AsyncMock(
return_value=content_pb2.LearningPath(
points=[
content_pb2.KnowledgePoint(id="kp_1", title="知识点1"),
content_pb2.KnowledgePoint(id="kp_2", title="知识点2"),
],
),
)
mock_q_stub = MagicMock()
mock_q_stub.BatchCreateQuestions = AsyncMock(
return_value=content_pb2.BatchCreateQuestionsResponse(ids=["q_new_1"]),
)
client._kg_stub = mock_kg_stub
client._q_stub = mock_q_stub
assert client.is_available() is True
pre = await client.get_prerequisites("kp-1")
assert len(pre) == 1
assert pre[0].id == "kp_pre"
path = await client.get_learning_path("s-1", "math")
assert len(path) == 3
assert len(path) == 2
questions = [
QuestionInput(
question="q1",
@@ -459,13 +480,25 @@ async def test_content_grpc_methods_connected() -> None:
]
result = await client.create_questions(questions, user_id="u-1")
assert len(result) == 1
assert result[0].id == "q_new_1"
async def test_content_grpc_create_questions_exception_raises() -> None:
"""gRPC 调用失败时包装为 AIError."""
import grpc
client = ContentClientGrpc()
client._channel = MagicMock()
client._mock = MagicMock()
client._mock.create_questions = AsyncMock(side_effect=RuntimeError("boom"))
mock_q_stub = MagicMock()
mock_q_stub.BatchCreateQuestions = AsyncMock(
side_effect=grpc.aio.AioRpcError(
code=grpc.StatusCode.UNAVAILABLE,
initial_metadata=[],
trailing_metadata=[],
details="service unavailable",
),
)
client._q_stub = mock_q_stub
with pytest.raises(AIError) as exc_info:
await client.create_questions([], user_id="u-1")
assert exc_info.value.code == ErrorCode.AI_DOWNSTREAM_UNAVAILABLE
@@ -490,29 +523,71 @@ async def test_data_ana_grpc_connect_close() -> None:
async def test_data_ana_grpc_methods_not_connected() -> None:
"""未连接时调用方法应抛 AIError."""
client = DataAnaClientGrpc()
assert client.is_available() is False
perf = await client.get_class_performance("c-1", "math")
assert perf.class_id == "c-1"
assert perf.average_score > 0
weak = await client.get_student_weakness("s-1", "math")
assert weak.student_id == "s-1"
assert len(weak.weak_points) > 0
trend = await client.get_learning_trend("s-1")
assert trend.student_id == "s-1"
assert len(trend.points) > 0
with pytest.raises(AIError, match="not connected"):
await client.get_class_performance("c-1", "math")
with pytest.raises(AIError, match="not connected"):
await client.get_student_weakness("s-1", "math")
with pytest.raises(AIError, match="not connected"):
await client.get_learning_trend("s-1")
async def test_data_ana_grpc_methods_connected() -> None:
"""连接后通过 gRPC stub 真实调用."""
from src.ai.proto_gen import analytics_pb2
client = DataAnaClientGrpc()
client._channel = MagicMock()
mock_stub = MagicMock()
mock_stub.GetClassPerformance = AsyncMock(
return_value=analytics_pb2.ClassPerformance(
class_id="c-1",
average_score=82.5,
pass_rate=0.9,
scores=[
analytics_pb2.StudentScore(student_id="s-1", score=85.0, grade="A"),
],
),
)
mock_stub.GetStudentWeakness = AsyncMock(
return_value=analytics_pb2.StudentWeakness(
student_id="s-1",
weak_points=[
analytics_pb2.WeakPoint(
knowledge_point_id="kp-1",
title="函数",
mastery=0.4,
),
],
),
)
mock_stub.GetLearningTrend = AsyncMock(
return_value=analytics_pb2.LearningTrend(
student_id="s-1",
points=[
analytics_pb2.TrendPoint(date=20260101, score=70.0),
analytics_pb2.TrendPoint(date=20260201, score=75.0),
],
),
)
client._stub = mock_stub
assert client.is_available() is True
perf = await client.get_class_performance("c-1", "math")
assert perf.class_id == "c-1"
assert perf.average_score == 82.5
assert len(perf.scores) == 1
weak = await client.get_student_weakness("s-1", "math")
assert weak.student_id == "s-1"
assert len(weak.weak_points) == 1
assert weak.weak_points[0].knowledge_point_id == "kp-1"
trend = await client.get_learning_trend("s-1")
assert trend.student_id == "s-1"
assert len(trend.points) == 2
# ---------------------------------------------------------------------------
@@ -534,16 +609,46 @@ async def test_iam_grpc_connect_close() -> None:
async def test_iam_grpc_get_effective_data_scope() -> None:
"""未连接时抛 AIError；连接后通过 gRPC stub 真实调用."""
from src.ai.proto_gen import iam_pb2
client = IamClientGrpc()
assert client.is_available() is False
with pytest.raises(AIError, match="not connected"):
await client.get_effective_data_scope("u-1")
# 连接后通过 mock stub 调用
client._channel = MagicMock()
mock_stub = MagicMock()
mock_stub.GetEffectiveDataScope = AsyncMock(
return_value=iam_pb2.EffectiveDataScope(
user_id="u-1",
level="CLASS",
scope_ids=["class_001", "class_002"],
school_id="school_001",
),
)
client._stub = mock_stub
assert client.is_available() is True
scope = await client.get_effective_data_scope("u-1")
assert scope.user_id == "u-1"
assert scope.school_id == "school_mock_001"
assert len(scope.class_ids) > 0
client._channel = MagicMock()
assert client.is_available() is True
scope2 = await client.get_effective_data_scope("u-2")
assert scope2.user_id == "u-2"
assert scope.school_id == "school_001"
assert scope.class_ids == ["class_001", "class_002"]
assert scope.is_admin is False
# ALL level 测试
mock_stub.GetEffectiveDataScope = AsyncMock(
return_value=iam_pb2.EffectiveDataScope(
user_id="u-admin",
level="ALL",
scope_ids=[],
school_id="",
),
)
scope2 = await client.get_effective_data_scope("u-admin")
assert scope2.is_admin is True
assert scope2.role == "admin"
# ---------------------------------------------------------------------------

57
services/ai/tests/test_lesson_workflow.py
View File

@@ -23,13 +23,15 @@ from src.ai.workflow.state_store import WorkflowState, WorkflowStateStore
from .conftest import MockProvider
# 有效的 LLM JSON 输出（通过三道防线评估）
VALID_QUESTION_JSON = json.dumps({
"question": "什么是函数？",
"answer": "函数是一种对应关系",
"explanation": "函数定义",
"difficulty": "medium",
"question_type": "short_answer",
})
VALID_QUESTION_JSON = json.dumps(
{
"question": "什么是函数？",
"answer": "函数是一种对应关系",
"explanation": "函数定义",
"difficulty": "medium",
"question_type": "short_answer",
}
)
def _make_chain(provider: MockProvider | None = None) -> ProviderFailoverChain:
@@ -135,11 +137,16 @@ class TestLessonPlanWorkflowConfirm:
store = WorkflowStateStore(redis=None)
state = _make_state(
status="pending_review",
questions=[GeneratedQuestionData(
question="q1", answer="a1", explanation="e1",
question_type="short_answer", difficulty="easy",
knowledge_point_ids=["kp_1"],
)],
questions=[
GeneratedQuestionData(
question="q1",
answer="a1",
explanation="e1",
question_type="short_answer",
difficulty="easy",
knowledge_point_ids=["kp_1"],
)
],
)
await store.create(state)
svc = _make_service(store=store, content_client=ContentClientMock())
@@ -166,11 +173,16 @@ class TestLessonPlanWorkflowConfirm:
store = WorkflowStateStore(redis=None)
state = _make_state(
status="pending_review",
questions=[GeneratedQuestionData(
question="original", answer="a1", explanation="e1",
question_type="short_answer", difficulty="easy",
knowledge_point_ids=["kp_1"],
)],
questions=[
GeneratedQuestionData(
question="original",
answer="a1",
explanation="e1",
question_type="short_answer",
difficulty="easy",
knowledge_point_ids=["kp_1"],
)
],
)
await store.create(state)
@@ -200,7 +212,7 @@ class TestLessonPlanWorkflowSteps:
analysis = await svc._step1_analyze(state)
assert "class_performance" in analysis
assert analysis["class_performance"]["average_score"] == 78.5
assert analysis["class_performance"]["student_count"] == 3
assert analysis["class_performance"]["student_count"] == 2
assert "weak_students" in analysis
async def test_step1_analyze_no_client_degraded(self) -> None:
@@ -214,13 +226,14 @@ class TestLessonPlanWorkflowSteps:
svc = _make_service(content_client=ContentClientMock())
state = _make_state()
kps = await svc._step2_recommend(state)
assert len(kps) == 3
assert len(kps) == 2
assert kps[0]["id"] == "kp_001"
async def test_step2_recommend_no_client_fallback(self) -> None:
svc = _make_service(content_client=None)
state = _make_state(topic="函数")
kps = await svc._step2_recommend(state)
# 无 content_client 时降级到内置 3 个默认知识点
assert len(kps) == 3
assert "基础概念" in kps[0]["title"]
assert "函数" in kps[0]["title"]
@@ -233,7 +246,8 @@ class TestLessonPlanWorkflowSteps:
)
state = _make_state(question_count=1, target_difficulty="medium")
questions = await svc._step3_generate(
state, [{"id": "kp_1", "title": "KP1"}],
state,
[{"id": "kp_1", "title": "KP1"}],
)
assert len(questions) == 1
assert questions[0].question == "什么是函数？"
@@ -245,7 +259,8 @@ class TestLessonPlanWorkflowSteps:
svc = _make_service(provider=provider)
state = _make_state(question_count=1)
questions = await svc._step3_generate(
state, [{"id": "kp_1", "title": "KP1"}],
state,
[{"id": "kp_1", "title": "KP1"}],
)
assert len(questions) == 1
assert questions[0].degraded is True

27
services/ai/tests/test_main_app.py
View File

@@ -37,17 +37,24 @@ from src.ai.middleware.error_handler import (
@pytest.fixture
async def client() -> AsyncGenerator[httpx.AsyncClient, None]:
"""HTTP client wired to the FastAPI app (dev_mode=True, permissions skipped)."""
from src.ai.main import _permission_guard, app
"""HTTP client wired to the FastAPI app (dev_mode=True, permissions skipped).
注入 ContentClientMock 到 workflow_service，因为测试环境无真实 content gRPC server。
"""
from src.ai.clients import ContentClientMock
from src.ai.main import _permission_guard, _workflow_service, app
original = _permission_guard._dev_mode
original_content = _workflow_service._content_client # noqa: SLF001
_permission_guard._dev_mode = True
_workflow_service._content_client = ContentClientMock() # noqa: SLF001
try:
transport = ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
yield c
finally:
_permission_guard._dev_mode = original
_workflow_service._content_client = original_content # noqa: SLF001
@pytest.fixture
@@ -136,14 +143,14 @@ async def test_handle_unknown_error() -> None:
async def test_grpc_error_mapper_ai_error() -> None:
"""grpc_error_mapper maps AIError to correct gRPC status code."""
cases = [
(ErrorCode.AI_UNAUTHORIZED, 8), # UNAUTHENTICATED
(ErrorCode.AI_FORBIDDEN, 7), # PERMISSION_DENIED
(ErrorCode.AI_RATE_LIMITED, 9), # RESOURCE_EXHAUSTED
(ErrorCode.AI_QUOTA_EXCEEDED, 9), # RESOURCE_EXHAUSTED
(ErrorCode.AI_INVALID_MODEL, 3), # INVALID_ARGUMENT
(ErrorCode.AI_UNAUTHORIZED, 8), # UNAUTHENTICATED
(ErrorCode.AI_FORBIDDEN, 7), # PERMISSION_DENIED
(ErrorCode.AI_RATE_LIMITED, 9), # RESOURCE_EXHAUSTED
(ErrorCode.AI_QUOTA_EXCEEDED, 9), # RESOURCE_EXHAUSTED
(ErrorCode.AI_INVALID_MODEL, 3), # INVALID_ARGUMENT
(ErrorCode.AI_WORKFLOW_NOT_FOUND, 5), # NOT_FOUND
(ErrorCode.AI_WORKFLOW_STATE_INVALID, 10), # FAILED_PRECONDITION
(ErrorCode.AI_INTERNAL_ERROR, 13), # INTERNAL
(ErrorCode.AI_INTERNAL_ERROR, 13), # INTERNAL
]
for code, expected_grpc_status in cases:
exc = AIError(code, f"test {code.value}")
@@ -390,9 +397,7 @@ async def test_confirm_lesson_plan_success(client: httpx.AsyncClient) -> None:
break
await asyncio.sleep(0.1)
assert status == "pending_review", (
f"Workflow did not reach pending_review, got: {status}"
)
assert status == "pending_review", f"Workflow did not reach pending_review, got: {status}"
resp = await client.post(f"/v1/ai/lesson-plan/confirm/{workflow_id}")
assert resp.status_code == 200