snowy-yang
f8e093c98f
三件事凑成一次不兼容变更：WebSocket 通道、Hypercorn 服务器、「手动推进一天」都消失了。 实时通道：WebSocket -> SSE - hub 改为 SSE 订阅池，GET /api/stream 单向推送 welcome/state/day/notice/bye， 250ms 节流 + 15s 心跳注释；写操作一律走 HTTP POST /api/actions，不再有 WS 回执。 - 停机三处一起修才干净：close_all() 发 bye + timeout_graceful_shutdown 兜底 + 客户端 stop() 直接 shutdown 底层 socket（光 close() 会等读线程的锁，最长卡到下次心跳）。 - 网页版改 EventSource + fetch；tools/ws_demo.py 换成 tools/sse_demo.py。 服务器：Hypercorn -> uvicorn + zttp 的 HTTP/2 - 同一端口同时说 HTTP/1.1 与 HTTP/2。uvicorn 0.53 不再内置 h2，必须装 zttp 并显式 http="zttp"，否则 Config.load() 直接 raise。 - 信号接管改成子类覆盖 Server.handle_exit（0.53 没有 install_signal_handlers 可覆盖）， 不调 super()、不写 _captured_signals，退出时不重放信号；第二次信号强制退出。 - 入口启动前用 daemon.port_busy() 自检端口：Windows 上 SO_REUSEADDR 会让第二个实例 「绑上」已监听端口并抢走新连接，所以判断「能不能连上」而不是「能不能 bind」。 - 修掉 clear_pid_file 的 Windows 真 bug：文件还开着就 os.remove 会失败且被 except OSError 吞掉，导致每次正常退出都留 stale pid 文件。 节奏：一天 15 秒 -> 40 分钟，并整体移除「休息」 - DEFAULT_TICK_SECONDS=2400，MAX_TICK_SECONDS 放宽到 86400（旧上限会把 40 分钟夹回去）。 - engine.sleep / SLEEP_COOLDOWN / /api/sleep / TUI 的 [N]休息 / 网页版按钮全部删掉， 时间只能自然流动；机器人改为 wait_for_next_day(timeout)，CLI 加 bot --wait。 - 时长显示统一走 models.format_duration()，不再出现 2399.6s 这种写法。 键盘：修复方向键/空格/Tab/ESC 全部静默失效 - keys.py 的具名键统一小写并成为唯一来源，tui.py 一律引用 keys.XXX 而不手写字符串 （历史 bug 就是 Windows 侧产大写 UP、TUI 里写小写 up，字母键正常所以很难发现）。 - 顺带修 POSIX 路径一次 os.read 只取首字节、其余按键静默丢弃的老毛病。 测试：重建 500 项 - test_engine 110 / test_actions 35 / test_server 131(skip 35) / test_tui_keys 75 / test_bot 36 / test_client 113。未安装的第三方依赖用 skipUnless 自动跳过，其余照跑， 所以是全绿而不是靠跳过伪装。 - 被测试逼出两处真隐患并修掉：RemoteFarm._normalize() 必须「先补 scheme 再削尾斜杠」 （顺序反了畸形端点会溜到系统代理、报一个和真实原因无关的 502）；StateView 空视图要 预置 plots/animals，否则连不上服务器时读 state.plots 直接 AttributeError。 优化：先量后改，依据与前后数字都在 tools/bench.py - 新增 engine.brief_source()：brief/summary/动作回执只要十几个标量，不再为它们序列化 整局；to_dict 里 net_worth 只算一次。brief -76%、summary -77%、leaderboard(32) -78%、 GET /api/world -31%、POST /api/actions?full=1 -16%。 - 拆掉三个巨型函数：create_app 475->32 行（_Services + _register_*）、 build_parser 与 available_actions 分组、__main__.main 160->7 行。 - 消除重复：等级门槛 5 处 -> _is_locked/_require_level/_newly_unlocked；hub 的 broadcast/broadcast_all/close_all -> _push_many；daemon.probe/world -> _get_json； main.py 的 _add_endpoint_args/_leaderboard_lines。 - tools/bench.py 作为长期性能守卫（含调用计数器），谁把轻量路径改回 snapshot() 会立刻报警。 门禁：ruff format/check 全绿、pyright 0 错误、500 项测试 OK（skip 35）。
333 lines
13 KiB
Python
333 lines
13 KiB
Python
"""动作表测试：所有入口共用的那唯一一条执行通道。
|
||
|
||
动作表（``farm/server/actions.py``）是 HTTP、命令行、网页、机器人共同的分发点，
|
||
所以"参数校验只在这里做一次"这件事必须被钉住——校验漏一处，就等于每个入口都漏。
|
||
|
||
uv run python -m unittest tests.test_actions -v
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import unittest
|
||
|
||
from farm import config
|
||
from farm.engine import FarmEngine, GameError
|
||
from farm.server.actions import (
|
||
ACTIONS,
|
||
Param,
|
||
action_names,
|
||
describe_actions,
|
||
run_action,
|
||
)
|
||
|
||
EXPECTED_ACTIONS = {
|
||
"plant",
|
||
"water",
|
||
"harvest",
|
||
"harvest_all",
|
||
"buy_animal",
|
||
"feed",
|
||
"collect",
|
||
"buy_feed",
|
||
"buy_plot",
|
||
"pause",
|
||
}
|
||
|
||
|
||
def make_engine() -> FarmEngine:
|
||
engine = FarmEngine()
|
||
engine.state.gold = 5000
|
||
return engine
|
||
|
||
|
||
class ParamTestCase(unittest.TestCase):
|
||
"""参数规格的取值转换——外部 JSON 什么类型都可能传进来。"""
|
||
|
||
def test_int_accepts_the_obvious_spellings(self):
|
||
spec = Param("plot", "int")
|
||
self.assertEqual(spec.coerce(3), 3)
|
||
self.assertEqual(spec.coerce("3"), 3)
|
||
self.assertEqual(spec.coerce(" 12 "), 12)
|
||
self.assertEqual(spec.coerce("-4"), -4)
|
||
self.assertEqual(spec.coerce("+7"), 7)
|
||
self.assertEqual(spec.coerce(3.0), 3)
|
||
|
||
def test_int_rejects_booleans_floats_and_words(self):
|
||
spec = Param("plot", "int")
|
||
for bad in (True, 3.5, "3.5", "三", "", None, [], {}):
|
||
with self.subTest(raw=bad):
|
||
with self.assertRaises(GameError) as ctx:
|
||
spec.coerce(bad)
|
||
self.assertEqual(ctx.exception.code, "bad_param")
|
||
|
||
def test_str_accepts_scalars(self):
|
||
spec = Param("crop", "str")
|
||
self.assertEqual(spec.coerce("wheat"), "wheat")
|
||
self.assertEqual(spec.coerce(123), "123")
|
||
self.assertEqual(spec.coerce(1.5), "1.5")
|
||
|
||
def test_str_rejects_containers(self):
|
||
spec = Param("crop", "str")
|
||
for bad in (["wheat"], {"a": 1}, None):
|
||
with self.subTest(raw=bad), self.assertRaises(GameError):
|
||
spec.coerce(bad)
|
||
|
||
def test_bool_accepts_common_spellings(self):
|
||
spec = Param("paused", "bool")
|
||
self.assertIs(spec.coerce(True), True)
|
||
self.assertIs(spec.coerce(False), False)
|
||
self.assertIs(spec.coerce("true"), True)
|
||
self.assertIs(spec.coerce("FALSE"), False)
|
||
self.assertIs(spec.coerce("1"), True)
|
||
self.assertIs(spec.coerce("0"), False)
|
||
self.assertIs(spec.coerce(1), True)
|
||
self.assertIs(spec.coerce(0), False)
|
||
|
||
def test_bool_rejects_ambiguous_words(self):
|
||
spec = Param("paused", "bool")
|
||
for bad in ("yes", "no", "暂停", "", [], 1.5):
|
||
with self.subTest(raw=bad):
|
||
with self.assertRaises(GameError) as ctx:
|
||
spec.coerce(bad)
|
||
self.assertEqual(ctx.exception.code, "bad_param")
|
||
|
||
def test_describe_matches_the_spec(self):
|
||
spec = Param("quantity", "int", required=False, default=10, desc="份数")
|
||
self.assertEqual(
|
||
spec.to_dict(),
|
||
{"name": "quantity", "type": "int", "required": False, "default": 10, "desc": "份数"},
|
||
)
|
||
|
||
|
||
class ActionTableTestCase(unittest.TestCase):
|
||
def test_the_table_holds_exactly_the_expected_actions(self):
|
||
self.assertEqual(set(action_names()), EXPECTED_ACTIONS)
|
||
|
||
def test_action_names_are_sorted(self):
|
||
names = action_names()
|
||
self.assertEqual(names, sorted(names))
|
||
|
||
def test_save_and_reset_are_deliberately_absent(self):
|
||
"""存档类动作必须由牧场会话处理，否则会写默认路径、覆盖别人的档。"""
|
||
self.assertNotIn("save", ACTIONS)
|
||
self.assertNotIn("reset", ACTIONS)
|
||
for name, spec in ACTIONS.items():
|
||
with self.subTest(action=name):
|
||
self.assertNotIn(name, ("save", "reset"))
|
||
self.assertNotEqual(spec.summary, "")
|
||
|
||
def test_removed_sleep_action_stays_removed(self):
|
||
"""「休息 / 手动推进一天」已经整体移除，别悄悄回来。"""
|
||
self.assertNotIn("sleep", ACTIONS)
|
||
self.assertNotIn("rest", ACTIONS)
|
||
self.assertNotIn("next_day", ACTIONS)
|
||
|
||
def test_describe_actions_is_json_shaped(self):
|
||
described = describe_actions()
|
||
self.assertEqual(len(described), len(ACTIONS))
|
||
for item in described:
|
||
with self.subTest(action=item["action"]):
|
||
self.assertEqual(set(item), {"action", "summary", "params"})
|
||
for param in item["params"]:
|
||
self.assertEqual(set(param), {"name", "type", "required", "default", "desc"})
|
||
|
||
def test_every_action_handler_returns_a_dict(self):
|
||
"""返回类型约定是 dict：HTTP 层直接把它铺进 JSON 响应。"""
|
||
engine = make_engine()
|
||
engine.state.level = 5
|
||
results = [
|
||
run_action(engine, "pause", {"paused": True}),
|
||
run_action(engine, "buy_feed", {"quantity": 1}),
|
||
run_action(engine, "plant", {"plot": 1, "crop": "wheat"}),
|
||
]
|
||
for result in results:
|
||
with self.subTest(result=result):
|
||
self.assertIsInstance(result, dict)
|
||
|
||
|
||
class RunActionValidationTestCase(unittest.TestCase):
|
||
"""统一执行通道的输入校验。"""
|
||
|
||
def setUp(self):
|
||
self.engine = make_engine()
|
||
|
||
def assert_code(self, code: str, name, params=None) -> GameError:
|
||
with self.assertRaises(GameError) as ctx:
|
||
run_action(self.engine, name, params)
|
||
self.assertEqual(ctx.exception.code, code)
|
||
return ctx.exception
|
||
|
||
def test_missing_or_blank_action_name(self):
|
||
for bad in (None, "", " ", 123, ["plant"]):
|
||
with self.subTest(name=bad):
|
||
self.assert_code("bad_param", bad)
|
||
|
||
def test_unknown_action_lists_the_valid_ones(self):
|
||
exc = self.assert_code("unknown_action", "fly_to_moon")
|
||
for name in EXPECTED_ACTIONS:
|
||
self.assertIn(name, exc.message)
|
||
|
||
def test_action_name_is_normalized(self):
|
||
result = run_action(self.engine, " PLANT ", {"plot": 1, "crop": "wheat"})
|
||
self.assertEqual(result["action"], "plant")
|
||
self.assertTrue(result["ok"])
|
||
|
||
def test_params_must_be_an_object(self):
|
||
for bad in ([1, 2], "plot=1", 3):
|
||
with self.subTest(params=bad):
|
||
self.assert_code("bad_param", "plant", bad)
|
||
|
||
def test_none_params_becomes_an_empty_mapping(self):
|
||
self.assert_code("missing_param", "plant", None)
|
||
|
||
def test_missing_required_parameter(self):
|
||
exc = self.assert_code("missing_param", "plant", {"plot": 1})
|
||
self.assertIn("crop", exc.message)
|
||
|
||
def test_unknown_parameter_is_rejected_not_ignored(self):
|
||
"""静默忽略拼错的参数，会让 AI 以为动作生效了。"""
|
||
exc = self.assert_code("bad_param", "plant", {"plot": 1, "crop": "wheat", "pot": 2})
|
||
self.assertIn("pot", exc.message)
|
||
|
||
def test_wrong_parameter_type(self):
|
||
self.assert_code("bad_param", "plant", {"plot": "第一块", "crop": "wheat"})
|
||
|
||
def test_null_parameter_counts_as_missing(self):
|
||
self.assert_code("missing_param", "plant", {"plot": None, "crop": "wheat"})
|
||
|
||
def test_optional_parameter_falls_back_to_its_default(self):
|
||
before = self.engine.state.feed
|
||
result = run_action(self.engine, "buy_feed", {})
|
||
self.assertEqual(self.engine.state.feed, before + config.FEED_BULK)
|
||
self.assertEqual(result["action"], "buy_feed")
|
||
|
||
def test_error_dict_is_machine_readable(self):
|
||
exc = self.assert_code("unknown_action", "nope")
|
||
payload = exc.to_dict()
|
||
self.assertEqual(payload, {"ok": False, "error": "unknown_action", "message": exc.message})
|
||
|
||
def test_result_always_carries_ok_and_action(self):
|
||
run_action(self.engine, "plant", {"plot": 1, "crop": "wheat"})
|
||
self.engine.state.plots[0].growth = 2
|
||
result = run_action(self.engine, "harvest_all", {})
|
||
self.assertIs(result["ok"], True)
|
||
self.assertEqual(result["action"], "harvest_all")
|
||
|
||
|
||
class ActionBehaviourTestCase(unittest.TestCase):
|
||
"""经过动作表之后的玩法行为——确认分发没有接错线。"""
|
||
|
||
def setUp(self):
|
||
self.engine = make_engine()
|
||
self.engine.state.level = 5
|
||
|
||
def test_plant_then_water_then_harvest(self):
|
||
run_action(self.engine, "plant", {"plot": 2, "crop": "carrot"})
|
||
plot = self.engine.state.plots[1]
|
||
self.assertEqual(plot.crop, "carrot")
|
||
self.assertEqual(plot.index, 1)
|
||
|
||
plot.moisture = 0
|
||
run_action(self.engine, "water", {"plot": 2})
|
||
self.assertEqual(plot.moisture, config.WATER_CAPACITY)
|
||
|
||
plot.growth = plot.grow_days
|
||
result = run_action(self.engine, "harvest", {"plot": 2})
|
||
self.assertEqual(result["action"], "harvest")
|
||
self.assertTrue(plot.is_empty)
|
||
|
||
def test_buy_and_feed_and_collect_an_animal(self):
|
||
bought = run_action(self.engine, "buy_animal", {"kind": "chicken"})
|
||
aid = bought["animal"]["id"]
|
||
|
||
fed = run_action(self.engine, "feed", {"animal_id": aid})
|
||
self.assertEqual(fed["animal"]["fed"], True)
|
||
|
||
self.engine.state.animals[0].pending = 2
|
||
collected = run_action(self.engine, "collect", {"animal_id": aid})
|
||
self.assertEqual(collected["quantity"], 2)
|
||
|
||
def test_feed_accepts_an_explicit_quantity(self):
|
||
bought = run_action(self.engine, "buy_animal", {"kind": "cow"})
|
||
aid = bought["animal"]["id"]
|
||
self.engine.state.feed = 20
|
||
run_action(self.engine, "feed", {"animal_id": aid, "quantity": "2"})
|
||
self.assertEqual(self.engine.state.feed, 18, "字符串数字要能转换")
|
||
|
||
def test_buy_plot_returns_the_next_price(self):
|
||
result = run_action(self.engine, "buy_plot", {})
|
||
self.assertEqual(result["next_plot_price"], self.engine.state.next_plot_price)
|
||
|
||
def test_pause_through_the_table(self):
|
||
self.assertEqual(run_action(self.engine, "pause", {"paused": "true"})["paused"], True)
|
||
self.assertEqual(run_action(self.engine, "pause", {"paused": False})["paused"], False)
|
||
self.assertEqual(
|
||
run_action(self.engine, "pause", {})["paused"],
|
||
True,
|
||
"省略参数表示反转，此时是 False -> True",
|
||
)
|
||
|
||
def test_pause_rejects_values_that_are_not_boolean(self):
|
||
"""参数校验在前：含糊的字符串会被挡在门口，而不是当成 true。"""
|
||
with self.assertRaises(GameError) as ctx:
|
||
run_action(self.engine, "pause", {"paused": "不确定算不算"})
|
||
self.assertEqual(ctx.exception.code, "bad_param")
|
||
|
||
|
||
class HarvestAllTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.engine = make_engine()
|
||
self.engine.state.level = 5
|
||
|
||
def test_nothing_ripe_is_an_error(self):
|
||
with self.assertRaises(GameError) as ctx:
|
||
run_action(self.engine, "harvest_all", {})
|
||
self.assertEqual(ctx.exception.code, "nothing_to_harvest")
|
||
|
||
def test_harvests_every_ripe_plot_and_skips_the_rest(self):
|
||
for number in (1, 2, 3):
|
||
run_action(self.engine, "plant", {"plot": number, "crop": "wheat"})
|
||
for index in (0, 1):
|
||
self.engine.state.plots[index].growth = 2 # 1、2 号成熟
|
||
result = run_action(self.engine, "harvest_all", {})
|
||
|
||
self.assertEqual(result["plots"], [1, 2])
|
||
self.assertEqual(
|
||
result["quantity"], result["earned"] // config.CROPS["wheat"]["base_price"]
|
||
)
|
||
self.assertEqual(result["errors"], [])
|
||
self.assertFalse(self.engine.state.plots[2].is_empty, "没熟的 3 号地要保持原样")
|
||
|
||
def test_partial_failure_does_not_stop_the_others(self):
|
||
"""一块地失败不能带走其它地块的收成。"""
|
||
for number in (1, 2):
|
||
run_action(self.engine, "plant", {"plot": number, "crop": "wheat"})
|
||
for index in (0, 1):
|
||
self.engine.state.plots[index].growth = 2
|
||
self.engine.state.ap = 10
|
||
|
||
# 把第一块地做成"看起来熟了、实际收不了"（作物键被改坏）
|
||
self.engine.state.plots[0].crop = "wheat"
|
||
original = self.engine.harvest
|
||
|
||
def flaky(number):
|
||
if int(number) == 1:
|
||
raise GameError("plot_empty", "1 号地是空地")
|
||
return original(number)
|
||
|
||
self.engine.harvest = flaky # type: ignore[method-assign]
|
||
result = run_action(self.engine, "harvest_all", {})
|
||
self.assertEqual(result["plots"], [2])
|
||
self.assertEqual(len(result["errors"]), 1)
|
||
self.assertIn("1号地", result["errors"][0])
|
||
|
||
def test_message_summarizes_the_haul(self):
|
||
run_action(self.engine, "plant", {"plot": 1, "crop": "wheat"})
|
||
self.engine.state.plots[0].growth = 2
|
||
result = run_action(self.engine, "harvest_all", {})
|
||
self.assertIn("1 块地", result["message"])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|