94 lines
3.3 KiB
Python
94 lines
3.3 KiB
Python
"""灵核情绪状态数据访问测试（设计 §11.4）。"""
|
|
|
|
import pytest
|
|
|
|
from app.db.session import AsyncSessionLocal
|
|
from app.models.personality_emotion import CharacterEmotionState
|
|
from app.repositories import personality_emotion as emotion_repo
|
|
from app.services.characters import get_active_character_model
|
|
|
|
|
|
def _state(character_id: str, **overrides) -> CharacterEmotionState:
|
|
base = {
|
|
"character_id": character_id,
|
|
"mood": "touched",
|
|
"intensity": 0.7,
|
|
"baseline_mood": "neutral",
|
|
"baseline_intensity": 0.2,
|
|
"last_trigger_event_id": "evt_abc",
|
|
}
|
|
base.update(overrides)
|
|
return CharacterEmotionState(**base)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_state_returns_none_when_absent() -> None:
|
|
async with AsyncSessionLocal() as session:
|
|
character = await get_active_character_model(session)
|
|
assert await emotion_repo.get_state(session, character.id) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upsert_state_inserts_new() -> None:
|
|
async with AsyncSessionLocal() as session:
|
|
character = await get_active_character_model(session)
|
|
cid = character.id
|
|
created = await emotion_repo.upsert_state(session, _state(cid))
|
|
await session.commit()
|
|
assert created.mood == "touched"
|
|
assert created.intensity == 0.7
|
|
async with AsyncSessionLocal() as session:
|
|
fetched = await emotion_repo.get_state(session, cid)
|
|
assert fetched is not None
|
|
assert fetched.mood == "touched"
|
|
assert fetched.last_trigger_event_id == "evt_abc"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upsert_state_updates_existing() -> None:
|
|
async with AsyncSessionLocal() as session:
|
|
character = await get_active_character_model(session)
|
|
cid = character.id
|
|
await emotion_repo.upsert_state(session, _state(cid, mood="touched", intensity=0.7))
|
|
await emotion_repo.upsert_state(
|
|
session,
|
|
_state(
|
|
cid,
|
|
mood="concerned",
|
|
intensity=0.5,
|
|
last_trigger_event_id="evt_def",
|
|
),
|
|
)
|
|
await session.commit()
|
|
async with AsyncSessionLocal() as session:
|
|
fetched = await emotion_repo.get_state(session, cid)
|
|
assert fetched is not None
|
|
assert fetched.mood == "concerned"
|
|
assert fetched.intensity == 0.5
|
|
assert fetched.last_trigger_event_id == "evt_def"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upsert_state_single_row_per_character() -> None:
|
|
"""单行 per character：重复 upsert 不产生重复行。"""
|
|
async with AsyncSessionLocal() as session:
|
|
character = await get_active_character_model(session)
|
|
cid = character.id
|
|
await emotion_repo.upsert_state(session, _state(cid))
|
|
await emotion_repo.upsert_state(session, _state(cid))
|
|
await emotion_repo.upsert_state(session, _state(cid))
|
|
await session.commit()
|
|
async with AsyncSessionLocal() as session:
|
|
from sqlalchemy import func, select
|
|
|
|
count = int(
|
|
(
|
|
await session.execute(
|
|
select(func.count())
|
|
.select_from(CharacterEmotionState)
|
|
.where(CharacterEmotionState.character_id == cid)
|
|
)
|
|
).scalar_one()
|
|
)
|
|
assert count == 1
|