test_tutorial_07.py

"""教程 07 验收脚本：核心路径离线运行，FastAPI 路径按依赖自动启用。"""

from __future__ import annotations

import asyncio
import importlib.util
import io
import json
import sys
from pathlib import Path


if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")


ROOT = Path(__file__).parent
EXAMPLES = ROOT / "examples" / "07-production"
sys.path.insert(0, str(EXAMPLES))

import production_core as core  # noqa: E402
import production_service as service_module  # noqa: E402
import resilience  # noqa: E402
import streaming_sse_adapter as streaming  # noqa: E402


def load_example(filename: str):
    path = EXAMPLES / filename
    module_name = f"tutorial07_{path.stem}"
    spec = importlib.util.spec_from_file_location(module_name, path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"无法加载示例：{path}")
    module = importlib.util.module_from_spec(spec)
    sys.modules[module_name] = module
    spec.loader.exec_module(module)
    return module


def run_test(name: str, test_func) -> bool:
    print(f"\n【{name}】")
    print("-" * 60)
    try:
        test_func()
        print(f"✓ {name}通过")
        return True
    except Exception as exc:
        print(f"✗ {name}失败：{exc}")
        return False


fastapi_example = load_example("04_fastapi_service.py")
capacity = load_example("05_deployment_and_capacity.py")


def test_settings_validation() -> None:
    settings = core.Settings.from_mapping(
        {
            "ENVIRONMENT": "staging",
            "AGENT_API_KEY": "secret",
            "MAX_CONCURRENCY": "12",
            "MAX_RETRIES": "1",
        }
    )
    assert settings.max_concurrency == 12
    assert settings.max_retries == 1

    try:
        core.Settings.from_mapping({"ENVIRONMENT": "production"})
    except ValueError as exc:
        assert "AGENT_API_KEY" in str(exc)
    else:
        raise AssertionError("生产环境接受了默认 API Key")
    print("✓ 环境变量完成类型转换，生产默认密钥被拒绝")


def test_lifecycle_and_health() -> None:
    async def scenario() -> None:
        agent_service = service_module.ProductionAgentService(core.Settings())
        assert not agent_service.health()["ready"]
        await agent_service.startup()
        assert agent_service.health()["live"]
        assert agent_service.health()["ready"]
        await agent_service.shutdown()
        assert agent_service.health()["live"]
        assert not agent_service.health()["ready"]

    asyncio.run(scenario())
    print("✓ startup 后就绪，shutdown 前先摘除就绪状态")


def test_service_success_and_metrics() -> None:
    async def scenario() -> None:
        agent_service = service_module.ProductionAgentService(core.Settings())
        await agent_service.startup()
        result = await agent_service.invoke("查询 ORD1001", "req-success")
        assert result.intent == "order_status"
        assert "明天送达" in result.answer
        assert result.attempts == 1
        snapshot = agent_service.metrics.snapshot()
        assert snapshot["requests_total"] == 1.0
        assert snapshot["requests_success"] == 1.0
        await agent_service.shutdown()

    asyncio.run(scenario())
    print("✓ 异步服务返回稳定契约并记录成功指标")


def test_retry_and_timeout() -> None:
    async def scenario() -> None:
        retry_agent = core.AsyncRuleAgent(failures_before_success=2)
        retry_service = service_module.ProductionAgentService(
            core.Settings(max_retries=2), core.ApplicationResources(retry_agent)
        )
        await retry_service.startup()
        result = await retry_service.invoke("退款政策", "req-retry")
        assert result.attempts == 3
        assert retry_agent.calls == 3
        await retry_service.shutdown()

        slow_service = service_module.ProductionAgentService(
            core.Settings(request_timeout_seconds=0.005, max_retries=0),
            core.ApplicationResources(core.AsyncRuleAgent(delay_seconds=0.05)),
        )
        await slow_service.startup()
        try:
            await slow_service.invoke("退款政策", "req-timeout")
        except TimeoutError:
            pass
        else:
            raise AssertionError("慢请求未按 deadline 终止")
        assert slow_service.metrics.snapshot()["requests_error"] == 1.0
        await slow_service.shutdown()

    asyncio.run(scenario())
    print("✓ 临时错误有限重试，慢请求由统一 deadline 终止")


def test_backpressure() -> None:
    async def scenario() -> None:
        settings = core.Settings(
            max_concurrency=1,
            acquire_timeout_seconds=0.005,
            request_timeout_seconds=1.0,
        )
        agent_service = service_module.ProductionAgentService(
            settings,
            core.ApplicationResources(core.AsyncRuleAgent(delay_seconds=0.05)),
        )
        await agent_service.startup()
        first = asyncio.create_task(agent_service.invoke("退款政策", "req-1"))
        while agent_service.gate.active == 0:
            await asyncio.sleep(0)
        try:
            await agent_service.invoke("VPN 怎么处理？", "req-2")
        except resilience.BusyError:
            pass
        else:
            raise AssertionError("并发已满时没有快速拒绝")
        await first
        assert agent_service.gate.peak == 1
        assert agent_service.metrics.snapshot()["requests_error"] == 1.0
        await agent_service.shutdown()

    asyncio.run(scenario())
    print("✓ 并发上限为 1，排队超时转换为背压错误")


def test_circuit_breaker() -> None:
    async def scenario() -> None:
        agent = core.AsyncRuleAgent(failures_before_success=10)
        settings = core.Settings(
            max_retries=0,
            circuit_failure_threshold=2,
            circuit_recovery_seconds=60,
        )
        agent_service = service_module.ProductionAgentService(
            settings, core.ApplicationResources(agent)
        )
        await agent_service.startup()
        for index in range(2):
            try:
                await agent_service.invoke("退款", f"req-fail-{index}")
            except core.TransientAgentError:
                pass
        assert agent_service.breaker.state == "open"
        try:
            await agent_service.invoke("退款", "req-open")
        except resilience.CircuitOpenError:
            pass
        else:
            raise AssertionError("熔断器打开后仍调用上游")
        assert agent.calls == 2
        await agent_service.shutdown()

    asyncio.run(scenario())
    print("✓ 连续失败打开熔断器，后续请求不会继续冲击上游")


def test_idempotency_cache() -> None:
    async def scenario() -> None:
        agent = core.AsyncRuleAgent()
        agent_service = service_module.ProductionAgentService(
            core.Settings(), core.ApplicationResources(agent)
        )
        await agent_service.startup()
        first = await agent_service.invoke("退款政策", "req-a", "idem-1")
        second = await agent_service.invoke("退款政策", "req-b", "idem-1")
        assert not first.cached
        assert second.cached
        assert second.request_id == "req-b"
        assert agent.calls == 1
        try:
            await agent_service.invoke("查询 ORD1001", "req-c", "idem-1")
        except resilience.IdempotencyConflictError:
            pass
        else:
            raise AssertionError("幂等键被不同请求复用")
        await agent_service.shutdown()

    asyncio.run(scenario())
    print("✓ 重复请求命中缓存，不同请求复用幂等键返回冲突")


def test_streaming_protocol_and_disconnect() -> None:
    async def scenario() -> None:
        agent_service = service_module.ProductionAgentService(core.Settings())
        await agent_service.startup()
        events = [
            event
            async for event in streaming.stream_agent_events(
                agent_service.stream_tokens("退款政策"), "req-stream"
            )
        ]
        assert "event: start" in events[0]
        assert "event: end" in events[-1]
        data_line = next(line for line in events[1].splitlines() if line.startswith("data:"))
        assert "text" in json.loads(data_line.removeprefix("data: "))
        assert agent_service.metrics.snapshot()["requests_success"] == 1.0

        checks = 0

        async def disconnect_early() -> bool:
            nonlocal checks
            checks += 1
            return checks >= 2

        disconnected = [
            event
            async for event in streaming.stream_agent_events(
                agent_service.stream_tokens("VPN 怎么处理？"),
                "req-disconnect",
                disconnect_early,
            )
        ]
        assert "event: end" not in "".join(disconnected)
        await agent_service.shutdown()

    asyncio.run(scenario())
    print("✓ SSE start/token/end 协议正确，客户端断连停止生成")


def test_graceful_shutdown() -> None:
    async def scenario() -> None:
        settings = core.Settings(shutdown_grace_seconds=0.2)
        agent_service = service_module.ProductionAgentService(
            settings,
            core.ApplicationResources(core.AsyncRuleAgent(delay_seconds=0.03)),
        )
        await agent_service.startup()
        request = asyncio.create_task(agent_service.invoke("退款", "req-active"))
        while agent_service.gate.active == 0:
            await asyncio.sleep(0)
        await agent_service.shutdown()
        result = await request
        assert result.intent == "refund"
        assert not agent_service.resources.ready
        assert agent_service.resources.agent.closed

    asyncio.run(scenario())
    print("✓ 关闭时先停止接流量，再等待在途请求并释放 Agent")


def test_capacity_plan() -> None:
    slots = capacity.required_concurrency(12.0, 1.8, 0.70)
    replicas = capacity.recommended_replicas(slots, 1, 8)
    plan = capacity.DeploymentPlan(replicas, 1, 8, 30, 45)
    plan.validate()
    assert slots == 31
    assert replicas == 4
    assert plan.max_in_flight == 32
    try:
        capacity.DeploymentPlan(1, 1, 8, 30, 20).validate()
    except ValueError as exc:
        assert "宽限时间" in str(exc)
    else:
        raise AssertionError("终止宽限时间小于请求超时仍被接受")
    print("✓ Little's Law 容量估算和终止宽限约束正确")


def test_fastapi_boundary() -> None:
    if fastapi_example.FastAPI is None:
        assert fastapi_example.app is None
        try:
            fastapi_example.create_app()
        except RuntimeError as exc:
            assert "requirements.txt" in str(exc)
        else:
            raise AssertionError("缺少 FastAPI 时未给出明确错误")
        print("⚠️ 未安装 FastAPI，已验证可选依赖降级路径")
        return

    from fastapi.testclient import TestClient

    app = fastapi_example.create_app(core.Settings(environment="test", api_key="test-key"))
    with TestClient(app) as client:
        assert client.get("/health/live").status_code == 200
        assert client.get("/health/ready").status_code == 200
        unauthorized = client.post("/v1/agent/invoke", json={"question": "退款"})
        assert unauthorized.status_code == 401
        response = client.post(
            "/v1/agent/invoke",
            json={"question": "退款政策"},
            headers={"X-API-Key": "test-key", "X-Request-ID": "req-http"},
        )
        assert response.status_code == 200
        assert response.json()["request_id"] == "req-http"
        assert response.headers["X-Request-ID"] == "req-http"
    print("✓ FastAPI 健康检查、认证、请求 ID 和 invoke 路由正确")


if __name__ == "__main__":
    print("=" * 80)
    print("教程 07 代码验证")
    print("=" * 80)
    tests = [
        ("配置校验", test_settings_validation),
        ("生命周期与健康检查", test_lifecycle_and_health),
        ("服务响应与指标", test_service_success_and_metrics),
        ("有限重试与超时", test_retry_and_timeout),
        ("并发背压", test_backpressure),
        ("熔断器", test_circuit_breaker),
        ("幂等缓存", test_idempotency_cache),
        ("SSE 与断连", test_streaming_protocol_and_disconnect),
        ("优雅关闭", test_graceful_shutdown),
        ("容量规划", test_capacity_plan),
        ("FastAPI HTTP 边界", test_fastapi_boundary),
    ]
    results = [(name, run_test(name, func)) for name, func in tests]

    print("\n" + "=" * 80)
    print("测试总结")
    print("=" * 80)
    for name, passed in results:
        print(f"{'✅ PASS' if passed else '❌ FAIL'} - {name}")
    if not all(passed for _, passed in results):
        sys.exit(1)
    print("\n🎉 所有可用测试通过！教程 07 代码验证完成。")