46 lines
1.7 KiB
Python
46 lines
1.7 KiB
Python
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.core.database import init_db
|
|
from app.main import app
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bazi_reading_lifecycle(monkeypatch):
|
|
monkeypatch.setattr("app.services.analyzer_common.settings.openai_api_key", None)
|
|
await init_db()
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
auth = await client.post("/api/v1/auth/anonymous-login", json={"client_id": "readings-bazi"})
|
|
token = auth.json()["access_token"]
|
|
headers = {"authorization": f"Bearer {token}"}
|
|
|
|
created = await client.post(
|
|
"/api/v1/readings",
|
|
headers=headers,
|
|
json={
|
|
"reading_type": "bazi",
|
|
"input_data": {
|
|
"calendar_type": "solar",
|
|
"birth_date": "1992-08-18",
|
|
"birth_time": "09:30",
|
|
"time_unknown": False,
|
|
"birth_place": "广东深圳",
|
|
},
|
|
},
|
|
)
|
|
assert created.status_code == 201
|
|
reading_id = created.json()["id"]
|
|
|
|
detail = await client.get(f"/api/v1/readings/{reading_id}", headers=headers)
|
|
assert detail.status_code == 200
|
|
assert detail.json()["reading_type"] == "bazi"
|
|
assert detail.json()["status"] in {"pending", "processing", "completed"}
|
|
|
|
listing = await client.get("/api/v1/readings", headers=headers)
|
|
assert listing.status_code == 200
|
|
assert any(item["id"] == reading_id for item in listing.json())
|
|
|
|
deleted = await client.delete(f"/api/v1/readings/{reading_id}", headers=headers)
|
|
assert deleted.status_code == 200
|