215 lines
7.8 KiB
Python
215 lines
7.8 KiB
Python
"""叩心枢（KnockingPivot）集成测试。
|
||
|
||
测试覆盖：
|
||
1. Tick 触发 —— 活跃角色条件匹配，执行完整触发链并推送一炁事件
|
||
2. Tick 空跑 —— 无活跃角色时跳过
|
||
3. 反 spam 冷却 —— 触发后冷却期内 ``_can_trigger`` 返回 False
|
||
4. 反 spam 日上限 —— 达到日上限后返回 False
|
||
5. 静默时段 —— [23:00, 08:00) 返回 True，其余返回 False
|
||
6. 活跃角色列表 —— DB 查询正确返回 ``status='active'`` 的角色 ID
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import AsyncGenerator
|
||
from datetime import UTC, datetime, timedelta
|
||
from unittest.mock import AsyncMock, patch
|
||
|
||
import pytest
|
||
import pytest_asyncio
|
||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||
|
||
from app.db.base import Base
|
||
|
||
# ── 模型注册 ──────────────────────────────────────────────────────────
|
||
# 在 DB 表创建前将模型注册到 Base.metadata，确保 create_all 创建对应表。
|
||
# conftest.py 已注册 ChatMessage / CuriosityQuestion / CharacterEmotionState / TimeAnniversary，
|
||
# 但未注册 Character 与 ChatSession。
|
||
from app.models.character import Character # noqa: F401
|
||
from app.models.chat import ChatSession # noqa: F401
|
||
from app.modules.knocking.conditions import ConditionResult
|
||
from app.modules.knocking.opener import OpenerResult
|
||
from app.modules.knocking.pivot import KnockingPivot
|
||
from app.services.personality import PersonalitySummary
|
||
|
||
_TEST_DATABASE_URL = "sqlite+aiosqlite://"
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def keeper_session_factory() -> AsyncGenerator[async_sessionmaker[AsyncSession], None]:
|
||
"""Yield an async_sessionmaker backed by an in-memory SQLite database."""
|
||
engine = create_async_engine(_TEST_DATABASE_URL, echo=False)
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
factory: async_sessionmaker[AsyncSession] = async_sessionmaker(
|
||
bind=engine,
|
||
class_=AsyncSession,
|
||
expire_on_commit=False,
|
||
)
|
||
yield factory
|
||
await engine.dispose()
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def dbevent_store() -> AsyncGenerator:
|
||
"""In-memory event store (no DB needed, supports _events for assertion)."""
|
||
from app.modules.runtime.events import InMemoryEventStore
|
||
|
||
yield InMemoryEventStore()
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def keeper(
|
||
keeper_session_factory: async_sessionmaker[AsyncSession],
|
||
dbevent_store,
|
||
) -> AsyncGenerator[KnockingPivot, None]:
|
||
"""KnockingPivot instance backed by an in-memory SQLite database."""
|
||
pivot = KnockingPivot(
|
||
event_store=dbevent_store,
|
||
session_factory=keeper_session_factory,
|
||
tick_interval=3600, # 测试期间不真的 tick
|
||
)
|
||
yield pivot
|
||
|
||
|
||
# ======================================================================
|
||
# 测试
|
||
# ======================================================================
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_tick_triggers_for_active_character(
|
||
keeper: KnockingPivot,
|
||
keeper_session_factory: async_sessionmaker[AsyncSession],
|
||
) -> None:
|
||
"""活跃角色 + 任意条件触发时，应完成完整触发链：开场 → 消息 → 事件。"""
|
||
async with keeper_session_factory() as session:
|
||
char = Character(
|
||
id="char_test_1",
|
||
name="测试人物",
|
||
status="active",
|
||
owner_scope="test",
|
||
language="zh-CN",
|
||
privacy_level="normal",
|
||
)
|
||
session.add(char)
|
||
await session.commit()
|
||
|
||
with (
|
||
patch.object(keeper.conditions[0], "check", AsyncMock(return_value=ConditionResult(triggered=True, trigger_reason="测试触发", condition_name="test"))),
|
||
patch(
|
||
"app.modules.knocking.pivot.build_personality_summary",
|
||
return_value=PersonalitySummary(text="你是一个测试人物。"),
|
||
),
|
||
patch(
|
||
"app.modules.knocking.pivot.generate_opener",
|
||
return_value=OpenerResult(text="测试开场白。"),
|
||
),
|
||
):
|
||
async with keeper_session_factory() as check_session:
|
||
processed = await keeper._check_and_trigger(
|
||
"char_test_1",
|
||
check_session,
|
||
now=datetime.now(UTC),
|
||
)
|
||
|
||
assert processed is True
|
||
|
||
# 验证消息已创建
|
||
async with keeper_session_factory() as session:
|
||
from sqlalchemy import select
|
||
|
||
result = await session.execute(
|
||
select(ChatSession).where(ChatSession.character_id == "char_test_1")
|
||
)
|
||
assert result.scalar_one_or_none() is not None
|
||
|
||
# 验证一炁事件已推送
|
||
events = keeper._event_store._events
|
||
assert len(events) > 0
|
||
last_event = events[-1]
|
||
assert last_event.source == "knocking_pivot"
|
||
assert last_event.character_id == "char_test_1"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_tick_skips_when_no_active_characters(keeper: KnockingPivot) -> None:
|
||
"""无活跃角色时 tick 不触发任何操作。"""
|
||
with (
|
||
patch("app.modules.knocking.pivot.create_proactive_message") as mock_create,
|
||
patch("app.modules.knocking.pivot.generate_opener"),
|
||
):
|
||
await keeper._tick()
|
||
|
||
mock_create.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cooldown_prevents_retrigger(keeper: KnockingPivot) -> None:
|
||
"""触发后冷却期内 _can_trigger 返回 False。"""
|
||
now = datetime.now(UTC)
|
||
assert keeper._can_trigger("char_1", now) is True
|
||
|
||
keeper._mark_triggered("char_1", now)
|
||
assert keeper._can_trigger("char_1", now) is False
|
||
|
||
# 冷却结束后应恢复
|
||
after_cooldown = now + timedelta(minutes=keeper._cooldown_minutes + 1)
|
||
assert keeper._can_trigger("char_1", after_cooldown) is True
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_daily_limit_prevents_excess(keeper: KnockingPivot) -> None:
|
||
"""同一天达到上限后 _can_trigger 返回 False。"""
|
||
now = datetime.now(UTC)
|
||
assert keeper._can_trigger("char_1", now) is True
|
||
|
||
keeper._daily_counts["char_1"] = (now.date().isoformat(), keeper._daily_limit)
|
||
assert keeper._can_trigger("char_1", now) is False
|
||
|
||
# 第二天恢复
|
||
next_day = now + timedelta(days=1)
|
||
assert keeper._can_trigger("char_1", next_day) is True
|
||
|
||
|
||
def test_quiet_hours(keeper: KnockingPivot) -> None:
|
||
"""静默时段 23:00–08:00 _in_quiet_hours 返回 True。"""
|
||
keeper._quiet_hours_start = 23
|
||
keeper._quiet_hours_end = 8
|
||
|
||
# 静默时段内
|
||
assert keeper._in_quiet_hours(datetime(2026, 1, 1, 23, 30, tzinfo=UTC)) is True
|
||
assert keeper._in_quiet_hours(datetime(2026, 1, 1, 3, 0, tzinfo=UTC)) is True
|
||
assert keeper._in_quiet_hours(datetime(2026, 1, 1, 7, 59, tzinfo=UTC)) is True
|
||
|
||
# 静默时段外
|
||
assert keeper._in_quiet_hours(datetime(2026, 1, 1, 8, 0, tzinfo=UTC)) is False
|
||
assert keeper._in_quiet_hours(datetime(2026, 1, 1, 12, 0, tzinfo=UTC)) is False
|
||
assert keeper._in_quiet_hours(datetime(2026, 1, 1, 22, 59, tzinfo=UTC)) is False
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_active_characters(
|
||
keeper: KnockingPivot,
|
||
keeper_session_factory: async_sessionmaker[AsyncSession],
|
||
) -> None:
|
||
"""_list_active_characters 只返回 status='active' 的角色。"""
|
||
async with keeper_session_factory() as session:
|
||
for i, status in enumerate(["active", "inactive", "paused", "active"]):
|
||
session.add(
|
||
Character(
|
||
id=f"char_{i}",
|
||
name=f"人物{i}",
|
||
status=status,
|
||
owner_scope="test",
|
||
language="zh-CN",
|
||
privacy_level="normal",
|
||
)
|
||
)
|
||
await session.commit()
|
||
|
||
async with keeper_session_factory() as session:
|
||
result = await keeper._list_active_characters(session)
|
||
|
||
assert set(result) == {"char_0", "char_3"}
|