ai-town/engine-python/tests/test_step.py

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