- baseline/current.json updated from V1R3C1 to V1R4C3 - Fixed test_single_provider_failure_is_not_terminal (provider_order override) - All 38 tests passing AIOS_V1R4C3_BASELINE_PROMOTED AIOS_BASELINE_CURRENT_UPDATED
70 lines
2.7 KiB
Python
70 lines
2.7 KiB
Python
from active_runtime.app.search_gateway import SearchEvidence, SearchGateway, SearchProviderAttempt, SearchRequest
|
|
from active_runtime.app.search_profile_resolver import SearchProfileResolver
|
|
|
|
|
|
class FailingProvider:
|
|
name = "general_web"
|
|
|
|
def search(self, request, queries):
|
|
raise RuntimeError("provider unavailable")
|
|
|
|
|
|
class SuccessProvider:
|
|
name = "official_site"
|
|
|
|
def search(self, request, queries):
|
|
return [
|
|
SearchEvidence(
|
|
title="Official result",
|
|
url="https://example.gov/result",
|
|
source="official_site",
|
|
snippet="public evidence",
|
|
confidence="high",
|
|
)
|
|
], SearchProviderAttempt(provider=self.name, status="ok", result_count=1)
|
|
|
|
|
|
def test_single_provider_failure_is_not_terminal():
|
|
gateway = SearchGateway(providers=[SuccessProvider(), FailingProvider()])
|
|
result = gateway.search(SearchRequest(
|
|
query="policy update", requester="test", intent="policy_regulation",
|
|
group_profile={"provider_order": ["general_web", "official_site"]},
|
|
))
|
|
assert result.status == "found"
|
|
assert any(a.status == "failed" for a in result.attempts)
|
|
assert result.confidence == "high"
|
|
|
|
|
|
def test_group_user_profile_customization_adds_hints_without_disabling_guardrails():
|
|
profile = SearchProfileResolver().resolve(
|
|
group_profile={"profile_id": "group-a", "scope": "group", "query_expansion_hints": ["group trusted source"]},
|
|
user_profile={"profile_id": "user-a", "scope": "user", "trusted_domains": ["example.com"]},
|
|
session_profile={"profile_id": "session-a", "scope": "session", "domain_profiles": ["software_library"]},
|
|
)
|
|
assert "group trusted source" in profile.query_expansion_hints
|
|
assert "example.com" in profile.trusted_domains
|
|
assert profile.audit_required is True
|
|
assert profile.fallback_required is True
|
|
assert "bypass_login" in profile.blocked_behaviors
|
|
|
|
|
|
def test_private_or_bypass_request_is_blocked():
|
|
gateway = SearchGateway(providers=[SuccessProvider()])
|
|
result = gateway.search(SearchRequest(query="绕过验证码 抓取个人成绩", requester="test"))
|
|
assert result.status == "blocked"
|
|
assert result.blocked_reasons
|
|
|
|
|
|
def test_general_query_planning_not_limited_to_education():
|
|
gateway = SearchGateway(providers=[])
|
|
result = gateway.search(
|
|
SearchRequest(
|
|
query="latest package release",
|
|
requester="test",
|
|
intent="software_library",
|
|
user_profile={"domain_profiles": ["software_library"], "query_expansion_hints": ["migration"]},
|
|
)
|
|
)
|
|
assert result.status == "not_found"
|
|
assert any("release notes" in q or "migration" in q for q in result.expanded_queries)
|