88 lines
3.2 KiB
Python
88 lines
3.2 KiB
Python
"""灵核情绪状态机端到端集成测试：Dispatcher + handle_time_event + DB。
|
|
|
|
验证一炁订阅基础设施（计划 5）+ 灵核情绪状态机（设计 §11）全链路：
|
|
EventStore.append(time.reunion) → Dispatcher 分发 → handle_time_event → DB 情绪状态更新。
|
|
这是 main.py lifespan ``dispatcher.register("time.*", handle_time_event)`` 的端到端验证。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from app.db.session import AsyncSessionLocal
|
|
from app.models.character import Character
|
|
from app.models.personality import PersonalityVersion
|
|
from app.modules.runtime.events import Event, InMemoryEventStore
|
|
from app.repositories.personality_emotion import get_state
|
|
from app.services.characters import get_active_character_model
|
|
from app.services.dispatcher import Dispatcher
|
|
from app.services.personality_emotion import handle_time_event
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatcher_drives_emotion_state_update() -> None:
|
|
"""append time.reunion → dispatcher 分发 → handle_time_event 写入 DB。"""
|
|
async with AsyncSessionLocal() as session:
|
|
character = await get_active_character_model(session)
|
|
cid = character.id
|
|
pv = PersonalityVersion(
|
|
character_id=cid,
|
|
identity_profile="测试身份",
|
|
emotion_baseline='{"mood": "neutral", "intensity": 0.2}',
|
|
)
|
|
session.add(pv)
|
|
await session.flush()
|
|
character.current_personality_version_id = pv.id
|
|
await session.commit()
|
|
|
|
store = InMemoryEventStore()
|
|
dispatcher = Dispatcher(store)
|
|
dispatcher.register("time.*", handle_time_event)
|
|
await dispatcher.start()
|
|
try:
|
|
await store.append(
|
|
Event(
|
|
type="time.reunion",
|
|
source="time_sense",
|
|
payload={"reunion_intensity": 0.6, "offline_band": "reunion_impact"},
|
|
character_id=cid,
|
|
)
|
|
)
|
|
await asyncio.sleep(0.1) # 等 pump + consumer 异步完成
|
|
finally:
|
|
await dispatcher.stop()
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
row = await get_state(session, cid)
|
|
assert row is not None
|
|
assert row.mood == "touched"
|
|
assert abs(row.intensity - 0.8) < 0.001 # 0.2 baseline + 0.6 delta
|
|
character = await session.get(Character, cid)
|
|
assert character is not None
|
|
assert character.current_mood_id == "touched"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatcher_ignores_non_time_events() -> None:
|
|
"""非 time.* 事件不触发 handle_time_event（无副作用：无 state row 写入）。"""
|
|
async with AsyncSessionLocal() as session:
|
|
character = await get_active_character_model(session)
|
|
cid = character.id
|
|
await session.commit()
|
|
|
|
store = InMemoryEventStore()
|
|
dispatcher = Dispatcher(store)
|
|
dispatcher.register("time.*", handle_time_event)
|
|
await dispatcher.start()
|
|
try:
|
|
await store.append(Event(type="run.started", source="run", character_id=cid))
|
|
await asyncio.sleep(0.05)
|
|
finally:
|
|
await dispatcher.stop()
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
# 非 time.* 事件 → handler 不被调用 → 无 state row
|
|
assert await get_state(session, cid) is None
|