49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
"""司命台 time.* 事件展示测试（辰正 §8 司命台接入）。"""
|
|
|
|
import pytest
|
|
|
|
from app.db.session import AsyncSessionLocal
|
|
from app.models.runtime import Event as EventRow
|
|
from app.repositories.json_utils import dumps
|
|
from app.services.admin import list_time_events
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_time_events_filters_by_source():
|
|
"""list_time_events 只返回 source='time_sense' 的事件。"""
|
|
async with AsyncSessionLocal() as session:
|
|
time_ev = EventRow(
|
|
id="event_time1",
|
|
type="time.reunion",
|
|
source="time_sense",
|
|
payload_json=dumps({"offline_minutes": 7200}),
|
|
visibility="internal",
|
|
sensitivity="medium",
|
|
run_id="run_x",
|
|
character_id="char_x",
|
|
trace_id="t1",
|
|
sequence=1,
|
|
evidence_ids_json="[]",
|
|
)
|
|
other_ev = EventRow(
|
|
id="event_other1",
|
|
type="run.started",
|
|
source="model",
|
|
payload_json=dumps({}),
|
|
visibility="internal",
|
|
sensitivity="low",
|
|
run_id="run_x",
|
|
character_id="char_x",
|
|
trace_id="t2",
|
|
sequence=2,
|
|
evidence_ids_json="[]",
|
|
)
|
|
session.add_all([time_ev, other_ev])
|
|
await session.commit()
|
|
|
|
items, total = await list_time_events(limit=50, offset=0)
|
|
types = [e.type for e in items]
|
|
assert "time.reunion" in types
|
|
assert "run.started" not in types
|
|
assert total >= 1
|