| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657 |
- import asyncio
- import json
- import httpx
- from standards_service.app import create_app
- def post_json(app, path, payload):
- async def send_request():
- transport = httpx.ASGITransport(app=app)
- async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
- return await client.post(path, json=payload)
- return asyncio.run(send_request())
- def test_batch_api_preserves_order_and_writes_one_log_per_query(standards_csv, tmp_path):
- log_path = tmp_path / "query.jsonl"
- app = create_app(csv_path=standards_csv, log_path=log_path)
- response = post_json(
- app,
- "/v1/search",
- {"keywords": ["GB 50330-2013", "本地规范库中不存在的规范"]},
- )
- assert response.status_code == 200
- payload = response.json()
- assert [item["query"] for item in payload["results"]] == [
- "GB 50330-2013",
- "本地规范库中不存在的规范",
- ]
- assert payload["results"][0]["candidates"][0]["code"] == "GB 50330-2013"
- assert payload["results"][1]["candidates"] == []
- entries = [json.loads(line) for line in log_path.read_text(encoding="utf-8").splitlines()]
- assert len(entries) == 2
- assert entries[0]["query"] == "GB 50330-2013"
- assert entries[0]["candidates"][0]["code"] == "GB 50330-2013"
- assert entries[1]["candidates"] == []
- assert "+" in entries[0]["time"] or entries[0]["time"].endswith("Z")
- assert entries[0]["elapsed_ms"] >= 0
- def test_service_exposes_only_one_route(standards_csv, tmp_path):
- app = create_app(csv_path=standards_csv, log_path=tmp_path / "query.jsonl")
- assert [route.path for route in app.routes] == ["/v1/search"]
- def test_api_rejects_more_than_one_hundred_keywords(standards_csv, tmp_path):
- app = create_app(csv_path=standards_csv, log_path=tmp_path / "query.jsonl")
- response = post_json(app, "/v1/search", {"keywords": ["GB 1"] * 101})
- assert response.status_code == 422
|
