af279bedd9
- 新增 Opinion 模型，记录角色对事件的观点 - 新增 opinions.py，基于规则生成观点（支持5种事件×3种情绪） - 同一事件生命周期内每个 agent 只生成一次观点 - 观点同时记录到 agent.memory - 新增 3 个测试用例 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
232 lines
7.0 KiB
Python
232 lines
7.0 KiB
Python
import pytest
|
||
from fastapi.testclient import TestClient
|
||
import os
|
||
import json
|
||
from pathlib import Path
|
||
|
||
from app.main import app
|
||
from app.state import STATE_FILE, get_default_state, save_state
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def reset_state():
|
||
"""每个测试前重置状态文件"""
|
||
if STATE_FILE.exists():
|
||
STATE_FILE.unlink()
|
||
save_state(get_default_state())
|
||
yield
|
||
if STATE_FILE.exists():
|
||
STATE_FILE.unlink()
|
||
|
||
|
||
@pytest.fixture
|
||
def client():
|
||
return TestClient(app)
|
||
|
||
|
||
def test_step_tick_increments(client):
|
||
"""测试 tick 递增"""
|
||
# 第一次调用
|
||
resp1 = client.post("/step", json={"events": []})
|
||
assert resp1.status_code == 200
|
||
data1 = resp1.json()
|
||
assert data1["world_state"]["tick"] == 1
|
||
|
||
# 第二次调用
|
||
resp2 = client.post("/step", json={"events": []})
|
||
assert resp2.status_code == 200
|
||
data2 = resp2.json()
|
||
assert data2["world_state"]["tick"] == 2
|
||
|
||
|
||
def test_mood_bounded(client):
|
||
"""测试 town_mood 有界 (-10 到 10)"""
|
||
# 发送多个"支持"事件，测试上限
|
||
events = [
|
||
{"type": "comment", "text": "支持", "user": "test", "ts": i}
|
||
for i in range(15)
|
||
]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
assert data["world_state"]["town_mood"] <= 10
|
||
|
||
# 重置状态
|
||
save_state(get_default_state())
|
||
|
||
# 发送多个"混乱"事件，测试下限
|
||
events = [
|
||
{"type": "comment", "text": "混乱", "user": "test", "ts": i}
|
||
for i in range(15)
|
||
]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
assert data["world_state"]["town_mood"] >= -10
|
||
|
||
|
||
def test_memory_max_length(client):
|
||
"""测试 memory 长度不超过 3"""
|
||
events = [
|
||
{"type": "comment", "text": f"消息{i}", "user": "user1", "ts": i}
|
||
for i in range(10)
|
||
]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
|
||
for agent_id, agent in data["world_state"]["agents"].items():
|
||
assert len(agent["memory"]) <= 3
|
||
|
||
|
||
def test_weather_change(client):
|
||
"""测试天气变化"""
|
||
# 下雨
|
||
resp = client.post("/step", json={
|
||
"events": [{"type": "comment", "text": "下雨了", "user": "test", "ts": 1}]
|
||
})
|
||
assert resp.json()["world_state"]["weather"] == "rainy"
|
||
|
||
# 天晴
|
||
resp = client.post("/step", json={
|
||
"events": [{"type": "comment", "text": "天晴了", "user": "test", "ts": 2}]
|
||
})
|
||
assert resp.json()["world_state"]["weather"] == "sunny"
|
||
|
||
|
||
def test_actions_generated(client):
|
||
"""测试 actions 生成"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
assert len(data["actions"]) == 2
|
||
agent_ids = [a["agent_id"] for a in data["actions"]]
|
||
assert "alice" in agent_ids
|
||
assert "bob" in agent_ids
|
||
|
||
for action in data["actions"]:
|
||
assert "say" in action
|
||
assert "do" in action
|
||
|
||
|
||
def test_global_meter_accumulates(client):
|
||
"""测试能量累计"""
|
||
# comment 类型 +1
|
||
events = [{"type": "comment", "text": "hello", "user": "test", "ts": 1}]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
assert data["world_state"]["global_meter"]["value"] == 1
|
||
|
||
# support 类型 +3
|
||
events = [{"type": "support", "text": "支持", "user": "test", "ts": 2}]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
assert data["world_state"]["global_meter"]["value"] == 4
|
||
|
||
# chaos 类型 +5
|
||
events = [{"type": "chaos", "text": "混乱", "user": "test", "ts": 3}]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
assert data["world_state"]["global_meter"]["value"] == 9
|
||
|
||
|
||
def test_global_event_triggers(client):
|
||
"""测试世界级事件触发"""
|
||
# 发送足够多的 chaos 事件达到阈值 (100)
|
||
# chaos +5, 需要 20 个事件
|
||
events = [
|
||
{"type": "chaos", "text": f"chaos{i}", "user": "test", "ts": i}
|
||
for i in range(20)
|
||
]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
|
||
# 验证事件触发
|
||
assert data["global_event"]["triggered"] is True
|
||
assert data["global_event"]["event"] is not None
|
||
assert "name" in data["global_event"]["event"]
|
||
assert "description" in data["global_event"]["event"]
|
||
|
||
# 验证能量重置和冷却设置
|
||
assert data["world_state"]["global_meter"]["value"] == 0
|
||
assert data["world_state"]["global_meter"]["cooldown"] == 5
|
||
|
||
|
||
def test_global_event_cooldown(client):
|
||
"""测试冷却期间不触发事件"""
|
||
# 先触发一次事件
|
||
events = [
|
||
{"type": "chaos", "text": f"chaos{i}", "user": "test", "ts": i}
|
||
for i in range(20)
|
||
]
|
||
resp = client.post("/step", json={"events": events})
|
||
assert resp.json()["global_event"]["triggered"] is True
|
||
|
||
# 冷却期间再次达到阈值，不应触发
|
||
events = [
|
||
{"type": "chaos", "text": f"chaos{i}", "user": "test", "ts": i}
|
||
for i in range(20)
|
||
]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
|
||
# 不应触发（冷却中）
|
||
assert data["global_event"]["triggered"] is False
|
||
# 冷却递减
|
||
assert data["world_state"]["global_meter"]["cooldown"] == 4
|
||
|
||
|
||
def test_opinion_generated_on_world_effect(client):
|
||
"""测试世界事件触发时生成观点"""
|
||
# 触发世界事件
|
||
events = [
|
||
{"type": "chaos", "text": f"chaos{i}", "user": "test", "ts": i}
|
||
for i in range(20)
|
||
]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
|
||
# 验证事件触发
|
||
assert data["global_event"]["triggered"] is True
|
||
assert len(data["world_state"]["world_effects"]) > 0
|
||
|
||
# 验证每个 agent 都有 opinion
|
||
for agent_id, agent in data["world_state"]["agents"].items():
|
||
assert agent["opinion"] is not None
|
||
assert "about" in agent["opinion"]
|
||
assert "text" in agent["opinion"]
|
||
assert "tick" in agent["opinion"]
|
||
|
||
|
||
def test_opinion_only_once_per_effect(client):
|
||
"""测试同一事件只生成一次观点"""
|
||
# 触发世界事件
|
||
events = [
|
||
{"type": "chaos", "text": f"chaos{i}", "user": "test", "ts": i}
|
||
for i in range(20)
|
||
]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
|
||
# 记录第一次的观点
|
||
first_opinions = {
|
||
aid: agent["opinion"]["text"]
|
||
for aid, agent in data["world_state"]["agents"].items()
|
||
}
|
||
|
||
# 再次调用 step（事件仍在持续）
|
||
resp2 = client.post("/step", json={"events": []})
|
||
data2 = resp2.json()
|
||
|
||
# 观点应该保持不变（不重新生成）
|
||
for aid, agent in data2["world_state"]["agents"].items():
|
||
if agent["opinion"]:
|
||
assert agent["opinion"]["text"] == first_opinions[aid]
|
||
|
||
|
||
def test_opinion_cleared_when_no_effect(client):
|
||
"""测试无活跃事件时观点为空"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
# 无世界事件时，opinion 应为 None
|
||
for agent_id, agent in data["world_state"]["agents"].items():
|
||
assert agent["opinion"] is None
|