67 lines
1.9 KiB
Python
67 lines
1.9 KiB
Python
from app.search_gateway import (
|
|
SearchEvidence,
|
|
SearchGateway,
|
|
SearchProviderAttempt,
|
|
SearchRequest,
|
|
)
|
|
|
|
|
|
class FailingProvider:
|
|
name = "failing-provider"
|
|
|
|
def search(self, request, queries):
|
|
raise RuntimeError("provider unavailable")
|
|
|
|
|
|
class SuccessProvider:
|
|
name = "success-provider"
|
|
|
|
def search(self, request, queries):
|
|
return [
|
|
SearchEvidence(
|
|
title="public result",
|
|
url="https://example.com/public",
|
|
source="example",
|
|
snippet="public evidence",
|
|
confidence="high",
|
|
)
|
|
], SearchProviderAttempt(provider=self.name, status="ok", result_count=1)
|
|
|
|
|
|
def test_provider_failure_falls_back_to_next_provider():
|
|
gateway = SearchGateway(providers=[FailingProvider(), SuccessProvider()])
|
|
result = gateway.search(SearchRequest(query="public data", requester="test"))
|
|
assert result.status == "found"
|
|
assert len(result.attempts) == 2
|
|
assert result.attempts[0].status == "failed"
|
|
assert result.confidence == "high"
|
|
|
|
|
|
def test_private_student_data_is_blocked_without_authorized_file():
|
|
gateway = SearchGateway()
|
|
result = gateway.search(
|
|
SearchRequest(
|
|
query="查询学生个人成绩",
|
|
requester="test",
|
|
data_type="personal_grade",
|
|
user_scope="public",
|
|
)
|
|
)
|
|
assert result.status == "blocked"
|
|
assert result.blocked_reasons
|
|
|
|
|
|
def test_education_exam_not_found_requests_clarification():
|
|
gateway = SearchGateway()
|
|
result = gateway.search(
|
|
SearchRequest(
|
|
query="湖南2025年秋季高中第一次月考数据",
|
|
requester="test",
|
|
intent="education_exam_data",
|
|
region="湖南",
|
|
time_range="2025秋季",
|
|
)
|
|
)
|
|
assert result.status == "not_found"
|
|
assert "学校名称" in result.need_clarification[0]
|