zhifu-platform/tests/mixed_sustained_acceptance.py

import asyncio
import json
from pathlib import Path

from mixed_capacity_acceptance import mixed_admission
from test_after_sales import scenario  # noqa: F401


async def test_sustained_tenant_reservation_with_knowledge(scenario):  # noqa: F811
    # 32轮各等待1秒，验证持续竞争；生成调用为零，总执行期限由共用测试限制为60秒。
    await mixed_admission(
        scenario,
        rounds=32,
        b_requests=32,
        knowledge_queries=32,
        pause=1,
        report_path=".work/mixed-sustained.json",
    )
    report = json.loads(
        await asyncio.to_thread(Path(".work/mixed-sustained.json").read_text, encoding="utf-8")
    )
    assert 32 <= report["load_seconds"] < 60
    assert report["a_statuses"] == {"429": 256}
    assert report["b_accepted"] == 32 and report["generation_attempts"] == 0