75e84f2ba3
- 新增 Votes 模型，包含 optimists/fearful 计数和 voted_users 去重列表 - 扩展 Event 模型，添加可选 faction 字段支持投票事件 - 新增 voting.py 模块处理投票逻辑 - 投票规则：每用户每 tick 限投一次，下一 tick 生效 - 投票累加到 factions.power，达到 threshold 触发阵营技能 - 添加 5 个投票系统测试用例，全部 26 个测试通过 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
484 lines
16 KiB
Python
484 lines
16 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
|
||
|
||
|
||
def test_stance_exists_in_agent(client):
|
||
"""测试 agent 包含 stance 字段"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
for agent_id, agent in data["world_state"]["agents"].items():
|
||
assert "stance" in agent
|
||
assert "optimism" in agent["stance"]
|
||
assert "fear" in agent["stance"]
|
||
assert 0.0 <= agent["stance"]["optimism"] <= 1.0
|
||
assert 0.0 <= agent["stance"]["fear"] <= 1.0
|
||
|
||
|
||
def test_stance_changes_on_world_event(client):
|
||
"""测试世界事件触发时 stance 变化"""
|
||
# 记录初始 stance
|
||
resp1 = client.post("/step", json={"events": []})
|
||
initial_stances = {
|
||
aid: agent["stance"].copy()
|
||
for aid, agent in resp1.json()["world_state"]["agents"].items()
|
||
}
|
||
|
||
# 触发世界事件
|
||
events = [
|
||
{"type": "chaos", "text": f"chaos{i}", "user": "test", "ts": i}
|
||
for i in range(20)
|
||
]
|
||
resp2 = client.post("/step", json={"events": events})
|
||
data = resp2.json()
|
||
|
||
# 验证事件触发
|
||
assert data["global_event"]["triggered"] is True
|
||
|
||
# stance 应该有变化（由于情绪和社交影响）
|
||
for aid, agent in data["world_state"]["agents"].items():
|
||
# stance 值仍在有效范围内
|
||
assert 0.0 <= agent["stance"]["optimism"] <= 1.0
|
||
assert 0.0 <= agent["stance"]["fear"] <= 1.0
|
||
|
||
|
||
def test_social_influence_deterministic(client):
|
||
"""测试社交影响是确定性的（同输入同输出）"""
|
||
events = [{"type": "comment", "text": "hello", "user": "test", "ts": 1}]
|
||
|
||
# 第一次执行
|
||
save_state(get_default_state())
|
||
resp1 = client.post("/step", json={"events": events})
|
||
stances1 = {
|
||
aid: agent["stance"]
|
||
for aid, agent in resp1.json()["world_state"]["agents"].items()
|
||
}
|
||
|
||
# 重置并再次执行
|
||
save_state(get_default_state())
|
||
resp2 = client.post("/step", json={"events": events})
|
||
stances2 = {
|
||
aid: agent["stance"]
|
||
for aid, agent in resp2.json()["world_state"]["agents"].items()
|
||
}
|
||
|
||
# 结果应该相同
|
||
for aid in stances1:
|
||
assert stances1[aid]["optimism"] == stances2[aid]["optimism"]
|
||
assert stances1[aid]["fear"] == stances2[aid]["fear"]
|
||
|
||
|
||
def test_factions_exists_in_world_state(client):
|
||
"""测试 world_state 包含 factions 字段"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
assert "factions" in data["world_state"]
|
||
factions = data["world_state"]["factions"]
|
||
assert "optimists" in factions
|
||
assert "fearful" in factions
|
||
# 验证新结构：每个派系包含 power 和 members
|
||
assert "power" in factions["optimists"]
|
||
assert "members" in factions["optimists"]
|
||
assert "power" in factions["fearful"]
|
||
assert "members" in factions["fearful"]
|
||
|
||
|
||
def test_factions_count_matches_agents(client):
|
||
"""测试派系成员总数等于非中立 agent 数量"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
factions = data["world_state"]["factions"]
|
||
# 新结构：统计 members 列表长度
|
||
total_in_factions = (
|
||
len(factions["optimists"]["members"]) +
|
||
len(factions["fearful"]["members"])
|
||
)
|
||
# 中立 agent 不在任何派系中
|
||
# 验证 power 等于 members 数量
|
||
assert factions["optimists"]["power"] == len(factions["optimists"]["members"])
|
||
assert factions["fearful"]["power"] == len(factions["fearful"]["members"])
|
||
|
||
|
||
def test_agent_has_faction_field(client):
|
||
"""测试 agent 包含 faction 字段"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
for agent_id, agent in data["world_state"]["agents"].items():
|
||
assert "faction" in agent
|
||
assert agent["faction"] in ["optimists", "fearful", "neutral"]
|
||
|
||
|
||
def test_triggered_faction_event_in_response(client):
|
||
"""测试响应包含 triggered_faction_event 字段"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
assert "triggered_faction_event" in data
|
||
assert "type" in data["triggered_faction_event"]
|
||
assert "source_faction" in data["triggered_faction_event"]
|
||
|
||
|
||
def test_story_arcs_exists_in_world_state(client):
|
||
"""测试 world_state 包含 story_arcs 字段"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
assert "story_arcs" in data["world_state"]
|
||
story_arcs = data["world_state"]["story_arcs"]
|
||
assert "civil_unrest" in story_arcs
|
||
assert "golden_age" in story_arcs
|
||
|
||
|
||
def test_story_arc_has_required_fields(client):
|
||
"""测试 story_arc 包含必要字段"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
for arc_id, arc in data["world_state"]["story_arcs"].items():
|
||
assert "progress" in arc
|
||
assert "threshold" in arc
|
||
assert "active" in arc
|
||
assert "stage" in arc
|
||
|
||
|
||
def test_story_event_in_response(client):
|
||
"""测试响应包含 story_event 字段"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
assert "story_event" in data
|
||
assert "triggered" in data["story_event"]
|
||
assert "arc_id" in data["story_event"]
|
||
|
||
|
||
# ==================== 投票系统测试 ====================
|
||
|
||
def test_votes_exists_in_world_state(client):
|
||
"""测试 world_state 包含 votes 字段"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
assert "votes" in data["world_state"]
|
||
votes = data["world_state"]["votes"]
|
||
assert "optimists" in votes
|
||
assert "fearful" in votes
|
||
assert "voted_users" in votes
|
||
|
||
|
||
def test_vote_records_correctly(client):
|
||
"""测试投票正确记录"""
|
||
events = [
|
||
{"type": "vote", "faction": "optimists", "text": "", "user": "user1", "ts": 1},
|
||
{"type": "vote", "faction": "fearful", "text": "", "user": "user2", "ts": 2},
|
||
]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
|
||
votes = data["world_state"]["votes"]
|
||
assert votes["optimists"] == 1
|
||
assert votes["fearful"] == 1
|
||
assert "user1" in votes["voted_users"]
|
||
assert "user2" in votes["voted_users"]
|
||
|
||
|
||
def test_vote_user_deduplication(client):
|
||
"""测试同一用户在同一 tick 只能投一次票"""
|
||
events = [
|
||
{"type": "vote", "faction": "optimists", "text": "", "user": "user1", "ts": 1},
|
||
{"type": "vote", "faction": "optimists", "text": "", "user": "user1", "ts": 2},
|
||
{"type": "vote", "faction": "fearful", "text": "", "user": "user1", "ts": 3},
|
||
]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
|
||
votes = data["world_state"]["votes"]
|
||
# 只计算第一次投票
|
||
assert votes["optimists"] == 1
|
||
assert votes["fearful"] == 0
|
||
|
||
|
||
def test_votes_apply_to_faction_power_next_tick(client):
|
||
"""测试投票在下一个 tick 累加到阵营能量"""
|
||
# 第一个 tick：投票
|
||
events = [
|
||
{"type": "vote", "faction": "optimists", "text": "", "user": "user1", "ts": 1},
|
||
{"type": "vote", "faction": "optimists", "text": "", "user": "user2", "ts": 2},
|
||
{"type": "vote", "faction": "fearful", "text": "", "user": "user3", "ts": 3},
|
||
]
|
||
resp1 = client.post("/step", json={"events": events})
|
||
data1 = resp1.json()
|
||
|
||
# 记录第一个 tick 后的阵营能量
|
||
power1_opt = data1["world_state"]["factions"]["optimists"]["power"]
|
||
power1_fear = data1["world_state"]["factions"]["fearful"]["power"]
|
||
|
||
# 第二个 tick：投票应该生效
|
||
resp2 = client.post("/step", json={"events": []})
|
||
data2 = resp2.json()
|
||
|
||
power2_opt = data2["world_state"]["factions"]["optimists"]["power"]
|
||
power2_fear = data2["world_state"]["factions"]["fearful"]["power"]
|
||
|
||
# 验证投票已累加（+2 optimists, +1 fearful）
|
||
assert power2_opt >= power1_opt + 2
|
||
assert power2_fear >= power1_fear + 1
|
||
|
||
# 验证投票已清空
|
||
assert data2["world_state"]["votes"]["optimists"] == 0
|
||
assert data2["world_state"]["votes"]["fearful"] == 0
|
||
|
||
|
||
def test_votes_trigger_faction_event(client):
|
||
"""测试投票累积触发阵营技能"""
|
||
# 多轮投票累积能量直到触发
|
||
for i in range(5):
|
||
events = [
|
||
{"type": "vote", "faction": "optimists", "text": "", "user": f"user_{i}_a", "ts": i},
|
||
{"type": "vote", "faction": "optimists", "text": "", "user": f"user_{i}_b", "ts": i + 0.1},
|
||
]
|
||
resp = client.post("/step", json={"events": events})
|
||
|
||
# 再投一轮让投票生效
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
# 验证阵营能量已累积（可能已触发并重置）
|
||
# 如果触发了，power 会重置为 0
|
||
factions = data["world_state"]["factions"]
|
||
triggered = data["triggered_faction_event"]["type"] is not None
|
||
|
||
# 要么触发了事件，要么能量在累积
|
||
assert triggered or factions["optimists"]["power"] > 0
|