5ae63d9df9
- 扩展 AgentState 添加 action_points, max_action_points, last_action_tick 字段 - 新增 ActionFeedback 模型用于返回行动执行结果 - 创建 action_points.py 模块实现行动点消耗与恢复逻辑 - 行动消耗表: vote=1, trigger_skill=2, influence=2, comment/support/chaos=0 - 每 tick 恢复 1 点行动点（不超过 max） - 行动点不足时拒绝执行并返回失败反馈 - 新增 7 个测试用例，全部 37 个测试通过 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
677 lines
22 KiB
Python
677 lines
22 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
|
||
|
||
|
||
# ==================== 技能树系统测试 ====================
|
||
|
||
def test_faction_skills_exists_in_world_state(client):
|
||
"""测试 world_state 包含 faction_skills 字段"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
assert "faction_skills" in data["world_state"]
|
||
skills = data["world_state"]["faction_skills"]
|
||
assert "optimists" in skills
|
||
assert "fearful" in skills
|
||
|
||
|
||
def test_faction_skills_has_required_fields(client):
|
||
"""测试技能包含必要字段"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
skills = data["world_state"]["faction_skills"]
|
||
opt_skills = skills["optimists"]["skills"]
|
||
|
||
assert "festival_boost" in opt_skills
|
||
skill = opt_skills["festival_boost"]
|
||
assert "unlocked" in skill
|
||
assert "cost" in skill
|
||
assert "effect" in skill
|
||
assert "requires" in skill
|
||
|
||
|
||
def test_skill_unlock_when_power_sufficient(client):
|
||
"""测试能量足够时解锁技能"""
|
||
# 多轮投票累积能量到 10+
|
||
for i in range(6):
|
||
events = [
|
||
{"type": "vote", "faction": "optimists", "text": "", "user": f"u{i}_a", "ts": i},
|
||
{"type": "vote", "faction": "optimists", "text": "", "user": f"u{i}_b", "ts": i+0.1},
|
||
]
|
||
client.post("/step", json={"events": events})
|
||
|
||
# 再执行一轮让投票生效
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
skills = data["world_state"]["faction_skills"]["optimists"]["skills"]
|
||
# festival_boost cost=10，应该已解锁
|
||
assert skills["festival_boost"]["unlocked"] is True
|
||
|
||
|
||
def test_skill_requires_prerequisite(client):
|
||
"""测试技能需要前置技能"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
skills = data["world_state"]["faction_skills"]["optimists"]["skills"]
|
||
# unity 需要 festival_boost 作为前置
|
||
assert skills["unity"]["requires"] == ["festival_boost"]
|
||
# 初始状态下 unity 未解锁
|
||
assert skills["unity"]["unlocked"] is False
|
||
|
||
|
||
# ==================== 行动点系统测试 ====================
|
||
|
||
def test_action_points_exists_in_agent(client):
|
||
"""测试 agent 包含行动点字段"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
for agent_id, agent in data["world_state"]["agents"].items():
|
||
assert "action_points" in agent
|
||
assert "max_action_points" in agent
|
||
assert "last_action_tick" in agent
|
||
assert agent["action_points"] >= 0
|
||
assert agent["max_action_points"] >= 1
|
||
|
||
|
||
def test_action_feedbacks_in_response(client):
|
||
"""测试响应包含 action_feedbacks 字段"""
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
assert "action_feedbacks" in data
|
||
assert isinstance(data["action_feedbacks"], list)
|
||
|
||
|
||
def test_vote_consumes_action_points(client):
|
||
"""测试投票消耗行动点"""
|
||
# alice 投票（消耗 1 点）
|
||
events = [
|
||
{"type": "vote", "faction": "optimists", "text": "", "user": "alice", "ts": 1}
|
||
]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
|
||
# 验证行动反馈
|
||
assert len(data["action_feedbacks"]) == 1
|
||
feedback = data["action_feedbacks"][0]
|
||
assert feedback["success"] is True
|
||
assert feedback["user"] == "alice"
|
||
# 初始 3 点，消耗 1 点，恢复 1 点 = 3 点
|
||
# 但消耗在恢复之后，所以是 3 - 1 = 2
|
||
assert feedback["remaining_ap"] == 2
|
||
|
||
|
||
def test_insufficient_action_points_rejected(client):
|
||
"""测试行动点不足时拒绝执行"""
|
||
# 重置状态，设置 alice 行动点为 0，last_action_tick 为当前 tick
|
||
# 这样在下一个 tick 恢复后只有 1 点
|
||
state = get_default_state()
|
||
state.tick = 10
|
||
state.agents["alice"].action_points = 0
|
||
state.agents["alice"].last_action_tick = 10 # 刚行动过，不会恢复
|
||
save_state(state)
|
||
|
||
# 尝试投票（需要 1 点），但恢复后有 1 点
|
||
# 为了测试不足，我们需要让 last_action_tick = tick，这样不会恢复
|
||
# 但 regenerate 检查的是 tick - last_action_tick >= 1
|
||
# tick 会先 +1 变成 11，11 - 10 = 1 >= 1，所以会恢复
|
||
|
||
# 重新设计：设置 last_action_tick 为未来的 tick
|
||
state = get_default_state()
|
||
state.tick = 10
|
||
state.agents["alice"].action_points = 0
|
||
state.agents["alice"].last_action_tick = 11 # 未来 tick，不会恢复
|
||
save_state(state)
|
||
|
||
events = [
|
||
{"type": "vote", "faction": "optimists", "text": "", "user": "alice", "ts": 10}
|
||
]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
|
||
# 验证被拒绝
|
||
assert len(data["action_feedbacks"]) == 1
|
||
feedback = data["action_feedbacks"][0]
|
||
assert feedback["success"] is False
|
||
assert "insufficient" in feedback["reason"]
|
||
|
||
|
||
def test_action_points_regenerate_per_tick(client):
|
||
"""测试行动点每 tick 恢复"""
|
||
# 设置 alice 行动点为 1
|
||
state = get_default_state()
|
||
state.agents["alice"].action_points = 1
|
||
state.agents["alice"].last_action_tick = 0
|
||
save_state(state)
|
||
|
||
# 执行一个 tick（无事件）
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
# 行动点应该恢复到 2（1 + 1）
|
||
assert data["world_state"]["agents"]["alice"]["action_points"] == 2
|
||
|
||
|
||
def test_action_points_not_exceed_max(client):
|
||
"""测试行动点不超过最大值"""
|
||
# 设置 alice 行动点为 3（已满）
|
||
state = get_default_state()
|
||
state.agents["alice"].action_points = 3
|
||
state.agents["alice"].max_action_points = 3
|
||
save_state(state)
|
||
|
||
# 执行多个 tick
|
||
for _ in range(3):
|
||
client.post("/step", json={"events": []})
|
||
|
||
resp = client.post("/step", json={"events": []})
|
||
data = resp.json()
|
||
|
||
# 行动点不应超过 max
|
||
assert data["world_state"]["agents"]["alice"]["action_points"] <= 3
|
||
|
||
|
||
def test_zero_cost_action_always_allowed(client):
|
||
"""测试 0 消耗的行动始终允许"""
|
||
# 设置 alice 行动点为 0
|
||
state = get_default_state()
|
||
state.agents["alice"].action_points = 0
|
||
save_state(state)
|
||
|
||
# comment 类型消耗 0 点
|
||
events = [
|
||
{"type": "comment", "text": "hello", "user": "alice", "ts": 1}
|
||
]
|
||
resp = client.post("/step", json={"events": events})
|
||
data = resp.json()
|
||
|
||
# 应该成功执行
|
||
assert len(data["action_feedbacks"]) == 1
|
||
feedback = data["action_feedbacks"][0]
|
||
assert feedback["success"] is True
|