52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
from uuid import uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.core.database import AsyncSessionLocal, init_db
|
|
from app.models.reading import Reading
|
|
from app.models.user import User
|
|
from app.services.bazi_backfill_service import backfill_bazi_wuxing_balance
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_bazi_wuxing_balance_updates_missing_chart_field():
|
|
await init_db()
|
|
user_id = str(uuid4())
|
|
reading_id = str(uuid4())
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
session.add(User(id=user_id, openid=f"backfill-{uuid4()}"))
|
|
session.add(
|
|
Reading(
|
|
id=reading_id,
|
|
user_id=user_id,
|
|
reading_type="bazi",
|
|
status="completed",
|
|
input_data={
|
|
"chart": {
|
|
"pillars": {
|
|
"year": "壬申",
|
|
"month": "戊申",
|
|
"day": "丙寅",
|
|
"time": "癸巳",
|
|
}
|
|
}
|
|
},
|
|
report_data={"overall_summary": "keep me"},
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
count = await backfill_bazi_wuxing_balance(session)
|
|
await session.commit()
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
result = await session.execute(select(Reading).where(Reading.id == reading_id))
|
|
reading = result.scalar_one()
|
|
|
|
assert count >= 1
|
|
assert reading.input_data["chart"]["wuxing_balance"] == {"木": 1, "火": 2, "土": 1, "金": 2, "水": 2}
|
|
assert reading.report_data == {"overall_summary": "keep me"}
|