aios/tests/test_search_gateway_general_capability.py

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)