feat: 实时通道改 SSE、服务器换 uvicorn+zttp、节奏改 40 分钟一天，并重建测试与优化热路径
三件事凑成一次不兼容变更：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）。
This commit is contained in:
snowy-yang committed
1 parent
47dd818360
commit
f8e093c98f
36 files changed
+6865
-2204
No files matched your search
@@ -0,0 +1,332 @@
|
||||
"""动作表测试：所有入口共用的那唯一一条执行通道。
|
||||
|
||||
动作表（``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)
|
||||
@@ -0,0 +1,394 @@
|
||||
"""规则机器人测试：打分策略、单步执行、以及"没事可做时怎么办"。
|
||||
|
||||
机器人是「服务器把状态和合法动作摊开后，外面是谁在决策都不重要」这个设计的
|
||||
活证明，所以它走的每一步都应当是**普通客户端**能走的。
|
||||
|
||||
全程使用鸭子类型的假客户端，不碰网络，也不依赖服务器。
|
||||
|
||||
uv run python -m unittest tests.test_bot -v
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import io
|
||||
import unittest
|
||||
|
||||
from farm.bot import RuleBot, random_bot
|
||||
from farm.engine import GameError
|
||||
from farm.models import format_duration
|
||||
|
||||
|
||||
class FakeClient:
|
||||
"""只实现机器人真正会用的那几个方法。"""
|
||||
|
||||
def __init__(self, actions=None, state=None, catalog=None) -> None:
|
||||
self.actions = list(actions or [])
|
||||
self.state = dict(state or {})
|
||||
self.catalog_data = catalog or {"crops": [], "animals": []}
|
||||
self.calls: list[tuple[str, dict]] = []
|
||||
self.fail_with: Exception | None = None
|
||||
self.label = "假通道"
|
||||
|
||||
def available_actions(self):
|
||||
return [dict(item) for item in self.actions]
|
||||
|
||||
def snapshot(self):
|
||||
return dict(self.state)
|
||||
|
||||
def catalog(self):
|
||||
return self.catalog_data
|
||||
|
||||
def transport_label(self) -> str:
|
||||
return self.label
|
||||
|
||||
def _action(self, name, params=None):
|
||||
self.calls.append((name, dict(params or {})))
|
||||
if self.fail_with is not None:
|
||||
raise self.fail_with
|
||||
return {"ok": True, "message": f"完成 {name}"}
|
||||
|
||||
|
||||
CATALOG = {
|
||||
"crops": [
|
||||
{"key": "wheat", "name": "小麦", "profit_per_day": 4.5},
|
||||
{"key": "carrot", "name": "胡萝卜", "profit_per_day": 6.0},
|
||||
],
|
||||
"animals": [
|
||||
{"key": "chicken", "name": "鸡", "cost": 90, "profit_per_day": 9.0},
|
||||
{"key": "cow", "name": "牛", "cost": 480, "profit_per_day": 60.0},
|
||||
],
|
||||
}
|
||||
|
||||
BASE_STATE = {
|
||||
"day": 1,
|
||||
"gold": 1000,
|
||||
"feed": 20,
|
||||
"ap": 10,
|
||||
"next_plot_price": 200,
|
||||
"paused": False,
|
||||
"seconds_to_tick": 120.0,
|
||||
}
|
||||
|
||||
|
||||
def make_bot(client=None, **kwargs) -> RuleBot:
|
||||
options = {"verbose": False, "pause": 0.0, "name": "tester"}
|
||||
options.update(kwargs)
|
||||
return RuleBot(client or FakeClient(state=BASE_STATE), **options)
|
||||
|
||||
|
||||
class ScoreTestCase(unittest.TestCase):
|
||||
"""打分表就是整个"策略"——顺序错了机器人就会先乱花钱。"""
|
||||
|
||||
def setUp(self):
|
||||
self.bot = make_bot()
|
||||
|
||||
def score(self, action: dict, **state_overrides) -> float:
|
||||
state = dict(BASE_STATE)
|
||||
state.update(state_overrides)
|
||||
return self.bot.score(action, state, CATALOG)
|
||||
|
||||
def test_harvest_comes_first(self):
|
||||
self.assertEqual(self.score({"action": "harvest"}), 100.0)
|
||||
|
||||
def test_free_money_actions_rank_above_others(self):
|
||||
order = [
|
||||
self.score({"action": "harvest"}),
|
||||
self.score({"action": "collect"}),
|
||||
self.score({"action": "feed"}),
|
||||
self.score({"action": "water"}),
|
||||
]
|
||||
self.assertEqual(order, sorted(order, reverse=True))
|
||||
self.assertEqual(len(set(order)), len(order), "分数要有区分度")
|
||||
|
||||
def test_plant_prefers_the_best_profit_per_day(self):
|
||||
wheat = self.score({"action": "plant", "params": {"crop": "wheat"}})
|
||||
carrot = self.score({"action": "plant", "params": {"crop": "carrot"}})
|
||||
self.assertGreater(carrot, wheat)
|
||||
|
||||
def test_plant_of_an_unknown_crop_still_gets_a_base_score(self):
|
||||
value = self.score({"action": "plant", "params": {"crop": "榴莲"}})
|
||||
self.assertEqual(value, 60.0)
|
||||
|
||||
def test_feed_is_bought_only_when_short_and_affordable(self):
|
||||
action = {"action": "buy_feed"}
|
||||
self.assertEqual(self.score(action, feed=1, gold=500), 40.0)
|
||||
self.assertEqual(self.score(action, feed=0, gold=500), 40.0)
|
||||
self.assertEqual(self.score(action, feed=5, gold=500), 5.0, "库存够就别花钱")
|
||||
self.assertEqual(self.score(action, feed=1, gold=50), 5.0, "钱少也别花")
|
||||
|
||||
def test_land_is_bought_only_with_a_cash_buffer(self):
|
||||
action = {"action": "buy_plot"}
|
||||
self.assertEqual(self.score(action, gold=600, next_plot_price=200), 34.0)
|
||||
self.assertEqual(self.score(action, gold=599, next_plot_price=200), 4.0)
|
||||
self.assertEqual(self.score(action, gold=9999, next_plot_price=-1), 4.0, "没地可买")
|
||||
|
||||
def test_animals_need_a_cash_buffer_and_a_free_action_point(self):
|
||||
action = {"action": "buy_animal", "params": {"kind": "chicken"}}
|
||||
self.assertEqual(self.score(action, gold=270, ap=1), 39.0)
|
||||
self.assertEqual(self.score(action, gold=269, ap=1), 3.0)
|
||||
self.assertEqual(self.score(action, gold=1000, ap=0), 3.0, "没行动点就别想")
|
||||
|
||||
def test_unknown_actions_have_the_lowest_score(self):
|
||||
self.assertEqual(self.score({"action": "fly_to_moon"}), 1.0)
|
||||
self.assertEqual(self.score({}), 1.0)
|
||||
|
||||
def test_score_survives_a_missing_payload(self):
|
||||
"""服务器偶尔给的是精简状态，机器人不能因此崩掉。"""
|
||||
self.assertEqual(self.bot.score({"action": "harvest"}, {}, {}), 100.0)
|
||||
|
||||
|
||||
class ChooseTestCase(unittest.TestCase):
|
||||
def test_picks_the_highest_scoring_action(self):
|
||||
client = FakeClient(
|
||||
actions=[
|
||||
{"action": "water", "params": {"plot": 1}},
|
||||
{"action": "harvest", "params": {"plot": 2}},
|
||||
{"action": "plant", "params": {"plot": 3, "crop": "carrot"}},
|
||||
],
|
||||
state=BASE_STATE,
|
||||
catalog=CATALOG,
|
||||
)
|
||||
choice = make_bot(client).choose()
|
||||
assert choice is not None
|
||||
self.assertEqual(choice["action"], "harvest")
|
||||
|
||||
def test_no_actions_means_no_choice(self):
|
||||
self.assertIsNone(make_bot(FakeClient(actions=[], state=BASE_STATE)).choose())
|
||||
|
||||
def test_nothing_worth_doing_means_no_choice(self):
|
||||
"""分数都 <= 1 时应当"什么都不做"，而不是硬挑一个。"""
|
||||
client = FakeClient(actions=[{"action": "fly"}], state=BASE_STATE, catalog=CATALOG)
|
||||
self.assertIsNone(make_bot(client).choose())
|
||||
|
||||
def test_choice_is_json_shaped(self):
|
||||
client = FakeClient(
|
||||
actions=[{"action": "harvest", "params": {"plot": 1}}],
|
||||
state=BASE_STATE,
|
||||
catalog=CATALOG,
|
||||
)
|
||||
choice = make_bot(client).choose()
|
||||
assert choice is not None
|
||||
self.assertEqual(choice["params"], {"plot": 1})
|
||||
|
||||
|
||||
class StepTestCase(unittest.TestCase):
|
||||
def test_successful_step_is_recorded(self):
|
||||
client = FakeClient(
|
||||
actions=[{"action": "harvest", "params": {"plot": 1}}],
|
||||
state=BASE_STATE,
|
||||
catalog=CATALOG,
|
||||
)
|
||||
bot = make_bot(client)
|
||||
self.assertEqual(bot.step(), "harvest")
|
||||
self.assertEqual(bot.actions_executed, 1)
|
||||
self.assertEqual(bot.done, ["harvest"])
|
||||
self.assertEqual(client.calls, [("harvest", {"plot": 1})])
|
||||
|
||||
def test_nothing_to_do_returns_none(self):
|
||||
bot = make_bot(FakeClient(actions=[], state=BASE_STATE))
|
||||
self.assertIsNone(bot.step())
|
||||
self.assertEqual(bot.actions_executed, 0)
|
||||
|
||||
def test_failure_is_counted_not_raised(self):
|
||||
client = FakeClient(
|
||||
actions=[{"action": "plant", "params": {"plot": 1, "crop": "wheat"}}],
|
||||
state=BASE_STATE,
|
||||
catalog=CATALOG,
|
||||
)
|
||||
client.fail_with = GameError("no_ap", "行动点不足")
|
||||
bot = make_bot(client)
|
||||
self.assertIsNone(bot.step())
|
||||
self.assertEqual(bot.failures, 1)
|
||||
self.assertEqual(bot.actions_executed, 0)
|
||||
|
||||
def test_a_failed_step_does_not_stop_the_bot(self):
|
||||
client = FakeClient(
|
||||
actions=[{"action": "plant", "params": {"plot": 1, "crop": "wheat"}}],
|
||||
state=BASE_STATE,
|
||||
catalog=CATALOG,
|
||||
)
|
||||
client.fail_with = GameError("no_ap", "行动点不足")
|
||||
bot = make_bot(client)
|
||||
bot.step()
|
||||
client.fail_with = None
|
||||
self.assertEqual(bot.step(), "plant")
|
||||
|
||||
|
||||
class WaitForNextDayTestCase(unittest.TestCase):
|
||||
"""没有「手动推进一天」之后，等待策略就是机器人的全部节奏感。"""
|
||||
|
||||
def test_paused_time_is_hopeless(self):
|
||||
client = FakeClient(state={**BASE_STATE, "paused": True})
|
||||
bot = make_bot(client)
|
||||
self.assertFalse(bot.wait_for_next_day(10.0))
|
||||
|
||||
def test_zero_timeout_means_clock_out_immediately(self):
|
||||
client = FakeClient(state=BASE_STATE)
|
||||
bot = make_bot(client)
|
||||
self.assertFalse(bot.wait_for_next_day(0))
|
||||
self.assertEqual(client.calls, [], "收工不等于发请求")
|
||||
|
||||
def test_negative_timeout_also_clocks_out(self):
|
||||
self.assertFalse(make_bot().wait_for_next_day(-5))
|
||||
|
||||
def test_waits_until_the_day_actually_changes(self):
|
||||
class TickingClient(FakeClient):
|
||||
"""第三次被问到时，"时间"已经跨天了。"""
|
||||
|
||||
def __init__(self, **kwargs) -> None:
|
||||
super().__init__(**kwargs)
|
||||
self.reads = 0
|
||||
|
||||
def snapshot(self):
|
||||
self.reads += 1
|
||||
data = super().snapshot()
|
||||
if self.reads >= 3:
|
||||
data["day"] = data.get("day", 1) + 1
|
||||
return data
|
||||
|
||||
bot = make_bot(TickingClient(state=BASE_STATE))
|
||||
self.assertTrue(bot.wait_for_next_day(3.0))
|
||||
|
||||
def test_gives_up_when_the_day_never_comes(self):
|
||||
bot = make_bot(FakeClient(state=BASE_STATE))
|
||||
self.assertFalse(bot.wait_for_next_day(0.4))
|
||||
|
||||
def test_the_message_uses_the_human_duration_format(self):
|
||||
client = FakeClient(state={**BASE_STATE, "seconds_to_tick": 1234.5})
|
||||
bot = make_bot(client)
|
||||
buffer = io.StringIO()
|
||||
with contextlib.redirect_stdout(buffer):
|
||||
bot.verbose = True
|
||||
bot.wait_for_next_day(0)
|
||||
self.assertIn(format_duration(1234.5), buffer.getvalue())
|
||||
|
||||
def test_stop_interrupts_the_wait(self):
|
||||
bot = make_bot(FakeClient(state=BASE_STATE))
|
||||
bot.stop()
|
||||
self.assertTrue(bot.stopped)
|
||||
self.assertFalse(bot.wait_for_next_day(5.0), "已经叫停了就不该干等")
|
||||
|
||||
|
||||
class RunTestCase(unittest.TestCase):
|
||||
def test_round_limit_is_respected(self):
|
||||
client = FakeClient(
|
||||
actions=[{"action": "harvest", "params": {"plot": 1}}],
|
||||
state=BASE_STATE,
|
||||
catalog=CATALOG,
|
||||
)
|
||||
summary = make_bot(client).run(rounds=3)
|
||||
self.assertEqual(summary["actions"], 3)
|
||||
self.assertEqual(summary["name"], "tester")
|
||||
|
||||
def test_summary_counts_the_days_that_passed(self):
|
||||
client = FakeClient(
|
||||
actions=[{"action": "harvest", "params": {"plot": 1}}],
|
||||
state=BASE_STATE,
|
||||
catalog=CATALOG,
|
||||
)
|
||||
summary = make_bot(client).run(rounds=1)
|
||||
self.assertEqual(set(summary) >= {"name", "days", "actions", "failures", "net_worth"}, True)
|
||||
|
||||
def test_no_action_points_means_clocking_out(self):
|
||||
client = FakeClient(
|
||||
actions=[{"action": "harvest", "params": {"plot": 1}}],
|
||||
state={**BASE_STATE, "ap": 0},
|
||||
catalog=CATALOG,
|
||||
)
|
||||
bot = make_bot(client)
|
||||
summary = bot.run(rounds=5)
|
||||
self.assertEqual(summary["actions"], 0)
|
||||
self.assertEqual(client.calls, [], "没行动点就不该发动作请求")
|
||||
|
||||
def test_idle_limit_triggers_the_waiting_path(self):
|
||||
client = FakeClient(actions=[], state=BASE_STATE)
|
||||
bot = make_bot(client)
|
||||
summary = bot.run(rounds=5, idle_limit=1, wait_seconds=0)
|
||||
self.assertEqual(summary["actions"], 0)
|
||||
self.assertEqual(bot.failures, 0)
|
||||
|
||||
def test_stopped_bot_exits_the_loop(self):
|
||||
client = FakeClient(
|
||||
actions=[{"action": "harvest", "params": {"plot": 1}}],
|
||||
state=BASE_STATE,
|
||||
catalog=CATALOG,
|
||||
)
|
||||
bot = make_bot(client)
|
||||
bot.stop()
|
||||
summary = bot.run(rounds=10)
|
||||
self.assertEqual(summary["actions"], 0)
|
||||
|
||||
def test_verbose_mode_prints_a_greeting(self):
|
||||
client = FakeClient(
|
||||
actions=[{"action": "harvest", "params": {"plot": 1}}],
|
||||
state=BASE_STATE,
|
||||
catalog=CATALOG,
|
||||
)
|
||||
buffer = io.StringIO()
|
||||
with contextlib.redirect_stdout(buffer):
|
||||
make_bot(client, verbose=True).run(rounds=1)
|
||||
self.assertIn("上线", buffer.getvalue())
|
||||
|
||||
def test_quiet_mode_prints_nothing(self):
|
||||
client = FakeClient(
|
||||
actions=[{"action": "harvest", "params": {"plot": 1}}],
|
||||
state=BASE_STATE,
|
||||
catalog=CATALOG,
|
||||
)
|
||||
buffer = io.StringIO()
|
||||
with contextlib.redirect_stdout(buffer):
|
||||
make_bot(client).run(rounds=1)
|
||||
self.assertEqual(buffer.getvalue(), "")
|
||||
|
||||
def test_transport_label_is_mentioned_on_startup(self):
|
||||
client = FakeClient(
|
||||
actions=[{"action": "harvest", "params": {"plot": 1}}],
|
||||
state=BASE_STATE,
|
||||
catalog=CATALOG,
|
||||
)
|
||||
client.label = "SSE 实时"
|
||||
buffer = io.StringIO()
|
||||
with contextlib.redirect_stdout(buffer):
|
||||
make_bot(client, verbose=True).run(rounds=1)
|
||||
self.assertIn("SSE 实时", buffer.getvalue())
|
||||
|
||||
def test_max_minutes_zero_means_no_deadline(self):
|
||||
client = FakeClient(
|
||||
actions=[{"action": "harvest", "params": {"plot": 1}}],
|
||||
state=BASE_STATE,
|
||||
catalog=CATALOG,
|
||||
)
|
||||
summary = make_bot(client).run(rounds=2, max_minutes=0)
|
||||
self.assertEqual(summary["actions"], 2)
|
||||
|
||||
|
||||
class RandomBotTestCase(unittest.TestCase):
|
||||
def test_random_bot_picks_from_the_legal_list(self):
|
||||
actions = [
|
||||
{"action": "water", "params": {"plot": 1}},
|
||||
{"action": "harvest", "params": {"plot": 2}},
|
||||
]
|
||||
client = FakeClient(actions=actions, state=BASE_STATE, catalog=CATALOG)
|
||||
bot = random_bot(client, verbose=False, pause=0.0)
|
||||
choice = bot.choose()
|
||||
assert choice is not None
|
||||
self.assertIn(choice["action"], {"water", "harvest"})
|
||||
self.assertEqual(bot.name, "random")
|
||||
|
||||
def test_random_bot_returns_none_without_options(self):
|
||||
bot = random_bot(FakeClient(actions=[], state=BASE_STATE), verbose=False, pause=0.0)
|
||||
self.assertIsNone(bot.choose())
|
||||
|
||||
def test_random_bot_still_executes_steps(self):
|
||||
client = FakeClient(
|
||||
actions=[{"action": "harvest", "params": {"plot": 1}}],
|
||||
state=BASE_STATE,
|
||||
catalog=CATALOG,
|
||||
)
|
||||
bot = random_bot(client, verbose=False, pause=0.0)
|
||||
self.assertEqual(bot.step(), "harvest")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
@@ -0,0 +1,990 @@
|
||||
"""远程客户端测试：视图语义、地址归一化、SSE 消息分发、以及 HTTP 传输层。
|
||||
|
||||
这里是「TUI 不需要知道自己在跟谁说话」这句承诺的兑现处：
|
||||
``StateView`` 必须把服务器发来的两种东西区分开——**完整快照**（带 plots/animals）
|
||||
要整体替换，**精简状态**（动作回执里那几个标量）只能做字段合并。搞反了的后果
|
||||
很具体：播一次种，整片农田就从界面上消失了。
|
||||
|
||||
除 ``HttpTransportTestCase`` 会起一个纯标准库的回环桩服务外，其余全部离线；
|
||||
桩服务只绑 127.0.0.1，不依赖 fastapi / httpx / uvicorn。
|
||||
|
||||
uv run python -m unittest tests.test_client -v
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import http.server
|
||||
import socket
|
||||
import threading
|
||||
import unittest
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from typing import ClassVar
|
||||
|
||||
from farm import config, net
|
||||
from farm.client import (
|
||||
DEFAULT_ENDPOINT,
|
||||
LOG_KEEP,
|
||||
AnimalView,
|
||||
PlotView,
|
||||
RemoteBotClient,
|
||||
RemoteFarm,
|
||||
StateView,
|
||||
_underlying_socket,
|
||||
)
|
||||
from farm.engine import GameError
|
||||
|
||||
|
||||
# --------------------------------------------------------------------- 构造假数据
|
||||
def plot_dict(**overrides):
|
||||
"""一份「服务器会发过来的」地块 JSON。"""
|
||||
data = {
|
||||
"index": 0,
|
||||
"position": 1,
|
||||
"crop": "carrot",
|
||||
"crop_name": "胡萝卜",
|
||||
"growth": 1.0,
|
||||
"grow_days": 3,
|
||||
"days_left": 2,
|
||||
"moisture": 3,
|
||||
"is_ripe": False,
|
||||
"is_empty": False,
|
||||
"stage": "生长中",
|
||||
}
|
||||
data.update(overrides)
|
||||
return data
|
||||
|
||||
|
||||
def animal_dict(**overrides):
|
||||
data = {
|
||||
"id": 7,
|
||||
"kind": "cow",
|
||||
"name": "奶牛",
|
||||
"product": "milk",
|
||||
"fed": True,
|
||||
"pending": 0,
|
||||
"cycle": 3,
|
||||
"progress": 1,
|
||||
"starve": 0,
|
||||
"days_to_product": 2,
|
||||
"is_sick": False,
|
||||
}
|
||||
data.update(overrides)
|
||||
return data
|
||||
|
||||
|
||||
def offline_client(**kwargs):
|
||||
"""不连网络的 RemoteFarm（autostart=False 是关键）。"""
|
||||
kwargs.setdefault("autostart", False)
|
||||
kwargs.setdefault("realtime", False)
|
||||
return RemoteFarm(**kwargs)
|
||||
|
||||
|
||||
def free_port() -> int:
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||||
sock.bind(("127.0.0.1", 0))
|
||||
return int(sock.getsockname()[1])
|
||||
|
||||
|
||||
class _StubHandler(http.server.BaseHTTPRequestHandler):
|
||||
"""按路径吐预设响应的小桩，够本文件用。**
|
||||
|
||||
``payloads`` 由测试在 ``setUp`` 里填：path -> (status, body_bytes, content_type)。
|
||||
"""
|
||||
|
||||
protocol_version = "HTTP/1.1"
|
||||
payloads: ClassVar[dict[str, tuple[int, bytes, str]]] = {}
|
||||
|
||||
def do_GET(self) -> None:
|
||||
self._reply()
|
||||
|
||||
def do_POST(self) -> None:
|
||||
length = int(self.headers.get("Content-Length") or 0)
|
||||
if length:
|
||||
self.rfile.read(length)
|
||||
self._reply()
|
||||
|
||||
def _reply(self) -> None:
|
||||
path = urllib.parse.urlsplit(self.path).path
|
||||
status, body, content_type = self.payloads.get(
|
||||
path, (404, b'{"ok": false, "error": "not_found"}', "application/json")
|
||||
)
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", content_type)
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def log_message(self, *args) -> None:
|
||||
"""别把访问日志刷到测试输出里。"""
|
||||
|
||||
|
||||
# ===================================================================== 视图字段映射
|
||||
class PlotViewTestCase(unittest.TestCase):
|
||||
def test_fields_follow_the_server_payload(self):
|
||||
view = PlotView(plot_dict())
|
||||
self.assertEqual(view.index, 0)
|
||||
self.assertEqual(view.position, 1)
|
||||
self.assertEqual(view.crop, "carrot")
|
||||
self.assertEqual(view.crop_name, "胡萝卜")
|
||||
self.assertEqual(view.growth, 1.0)
|
||||
self.assertEqual(view.grow_days, 3)
|
||||
self.assertEqual(view.days_left, 2)
|
||||
self.assertEqual(view.moisture, 3)
|
||||
self.assertFalse(view.is_ripe)
|
||||
self.assertFalse(view.is_empty)
|
||||
self.assertEqual(view.stage, "生长中")
|
||||
|
||||
def test_missing_fields_fall_back_to_defaults(self):
|
||||
view = PlotView({})
|
||||
self.assertEqual(view.index, 0)
|
||||
# position 缺省时由 index 推出（地块编号是给人看的，从 1 起）
|
||||
self.assertEqual(view.position, 1)
|
||||
self.assertIsNone(view.crop)
|
||||
self.assertEqual(view.crop_name, "")
|
||||
self.assertEqual(view.growth, 0.0)
|
||||
self.assertEqual(view.moisture, 0)
|
||||
self.assertEqual(view.stage, "")
|
||||
|
||||
def test_position_defaults_to_index_plus_one(self):
|
||||
self.assertEqual(PlotView({"index": 4}).position, 5)
|
||||
|
||||
def test_empty_defaults_to_crop_is_none(self):
|
||||
self.assertTrue(PlotView({"crop": None}).is_empty)
|
||||
self.assertFalse(PlotView({"crop": "carrot"}).is_empty)
|
||||
|
||||
def test_explicit_is_empty_wins_over_crop(self):
|
||||
self.assertTrue(PlotView({"crop": "carrot", "is_empty": True}).is_empty)
|
||||
|
||||
def test_numeric_strings_are_coerced(self):
|
||||
# JSON 里数字有可能是字符串，视图不该把它漏给渲染层
|
||||
view = PlotView({"growth": "2.5", "moisture": "3", "is_ripe": 1})
|
||||
self.assertEqual(view.growth, 2.5)
|
||||
self.assertEqual(view.moisture, 3)
|
||||
self.assertIs(view.is_ripe, True)
|
||||
|
||||
def test_to_dict_is_a_copy(self):
|
||||
view = PlotView(plot_dict())
|
||||
dumped = view.to_dict()
|
||||
dumped["crop"] = "corn"
|
||||
self.assertEqual(view.crop, "carrot")
|
||||
|
||||
def test_view_is_read_only(self):
|
||||
view = PlotView(plot_dict())
|
||||
with self.assertRaises(AttributeError):
|
||||
view.index = 9 # type: ignore[misc]
|
||||
|
||||
|
||||
class AnimalViewTestCase(unittest.TestCase):
|
||||
def test_fields_follow_the_server_payload(self):
|
||||
view = AnimalView(animal_dict())
|
||||
self.assertEqual(view.aid, 7)
|
||||
self.assertEqual(view.kind, "cow")
|
||||
self.assertEqual(view.name, "奶牛")
|
||||
self.assertEqual(view.product, "milk")
|
||||
self.assertTrue(view.fed)
|
||||
self.assertEqual(view.pending, 0)
|
||||
self.assertEqual(view.cycle, 3)
|
||||
self.assertEqual(view.progress, 1)
|
||||
self.assertEqual(view.days_to_product, 2)
|
||||
self.assertFalse(view.is_sick)
|
||||
|
||||
def test_id_maps_to_aid(self):
|
||||
# 服务器用 "id"，本地模型叫 .aid —— 同构靠的就是这一层改名
|
||||
self.assertEqual(AnimalView({"id": 12}).aid, 12)
|
||||
|
||||
def test_missing_fields_fall_back_to_defaults(self):
|
||||
view = AnimalView({})
|
||||
self.assertEqual(view.aid, 0)
|
||||
self.assertEqual(view.kind, "")
|
||||
self.assertEqual(view.product, "")
|
||||
self.assertFalse(view.fed)
|
||||
self.assertEqual(view.cycle, 1) # 牲畜至少处在第 1 个周期
|
||||
self.assertEqual(view.starve, 0)
|
||||
self.assertFalse(view.is_sick)
|
||||
|
||||
def test_to_dict_is_a_copy(self):
|
||||
view = AnimalView(animal_dict())
|
||||
dumped = view.to_dict()
|
||||
dumped["kind"] = "sheep"
|
||||
self.assertEqual(view.kind, "cow")
|
||||
|
||||
|
||||
# ===================================================================== 状态视图语义
|
||||
class StateViewTestCase(unittest.TestCase):
|
||||
def test_full_snapshot_replaces_everything(self):
|
||||
view = StateView({"day": 1, "plots": [plot_dict()], "animals": [animal_dict()]})
|
||||
self.assertEqual(len(view.plots), 1)
|
||||
self.assertEqual(len(view.animals), 1)
|
||||
|
||||
view.apply({"day": 2, "plots": [], "animals": []})
|
||||
self.assertEqual(view.day, 2)
|
||||
self.assertEqual(view.plots, [])
|
||||
self.assertEqual(view.animals, [])
|
||||
|
||||
def test_brief_state_merges_and_keeps_the_farm(self):
|
||||
"""回归：动作回执里的精简 state 绝不能清空农田。"""
|
||||
view = StateView(
|
||||
{"day": 1, "gold": 100, "plots": [plot_dict(), plot_dict(index=1)], "animals": []}
|
||||
)
|
||||
view.apply({"gold": 60, "ap": 5})
|
||||
|
||||
self.assertEqual(view.gold, 60)
|
||||
self.assertEqual(view.ap, 5)
|
||||
self.assertEqual(view.day, 1, "没提到的字段要保留")
|
||||
self.assertEqual(len(view.plots), 2, "一次播种不该让农田消失")
|
||||
|
||||
def test_brief_state_after_a_full_snapshot_keeps_views_usable(self):
|
||||
view = StateView({"plots": [plot_dict(crop="corn")], "animals": [animal_dict()]})
|
||||
view.apply({"gold": 1})
|
||||
self.assertEqual(view.plots[0].crop, "corn")
|
||||
self.assertEqual(len(view.plots), 1)
|
||||
self.assertEqual(len(view.animals), 1)
|
||||
|
||||
def test_empty_data_is_ignored(self):
|
||||
view = StateView({"day": 3, "gold": 50})
|
||||
view.apply({})
|
||||
self.assertEqual(view.day, 3)
|
||||
self.assertEqual(view.gold, 50)
|
||||
|
||||
def test_initial_construction_is_tolerant(self):
|
||||
view = StateView()
|
||||
self.assertEqual(view.plots, [])
|
||||
self.assertEqual(view.animals, [])
|
||||
self.assertEqual(view.day, 1)
|
||||
self.assertEqual(view.gold, 0)
|
||||
self.assertEqual(view.level, 1)
|
||||
self.assertEqual(view.market, {})
|
||||
|
||||
def test_an_empty_view_looks_like_an_empty_farm(self):
|
||||
"""没收到任何快照时也必须能安全渲染。
|
||||
|
||||
服务器连不上、或客户端是 ``autostart=False`` 起的，界面拿到的是空视图；
|
||||
这时读 ``.plots`` 必须是"空农田"，不能是 AttributeError。
|
||||
"""
|
||||
view = StateView()
|
||||
self.assertTrue(hasattr(view, "plots"))
|
||||
self.assertTrue(hasattr(view, "animals"))
|
||||
self.assertEqual(len(view.plots), 0)
|
||||
self.assertEqual(len(view.animals), 0)
|
||||
self.assertEqual(view.to_dict(), {})
|
||||
|
||||
def test_offline_client_exposes_a_renderable_empty_state(self):
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/")
|
||||
self.assertEqual(client.state.plots, [])
|
||||
self.assertEqual(client.snapshot(), {})
|
||||
|
||||
def test_non_dict_entries_are_filtered_out(self):
|
||||
view = StateView({"plots": [plot_dict(), "垃圾", 3], "animals": [None, animal_dict()]})
|
||||
self.assertEqual(len(view.plots), 1)
|
||||
self.assertEqual(len(view.animals), 1)
|
||||
|
||||
def test_scalar_accessors_read_through(self):
|
||||
view = StateView(
|
||||
{
|
||||
"day": 5,
|
||||
"gold": 320,
|
||||
"level": 4,
|
||||
"exp": 11,
|
||||
"ap": 2,
|
||||
"feed": 9,
|
||||
"market": {"corn": 12},
|
||||
}
|
||||
)
|
||||
self.assertEqual((view.day, view.gold, view.level), (5, 320, 4))
|
||||
self.assertEqual((view.exp, view.ap, view.feed), (11, 2, 9))
|
||||
self.assertEqual(view.market["corn"], 12)
|
||||
|
||||
def test_get_and_item_access(self):
|
||||
view = StateView({"online": 3, "paused": False})
|
||||
self.assertEqual(view.get("online"), 3)
|
||||
self.assertEqual(view.get("nope", "兜底"), "兜底")
|
||||
self.assertEqual(view["online"], 3)
|
||||
with self.assertRaises(KeyError):
|
||||
_ = view["nope"]
|
||||
|
||||
def test_to_dict_round_trips_the_raw_payload(self):
|
||||
payload = {"day": 2, "gold": 5, "plots": [plot_dict()]}
|
||||
self.assertEqual(StateView(payload).to_dict(), payload)
|
||||
|
||||
def test_to_dict_is_a_copy(self):
|
||||
view = StateView({"gold": 5})
|
||||
dumped = view.to_dict()
|
||||
dumped["gold"] = 999
|
||||
self.assertEqual(view.gold, 5)
|
||||
|
||||
|
||||
# ===================================================================== 地址与 URL
|
||||
class AddressTestCase(unittest.TestCase):
|
||||
def test_normalize_adds_the_scheme(self):
|
||||
self.assertEqual(RemoteFarm._normalize("127.0.0.1:8765"), "http://127.0.0.1:8765")
|
||||
|
||||
def test_normalize_maps_the_legacy_tcp_scheme(self):
|
||||
self.assertEqual(RemoteFarm._normalize("tcp://127.0.0.1:8765"), "http://127.0.0.1:8765")
|
||||
|
||||
def test_normalize_strips_whitespace_and_trailing_slashes(self):
|
||||
self.assertEqual(RemoteFarm._normalize(" http://host:9000/// "), "http://host:9000")
|
||||
|
||||
def test_normalize_keeps_https(self):
|
||||
self.assertEqual(RemoteFarm._normalize("https://farm.example"), "https://farm.example")
|
||||
|
||||
def test_normalize_falls_back_when_empty(self):
|
||||
self.assertEqual(RemoteFarm._normalize(""), DEFAULT_ENDPOINT)
|
||||
self.assertEqual(RemoteFarm._normalize(" "), DEFAULT_ENDPOINT)
|
||||
|
||||
def test_normalize_falls_back_for_degenerate_endpoints(self):
|
||||
"""畸形端点不能变成"没有主机名"的地址。
|
||||
|
||||
否则 host 是空串，``net.is_loopback`` 判成 False，请求会溜到系统代理上，
|
||||
报回来一个和真实原因毫无关系的 502。回落到本机默认才是对的。
|
||||
"""
|
||||
for endpoint in ("", " ", "//", "http://", "https://", "tcp://"):
|
||||
with self.subTest(endpoint=endpoint):
|
||||
self.assertEqual(RemoteFarm._normalize(endpoint), DEFAULT_ENDPOINT)
|
||||
|
||||
def test_a_normalized_endpoint_always_has_a_host(self):
|
||||
candidates = ("", " ", "http://", "127.0.0.1:1", "http://host:9000/", "tcp://h:1")
|
||||
for endpoint in candidates:
|
||||
with self.subTest(endpoint=endpoint):
|
||||
normalized = RemoteFarm._normalize(endpoint)
|
||||
self.assertTrue(urllib.parse.urlsplit(normalized).netloc, normalized)
|
||||
|
||||
def test_init_stores_a_normalized_endpoint(self):
|
||||
client = offline_client(endpoint=" tcp://127.0.0.1:9999/ ")
|
||||
self.assertEqual(client.endpoint, "http://127.0.0.1:9999")
|
||||
self.assertEqual(client.host, "127.0.0.1:9999")
|
||||
|
||||
def test_farm_name_defaults_to_default(self):
|
||||
self.assertEqual(offline_client(farm="").farm, "default")
|
||||
self.assertEqual(offline_client(farm=None).farm, "default") # type: ignore[arg-type]
|
||||
|
||||
def test_host_falls_back_when_the_endpoint_has_none(self):
|
||||
self.assertEqual(offline_client(endpoint="http://").host, "127.0.0.1:8765")
|
||||
|
||||
def test_url_always_carries_the_farm(self):
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="阿黄")
|
||||
parts = urllib.parse.urlsplit(client.url("/api/state"))
|
||||
self.assertEqual(parts.path, "/api/state")
|
||||
self.assertEqual(urllib.parse.parse_qs(parts.query), {"farm": ["阿黄"]})
|
||||
|
||||
def test_url_includes_the_token_when_present(self):
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a", token="s3cret")
|
||||
query = urllib.parse.parse_qs(urllib.parse.urlsplit(client.url("/api/state")).query)
|
||||
self.assertEqual(query["token"], ["s3cret"])
|
||||
|
||||
def test_url_omits_the_token_when_absent(self):
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a")
|
||||
query = urllib.parse.parse_qs(urllib.parse.urlsplit(client.url("/api/state")).query)
|
||||
self.assertNotIn("token", query)
|
||||
|
||||
def test_url_carries_extra_params(self):
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a")
|
||||
url = client.url("/api/log", limit=30)
|
||||
query = urllib.parse.parse_qs(urllib.parse.urlsplit(url).query)
|
||||
self.assertEqual(query["limit"], ["30"])
|
||||
|
||||
def test_url_drops_params_that_are_none(self):
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a")
|
||||
url = client.url("/api/log", limit=None, since=0)
|
||||
query = urllib.parse.parse_qs(urllib.parse.urlsplit(url).query)
|
||||
self.assertNotIn("limit", query)
|
||||
self.assertEqual(query["since"], ["0"], "0 是有效值，不能当 None 丢掉")
|
||||
|
||||
def test_url_is_fully_qualified(self):
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a")
|
||||
self.assertTrue(client.url("/api/state").startswith("http://127.0.0.1:1/api/state?"))
|
||||
|
||||
def test_stream_url_carries_farm_and_name(self):
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a", name="阿黄")
|
||||
parts = urllib.parse.urlsplit(client.stream_url)
|
||||
self.assertEqual(parts.path, "/api/stream")
|
||||
query = urllib.parse.parse_qs(parts.query)
|
||||
self.assertEqual(query["farm"], ["a"])
|
||||
# 令牌只能走查询参数：EventSource / urllib 都没法给长连接加自定义头
|
||||
self.assertEqual(query["name"], ["阿黄"])
|
||||
|
||||
def test_stream_url_skips_the_empty_name(self):
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a", name="")
|
||||
query = urllib.parse.parse_qs(urllib.parse.urlsplit(client.stream_url).query)
|
||||
self.assertNotIn("name", query)
|
||||
|
||||
def test_stream_url_includes_the_token(self):
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a", token="t")
|
||||
query = urllib.parse.parse_qs(urllib.parse.urlsplit(client.stream_url).query)
|
||||
self.assertEqual(query["token"], ["t"])
|
||||
|
||||
|
||||
# ===================================================================== SSE 消息分发
|
||||
class StreamMessageTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = offline_client(endpoint="http://127.0.0.1:1/", farm="a")
|
||||
|
||||
def push(self, message):
|
||||
import json
|
||||
|
||||
self.client._handle_stream_message(json.dumps(message, ensure_ascii=False))
|
||||
|
||||
def test_welcome_installs_state_catalog_and_server_info(self):
|
||||
self.push(
|
||||
{
|
||||
"type": "welcome",
|
||||
"farm": "b",
|
||||
"online": 4,
|
||||
"server": {"version": "2.3"},
|
||||
"catalog": {"crops": [{"key": "carrot"}]},
|
||||
"state": {"day": 6, "gold": 88, "plots": [plot_dict()], "animals": []},
|
||||
"log": [],
|
||||
}
|
||||
)
|
||||
self.assertEqual(self.client.farm, "b")
|
||||
self.assertEqual(self.client.server_info, {"version": "2.3"})
|
||||
self.assertEqual(self.client.catalog()["crops"], [{"key": "carrot"}])
|
||||
self.assertEqual(self.client.state.day, 6)
|
||||
self.assertEqual(self.client.state.gold, 88)
|
||||
self.assertEqual(self.client.state.get("online"), 4, "在线人数要合进状态视图")
|
||||
self.assertEqual(len(self.client.state.plots), 1)
|
||||
self.assertEqual(self.client.stats["pushes"], 1)
|
||||
|
||||
def test_state_and_sync_share_the_welcome_branch(self):
|
||||
for kind in ("state", "sync"):
|
||||
with self.subTest(kind=kind):
|
||||
self.push({"type": kind, "state": {"gold": 7}, "plots": None})
|
||||
self.assertEqual(self.client.state.gold, 7)
|
||||
self.assertEqual(self.client.stats["pushes"], 2)
|
||||
|
||||
def test_brief_state_push_does_not_wipe_the_farm(self):
|
||||
self.push({"type": "state", "state": {"plots": [plot_dict()], "animals": []}})
|
||||
self.push({"type": "state", "state": {"gold": 3}})
|
||||
self.assertEqual(len(self.client.state.plots), 1)
|
||||
self.assertEqual(self.client.state.gold, 3)
|
||||
|
||||
def test_push_without_state_still_counts_and_keeps_old_data(self):
|
||||
self.push({"type": "state", "state": {"day": 9}})
|
||||
self.push({"type": "state"})
|
||||
self.assertEqual(self.client.state.day, 9)
|
||||
self.assertEqual(self.client.stats["pushes"], 2)
|
||||
|
||||
def test_push_ignores_a_bad_online_value(self):
|
||||
self.push({"type": "state", "online": "很多", "state": {"day": 1}})
|
||||
self.assertIsNone(self.client.state.get("online"))
|
||||
|
||||
def test_online_is_dropped_when_the_state_is_empty(self):
|
||||
self.push({"type": "state", "online": 3})
|
||||
self.assertIsNone(self.client.state.get("online"))
|
||||
|
||||
def test_push_merges_the_log(self):
|
||||
self.push(
|
||||
{
|
||||
"type": "state",
|
||||
"state": {"day": 1},
|
||||
"log": [{"seq": 2, "text": "b"}, {"seq": 1, "text": "a"}],
|
||||
}
|
||||
)
|
||||
entries = self.client.recent_log()
|
||||
self.assertEqual([e["seq"] for e in entries], [1, 2], "日志要按 seq 排序")
|
||||
self.assertEqual([e["text"] for e in entries], ["a", "b"])
|
||||
|
||||
def test_day_message_appends_a_header_and_its_events(self):
|
||||
self.push({"type": "day", "day": 4, "events": ["奶牛生病了", "一块地旱了"]})
|
||||
entries = self.client.recent_log()
|
||||
self.assertEqual(len(entries), 3)
|
||||
self.assertEqual(entries[0]["text"], "第 4 天开始")
|
||||
self.assertEqual(entries[0]["kind"], "info")
|
||||
self.assertEqual([e["text"] for e in entries[1:]], ["奶牛生病了", "一块地旱了"])
|
||||
self.assertTrue(all(e["kind"] == "warn" for e in entries[1:]))
|
||||
self.assertTrue(all(e["day"] == 4 for e in entries))
|
||||
|
||||
def test_day_message_continues_the_sequence(self):
|
||||
self.push({"type": "state", "state": {"day": 1}, "log": [{"seq": 5, "text": "旧"}]})
|
||||
self.push({"type": "day", "day": 2, "events": ["一"]})
|
||||
self.assertEqual([e["seq"] for e in self.client.recent_log()], [5, 6, 7])
|
||||
|
||||
def test_day_message_survives_missing_events(self):
|
||||
self.push({"type": "day", "day": 2})
|
||||
entries = self.client.recent_log()
|
||||
self.assertEqual(len(entries), 1)
|
||||
self.assertEqual(entries[0]["text"], "第 2 天开始")
|
||||
|
||||
def test_error_message_is_recorded(self):
|
||||
self.push({"type": "error", "message": "牧场不存在"})
|
||||
self.assertEqual(self.client.last_error, "牧场不存在")
|
||||
self.assertEqual(self.client.stats["errors"], 1)
|
||||
|
||||
def test_bye_disconnects_with_a_reason(self):
|
||||
self.client.connected = True
|
||||
self.push({"type": "bye", "message": "服务器正在停机"})
|
||||
self.assertFalse(self.client.connected)
|
||||
self.assertEqual(self.client.last_error, "服务器正在停机")
|
||||
|
||||
def test_bye_without_a_reason_uses_a_default(self):
|
||||
self.client.connected = True
|
||||
self.push({"type": "bye"})
|
||||
self.assertFalse(self.client.connected)
|
||||
self.assertEqual(self.client.last_error, "服务器已停机")
|
||||
|
||||
def test_malformed_payloads_are_ignored(self):
|
||||
for payload in ("不是 JSON", "[1, 2, 3]", '"字符串"', "3", ""):
|
||||
with self.subTest(payload=payload):
|
||||
self.client._handle_stream_message(payload)
|
||||
self.assertEqual(self.client.stats["pushes"], 0)
|
||||
self.assertEqual(self.client.stats["errors"], 0)
|
||||
self.assertEqual(self.client.last_error, "")
|
||||
|
||||
def test_unknown_message_types_are_ignored(self):
|
||||
self.push({"type": "谁知道呢", "state": {"gold": 7}})
|
||||
self.assertEqual(self.client.state.gold, 0)
|
||||
self.assertEqual(self.client.stats["pushes"], 0)
|
||||
|
||||
|
||||
# ===================================================================== 日志与回调
|
||||
class LogAndCallbackTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = offline_client(endpoint="http://127.0.0.1:1/")
|
||||
|
||||
def test_merge_log_deduplicates_by_seq(self):
|
||||
self.client._merge_log([{"seq": 1, "text": "a"}])
|
||||
self.client._merge_log([{"seq": 1, "text": "a"}, {"seq": 2, "text": "b"}])
|
||||
self.assertEqual([e["seq"] for e in self.client.recent_log()], [1, 2])
|
||||
|
||||
def test_merge_log_skips_entries_without_a_sequence(self):
|
||||
self.client._merge_log([{"text": "无号"}, {"seq": 0, "text": "零号"}, {"seq": 1}])
|
||||
self.assertEqual([e["seq"] for e in self.client.recent_log()], [1])
|
||||
|
||||
def test_merge_log_keeps_only_the_most_recent_entries(self):
|
||||
self.client._merge_log([{"seq": n} for n in range(1, LOG_KEEP + 51)])
|
||||
entries = self.client.recent_log(limit=0)
|
||||
self.assertEqual(len(entries), LOG_KEEP)
|
||||
self.assertEqual(entries[-1]["seq"], LOG_KEEP + 50)
|
||||
|
||||
def test_recent_log_filters_by_since(self):
|
||||
self.client._merge_log([{"seq": n} for n in range(1, 6)])
|
||||
self.assertEqual([e["seq"] for e in self.client.recent_log(since=3)], [4, 5])
|
||||
|
||||
def test_recent_log_limit_takes_the_tail(self):
|
||||
self.client._merge_log([{"seq": n} for n in range(1, 6)])
|
||||
self.assertEqual([e["seq"] for e in self.client.recent_log(limit=2)], [4, 5])
|
||||
|
||||
def test_recent_log_limit_zero_returns_everything(self):
|
||||
self.client._merge_log([{"seq": n} for n in range(1, 6)])
|
||||
self.assertEqual(len(self.client.recent_log(limit=0)), 5)
|
||||
|
||||
def test_apply_fires_the_change_callback(self):
|
||||
hits = []
|
||||
self.client.on_change = lambda: hits.append(1)
|
||||
self.client._apply({"gold": 1})
|
||||
self.client._apply({"gold": 2})
|
||||
self.assertEqual(len(hits), 2)
|
||||
|
||||
def test_apply_does_not_fire_for_empty_state(self):
|
||||
hits = []
|
||||
self.client.on_change = lambda: hits.append(1)
|
||||
self.client._apply({})
|
||||
self.assertEqual(hits, [])
|
||||
|
||||
def test_a_broken_callback_cannot_break_data_sync(self):
|
||||
def boom():
|
||||
raise RuntimeError("界面炸了")
|
||||
|
||||
self.client.on_change = boom
|
||||
self.client._apply({"gold": 42})
|
||||
self.assertEqual(self.client.state.gold, 42)
|
||||
|
||||
def test_transport_label_reports_the_active_channel(self):
|
||||
self.client.mode = "sse"
|
||||
self.assertEqual(self.client.transport_label(), "SSE 实时")
|
||||
self.client.mode = "poll"
|
||||
self.assertEqual(self.client.transport_label(), "HTTP 轮询")
|
||||
self.client.mode = "http"
|
||||
self.assertEqual(self.client.transport_label(), "HTTP")
|
||||
|
||||
def test_catalog_starts_with_local_config_defaults(self):
|
||||
catalog = self.client.catalog()
|
||||
self.assertEqual(catalog["feed_price"], config.FEED_UNIT_PRICE)
|
||||
self.assertEqual(catalog["water_capacity"], config.WATER_CAPACITY)
|
||||
self.assertEqual(catalog["crops"], [])
|
||||
|
||||
|
||||
# ===================================================================== 动作转发
|
||||
class ActionForwardingTestCase(unittest.TestCase):
|
||||
"""写操作统一走 POST /api/actions —— 用假的 _http 把请求原样抓下来核对。"""
|
||||
|
||||
def setUp(self):
|
||||
self.client = offline_client(endpoint="http://127.0.0.1:1/")
|
||||
self.requests: list[tuple[str, str, dict, dict]] = []
|
||||
self.reply: dict = {}
|
||||
self.client._http = self._fake_http # type: ignore[method-assign]
|
||||
|
||||
def _fake_http(self, method, path, payload=None, **params):
|
||||
self.requests.append((method, path, payload or {}, params))
|
||||
return self.reply
|
||||
|
||||
def last(self):
|
||||
return self.requests[-1]
|
||||
|
||||
def test_plant_posts_the_right_payload(self):
|
||||
self.client.plant(3, "carrot")
|
||||
method, path, payload, params = self.last()
|
||||
self.assertEqual(method, "POST")
|
||||
self.assertEqual(path, "/api/actions")
|
||||
self.assertEqual(payload, {"action": "plant", "params": {"plot": 3, "crop": "carrot"}})
|
||||
self.assertEqual(params, {"full": 1}, "顺手要一份完整快照，省掉操作后再查一次")
|
||||
|
||||
def test_simple_actions_carry_their_arguments(self):
|
||||
cases = [
|
||||
(lambda: self.client.water(2), "water", {"plot": 2}),
|
||||
(lambda: self.client.harvest(1), "harvest", {"plot": 1}),
|
||||
(lambda: self.client.harvest_all(), "harvest_all", {}),
|
||||
(lambda: self.client.buy_animal("cow"), "buy_animal", {"kind": "cow"}),
|
||||
(lambda: self.client.collect(7), "collect", {"animal_id": 7}),
|
||||
(lambda: self.client.buy_plot(), "buy_plot", {}),
|
||||
]
|
||||
for call, action, params in cases:
|
||||
with self.subTest(action=action):
|
||||
call()
|
||||
self.assertEqual(self.last()[2], {"action": action, "params": params})
|
||||
|
||||
def test_feed_omits_quantity_when_not_given(self):
|
||||
self.client.feed(7)
|
||||
self.assertEqual(self.last()[2]["params"], {"animal_id": 7})
|
||||
|
||||
def test_feed_forwards_quantity(self):
|
||||
self.client.feed(7, 3)
|
||||
self.assertEqual(self.last()[2]["params"], {"animal_id": 7, "quantity": 3})
|
||||
|
||||
def test_buy_feed_defaults_to_bulk(self):
|
||||
self.client.buy_feed()
|
||||
self.assertEqual(self.last()[2]["params"], {"quantity": config.FEED_BULK})
|
||||
|
||||
def test_pause_without_argument_is_a_toggle(self):
|
||||
self.client.pause()
|
||||
self.assertEqual(self.last()[2], {"action": "pause", "params": {}})
|
||||
|
||||
def test_pause_forwards_an_explicit_value(self):
|
||||
self.client.pause(True)
|
||||
self.assertEqual(self.last()[2]["params"], {"paused": True})
|
||||
|
||||
def test_pause_forwards_an_explicit_false(self):
|
||||
# False 与"没传"必须区分开，否则永远只能切不能停
|
||||
self.client.pause(False)
|
||||
self.assertEqual(self.last()[2]["params"], {"paused": False})
|
||||
|
||||
def test_save_forwards_to_the_server(self):
|
||||
self.client.save()
|
||||
self.assertEqual(self.last()[2], {"action": "save", "params": {}})
|
||||
|
||||
def test_batch_posts_a_single_request(self):
|
||||
items = [{"action": "water", "params": {"plot": 1}}, {"action": "harvest", "params": {}}]
|
||||
self.client.batch(items)
|
||||
method, path, payload, params = self.last()
|
||||
self.assertEqual(method, "POST")
|
||||
self.assertEqual(path, "/api/batch")
|
||||
self.assertEqual(payload, {"actions": items})
|
||||
self.assertEqual(params, {"full": 1})
|
||||
self.assertEqual(len(self.requests), 1, "批量就该是一个请求")
|
||||
|
||||
def test_action_applies_the_returned_snapshot(self):
|
||||
self.reply = {"snapshot": {"plots": [plot_dict()], "gold": 10}, "log": [{"seq": 1}]}
|
||||
self.client.water(1)
|
||||
self.assertEqual(self.client.state.gold, 10)
|
||||
self.assertEqual(len(self.client.state.plots), 1)
|
||||
self.assertEqual(len(self.client.recent_log()), 1)
|
||||
|
||||
def test_action_falls_back_to_the_brief_state(self):
|
||||
self.reply = {"state": {"gold": 5}}
|
||||
self.client.water(1)
|
||||
self.assertEqual(self.client.state.gold, 5)
|
||||
|
||||
def test_snapshot_wins_over_brief_state(self):
|
||||
self.reply = {"snapshot": {"gold": 10}, "state": {"gold": 5}}
|
||||
self.client.water(1)
|
||||
self.assertEqual(self.client.state.gold, 10)
|
||||
|
||||
def test_reset_clears_the_log_and_refreshes(self):
|
||||
self.client._merge_log([{"seq": 1, "text": "旧"}])
|
||||
refreshed = []
|
||||
original = self.client.refresh
|
||||
self.client.refresh = lambda *a, **k: refreshed.append(1) or True # type: ignore[method-assign]
|
||||
self.client.reset()
|
||||
self.client.refresh = original # type: ignore[method-assign]
|
||||
self.assertEqual(self.client.recent_log(), [])
|
||||
self.assertEqual(self.last()[2], {"action": "reset", "params": {}})
|
||||
self.assertEqual(refreshed, [1], "重置后要重新拉全量")
|
||||
|
||||
|
||||
# ===================================================================== 实时通道
|
||||
class RealtimeTestCase(unittest.TestCase):
|
||||
def test_start_is_a_noop_when_realtime_is_off(self):
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/", realtime=False)
|
||||
client.start()
|
||||
self.assertIsNone(client._thread)
|
||||
|
||||
def test_start_spawns_only_one_thread(self):
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/", realtime=True)
|
||||
client._realtime_loop = lambda: None # type: ignore[method-assign]
|
||||
try:
|
||||
client.start()
|
||||
first = client._thread
|
||||
client.start()
|
||||
self.assertIs(client._thread, first, "重复 start 不该开第二条连接")
|
||||
finally:
|
||||
client.stop()
|
||||
|
||||
def test_stop_is_safe_without_a_connection(self):
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/")
|
||||
client.stop() # 不该抛
|
||||
self.assertTrue(client._stop.is_set())
|
||||
|
||||
def test_underlying_socket_walks_the_attribute_chain(self):
|
||||
class _Raw:
|
||||
_sock = "套接字"
|
||||
|
||||
class _Fp:
|
||||
raw = _Raw()
|
||||
|
||||
class _Response:
|
||||
fp = _Fp()
|
||||
|
||||
self.assertEqual(_underlying_socket(_Response()), "套接字")
|
||||
|
||||
def test_underlying_socket_degrades_to_none(self):
|
||||
"""掏不到就返回 None，调用方会退化成 close() —— 允许变慢，不允许崩。"""
|
||||
for fake in (object(), type("R", (), {"fp": None})(), type("R", (), {"fp": object()})()):
|
||||
with self.subTest(fake=type(fake).__name__):
|
||||
self.assertIsNone(_underlying_socket(fake))
|
||||
|
||||
def test_stop_shuts_down_the_stream_socket(self):
|
||||
events = []
|
||||
|
||||
class _Sock:
|
||||
def shutdown(self, how):
|
||||
events.append(("shutdown", how))
|
||||
|
||||
class _Resp:
|
||||
def close(self):
|
||||
events.append(("close", None))
|
||||
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/")
|
||||
client._stream_socket = _Sock()
|
||||
client._stream_response = _Resp()
|
||||
client.stop()
|
||||
|
||||
self.assertIn(("shutdown", socket.SHUT_RDWR), events)
|
||||
self.assertIn(("close", None), events)
|
||||
# shutdown 必须在 close 之前：close 会等读线程的锁，最长拖到下一次心跳
|
||||
self.assertLess(events.index(("shutdown", socket.SHUT_RDWR)), events.index(("close", None)))
|
||||
|
||||
def test_stop_tolerates_broken_stream_objects(self):
|
||||
class _Sock:
|
||||
def shutdown(self, how):
|
||||
raise OSError("已经断了")
|
||||
|
||||
class _Resp:
|
||||
def close(self):
|
||||
raise RuntimeError("也炸了")
|
||||
|
||||
client = offline_client(endpoint="http://127.0.0.1:1/")
|
||||
client._stream_socket = _Sock()
|
||||
client._stream_response = _Resp()
|
||||
client.stop() # 停机路径必须稳
|
||||
|
||||
|
||||
class BotClientTestCase(unittest.TestCase):
|
||||
def test_wait_online_returns_as_soon_as_refresh_works(self):
|
||||
client = RemoteBotClient(endpoint="http://127.0.0.1:1/", autostart=False, realtime=False)
|
||||
client.refresh = lambda *a, **k: True # type: ignore[method-assign]
|
||||
self.assertTrue(client.wait_online(seconds=5))
|
||||
|
||||
def test_wait_online_gives_up_after_the_deadline(self):
|
||||
client = RemoteBotClient(endpoint="http://127.0.0.1:1/", autostart=False, realtime=False)
|
||||
calls = []
|
||||
client.refresh = lambda *a, **k: calls.append(1) or False # type: ignore[method-assign]
|
||||
self.assertFalse(client.wait_online(seconds=0))
|
||||
self.assertEqual(calls, [], "时限已过就不该再去连")
|
||||
|
||||
|
||||
# ===================================================================== HTTP 传输层
|
||||
class HttpTransportTestCase(unittest.TestCase):
|
||||
"""用纯标准库的回环桩服务验证 _http 的成功与各类失败分支。
|
||||
|
||||
这不是"外部服务器"：桩只绑 127.0.0.1，起不来就整类跳过。
|
||||
顺带验证了 net.open_url 的回环直连——本机设了 HTTP_PROXY，
|
||||
走代理的话这里会拿到 502，而不是桩真正返回的内容。
|
||||
"""
|
||||
|
||||
server: http.server.ThreadingHTTPServer
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
payload = b'{"ok": true, "state": {"day": 3, "gold": 42}, "online": 2}'
|
||||
_StubHandler.payloads = {
|
||||
"/api/state": (200, payload, "application/json"),
|
||||
"/api/catalog": (
|
||||
200,
|
||||
b'{"ok": true, "catalog": {"feed_price": 6}}',
|
||||
"application/json",
|
||||
),
|
||||
"/api/log": (
|
||||
200,
|
||||
b'{"ok": true, "entries": [{"seq": 1, "text": "a"}]}',
|
||||
"application/json",
|
||||
),
|
||||
"/api/bad": (200, "<html>不是 JSON</html>".encode(), "text/html"),
|
||||
"/api/list": (200, b"[1, 2, 3]", "application/json"),
|
||||
"/api/empty": (204, b"", "application/json"),
|
||||
"/api/rejected": (
|
||||
400,
|
||||
'{"ok": false, "error": "locked", "message": "这块地还没解锁"}'.encode(),
|
||||
"application/json",
|
||||
),
|
||||
"/api/explode": (500, "内部错误，没有 JSON".encode(), "text/plain"),
|
||||
}
|
||||
try:
|
||||
cls.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _StubHandler)
|
||||
except OSError as exc: # pragma: no cover - 环境不允许绑端口
|
||||
raise unittest.SkipTest(f"起不了回环桩服务：{exc}") from exc
|
||||
cls.port = int(cls.server.server_address[1])
|
||||
cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
|
||||
cls.thread.start()
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.server.shutdown()
|
||||
cls.server.server_close()
|
||||
cls.thread.join(timeout=5)
|
||||
|
||||
def setUp(self):
|
||||
self.client = offline_client(endpoint=f"127.0.0.1:{self.port}", farm="测试场")
|
||||
|
||||
def test_get_returns_the_parsed_body(self):
|
||||
data = self.client._http("GET", "/api/state")
|
||||
self.assertEqual(data["state"]["gold"], 42)
|
||||
self.assertTrue(self.client.connected)
|
||||
self.assertEqual(self.client.last_error, "")
|
||||
self.assertEqual(self.client.stats["http"], 1)
|
||||
|
||||
def test_loopback_requests_do_not_go_through_the_proxy(self):
|
||||
# 桩返回的是我们自己的 JSON；被系统代理截走的话这里会变成 502 页面
|
||||
self.assertEqual(self.client._http("GET", "/api/catalog")["catalog"]["feed_price"], 6)
|
||||
|
||||
def test_open_url_accepts_a_plain_string(self):
|
||||
url = f"http://127.0.0.1:{self.port}/api/state"
|
||||
with net.open_url(url, timeout=5) as response:
|
||||
self.assertEqual(response.status, 200)
|
||||
self.assertIn("gold", response.read().decode("utf-8"))
|
||||
|
||||
def test_refresh_pulls_state_catalog_and_log(self):
|
||||
self.assertTrue(self.client.refresh())
|
||||
self.assertEqual(self.client.state.day, 3)
|
||||
self.assertEqual(self.client.state.gold, 42)
|
||||
self.assertEqual(self.client.state.get("online"), 2)
|
||||
self.assertEqual(self.client.catalog()["feed_price"], 6)
|
||||
self.assertEqual([e["seq"] for e in self.client.recent_log()], [1])
|
||||
self.assertEqual(self.client.stats["http"], 3)
|
||||
|
||||
def test_refresh_swallows_errors_by_default(self):
|
||||
client = offline_client(endpoint="127.0.0.1:1", farm="测试场") # 没人监听
|
||||
self.assertFalse(client.refresh())
|
||||
self.assertFalse(client.connected)
|
||||
self.assertTrue(client.last_error)
|
||||
|
||||
def test_refresh_can_raise_on_error(self):
|
||||
client = offline_client(endpoint="127.0.0.1:1", farm="测试场")
|
||||
with self.assertRaises(GameError) as ctx:
|
||||
client.refresh(raise_on_error=True)
|
||||
self.assertEqual(ctx.exception.code, "network")
|
||||
|
||||
def test_empty_body_is_treated_as_an_empty_object(self):
|
||||
self.assertEqual(self.client._http("GET", "/api/empty"), {})
|
||||
|
||||
def test_non_json_body_is_reported_as_a_bad_response(self):
|
||||
with self.assertRaises(GameError) as ctx:
|
||||
self.client._http("GET", "/api/bad")
|
||||
self.assertEqual(ctx.exception.code, "bad_response")
|
||||
self.assertEqual(ctx.exception.status, 502)
|
||||
|
||||
def test_json_that_is_not_an_object_is_rejected(self):
|
||||
with self.assertRaises(GameError) as ctx:
|
||||
self.client._http("GET", "/api/list")
|
||||
self.assertEqual(ctx.exception.code, "bad_response")
|
||||
self.assertIn("不是对象", ctx.exception.message)
|
||||
|
||||
def test_server_rejection_becomes_a_game_error(self):
|
||||
with self.assertRaises(GameError) as ctx:
|
||||
self.client._http("POST", "/api/rejected", {"action": "plant"})
|
||||
self.assertEqual(ctx.exception.code, "locked")
|
||||
self.assertEqual(ctx.exception.message, "这块地还没解锁")
|
||||
self.assertEqual(ctx.exception.status, 400)
|
||||
self.assertEqual(self.client.stats["errors"], 1)
|
||||
|
||||
def test_error_responses_that_are_not_json_say_so(self):
|
||||
with self.assertRaises(GameError) as ctx:
|
||||
self.client._http("GET", "/api/explode")
|
||||
self.assertEqual(ctx.exception.code, "bad_response")
|
||||
|
||||
def test_a_dead_server_becomes_a_network_error_with_a_hint(self):
|
||||
dead = offline_client(endpoint=f"127.0.0.1:{free_port()}", farm="测试场")
|
||||
with self.assertRaises(GameError) as ctx:
|
||||
dead._http("GET", "/api/state")
|
||||
self.assertEqual(ctx.exception.code, "network")
|
||||
self.assertEqual(ctx.exception.status, 503)
|
||||
self.assertIn("main.py serve", ctx.exception.message, "错误信息要告诉用户怎么补救")
|
||||
self.assertFalse(dead.connected)
|
||||
self.assertEqual(dead.stats["errors"], 1)
|
||||
|
||||
def test_action_response_is_applied_end_to_end(self):
|
||||
_StubHandler.payloads["/api/actions"] = (
|
||||
200,
|
||||
b'{"ok": true, "snapshot": {"gold": 7, "plots": []}, "log": [{"seq": 9}]}',
|
||||
"application/json",
|
||||
)
|
||||
try:
|
||||
self.client._action("harvest_all")
|
||||
finally:
|
||||
_StubHandler.payloads.pop("/api/actions", None)
|
||||
self.assertEqual(self.client.state.gold, 7)
|
||||
self.assertEqual([e["seq"] for e in self.client.recent_log()], [9])
|
||||
|
||||
|
||||
# ===================================================================== 代理绕行
|
||||
class NetTestCase(unittest.TestCase):
|
||||
def test_loopback_hosts_are_recognized(self):
|
||||
for host in ("127.0.0.1", "127.0.0.5", "localhost", "LocalHost", "::1", "[::1]", ""):
|
||||
with self.subTest(host=host):
|
||||
self.assertTrue(net.is_loopback(host))
|
||||
|
||||
def test_loopback_check_tolerates_surrounding_space(self):
|
||||
self.assertTrue(net.is_loopback(" 127.0.0.1 "))
|
||||
|
||||
def test_remote_hosts_are_not_loopback(self):
|
||||
for host in ("farm.example.com", "10.0.0.1", "192.168.1.9", "128.0.0.1", None):
|
||||
with self.subTest(host=host):
|
||||
self.assertFalse(net.is_loopback(host))
|
||||
|
||||
def test_host_of_extracts_from_a_string_or_a_request(self):
|
||||
self.assertEqual(net.host_of("http://127.0.0.1:8765/api/state"), "127.0.0.1")
|
||||
request = urllib.request.Request("http://example.com:9000/x")
|
||||
self.assertEqual(net.host_of(request), "example.com")
|
||||
|
||||
def test_host_of_returns_none_when_unparseable(self):
|
||||
self.assertIsNone(net.host_of("没有协议也没有主机"))
|
||||
|
||||
def test_endpoint_host_accepts_a_bare_host_port(self):
|
||||
self.assertEqual(net.endpoint_host("127.0.0.1:8765"), "127.0.0.1")
|
||||
|
||||
def test_endpoint_host_falls_back_to_the_default(self):
|
||||
self.assertEqual(net.endpoint_host("http://"), "127.0.0.1")
|
||||
self.assertEqual(net.endpoint_host("", default="10.0.0.1"), "10.0.0.1")
|
||||
|
||||
def test_the_loopback_opener_registers_no_proxy_handler_at_all(self):
|
||||
"""回环直连的机制：这个 opener 里压根没有代理处理器。
|
||||
|
||||
``ProxyHandler({})`` 不会注册任何 ``*_open`` 钩子，而 ``add_handler``
|
||||
只在"至少挂上一个钩子"时才把 handler 记进 ``handlers``，所以它自己
|
||||
反而没被收进去——结果是 http 请求直接进 ``HTTPHandler``，必然直连。
|
||||
"""
|
||||
handlers = net._no_proxy_opener.handlers
|
||||
self.assertFalse([h for h in handlers if isinstance(h, urllib.request.ProxyHandler)])
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
unittest.main()
|
||||
+887
-433
File diff suppressed because it is too large.
Load diff
+1130
-483
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,655 @@
|
||||
"""键盘与 TUI 测试：这里曾经漏掉一个只有真按下去才会发现的 bug。
|
||||
|
||||
历史事故
|
||||
--------
|
||||
``keys.py`` 在 Windows 上返回**大写** ``"UP"``，而 ``tui.py`` 的处理表里写的是
|
||||
**小写** ``"up"``，于是方向键、空格、Tab、ESC 全部**静默失效**——字母键因为
|
||||
``char.lower()`` 恰好对得上而正常，所以肉眼很难察觉。
|
||||
|
||||
因此这里的测试分两层：
|
||||
* 约定层：两套平台映射表必须落在**同一批小写键名**里；
|
||||
* 行为层：逐个具名键驱动**真实的** :class:`farm.tui.TuiApp`，确认真的有人处理。
|
||||
|
||||
uv run python -m unittest tests.test_tui_keys -v
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from farm import config, keys
|
||||
from farm.engine import FarmEngine
|
||||
from farm.keys import KeyReader
|
||||
from farm.tui import (
|
||||
ANSI_RE,
|
||||
TuiApp,
|
||||
bar,
|
||||
display_width,
|
||||
draw_box,
|
||||
hstack,
|
||||
pad,
|
||||
strip_ansi,
|
||||
truncate,
|
||||
)
|
||||
|
||||
# 主模式下"按下去必须有反应"的具名键。ENTER / HOME / END / DELETE 等在主界面
|
||||
# 没有绑定，静默忽略是设计如此，所以不在此列（ESC 同理：它只在菜单里取消）。
|
||||
NAMED_KEYS_WITH_MAIN_BINDINGS = (
|
||||
keys.UP,
|
||||
keys.DOWN,
|
||||
keys.LEFT,
|
||||
keys.RIGHT,
|
||||
keys.SPACE,
|
||||
keys.TAB,
|
||||
keys.CTRL_C,
|
||||
)
|
||||
|
||||
|
||||
def fake_reader(data: bytes) -> KeyReader:
|
||||
"""造一个只带解析能力的 KeyReader：绕过 tty 初始化，直接喂字节。"""
|
||||
reader = KeyReader.__new__(KeyReader)
|
||||
reader._fd = None
|
||||
reader._saved_term = None
|
||||
reader._raw = data
|
||||
reader.reason = ""
|
||||
reader.available = True
|
||||
return reader
|
||||
|
||||
|
||||
def make_app(**kwargs) -> TuiApp:
|
||||
"""真实引擎 + 真实 TuiApp，只是不写盘、不上色，方便断言。"""
|
||||
engine = FarmEngine()
|
||||
engine.state.gold = 3000
|
||||
directory = tempfile.mkdtemp(prefix="simplefarm-tui-")
|
||||
options = {
|
||||
"save_path": os.path.join(directory, "save.json"),
|
||||
"use_color": False,
|
||||
"autosave": False,
|
||||
}
|
||||
options.update(kwargs)
|
||||
return TuiApp(engine, **options)
|
||||
|
||||
|
||||
class KeyNameConventionTestCase(unittest.TestCase):
|
||||
"""约定层：全小写、集中定义、两端一致。"""
|
||||
|
||||
def test_every_named_key_constant_is_lowercase(self):
|
||||
for name in keys.NAMED_KEYS:
|
||||
with self.subTest(key=name):
|
||||
self.assertIsInstance(name, str)
|
||||
self.assertEqual(name, name.lower(), f"{name!r} 必须全小写")
|
||||
self.assertNotEqual(name, "", "空字符串不能当键名")
|
||||
|
||||
def test_named_keys_is_exactly_the_set_of_constants(self):
|
||||
constants = {
|
||||
getattr(keys, attr)
|
||||
for attr in dir(keys)
|
||||
if attr.isupper() and isinstance(getattr(keys, attr), str)
|
||||
}
|
||||
self.assertEqual(set(keys.NAMED_KEYS), constants)
|
||||
|
||||
def test_named_keys_do_not_collide_with_typeable_characters(self):
|
||||
"""具名键不能是单个字符，否则会和普通按键混在一起分不清。"""
|
||||
for name in keys.NAMED_KEYS:
|
||||
with self.subTest(key=name):
|
||||
self.assertGreater(len(name), 1)
|
||||
|
||||
def test_constant_values_are_the_lowercase_form_of_their_names(self):
|
||||
self.assertEqual(keys.UP, "up")
|
||||
self.assertEqual(keys.DOWN, "down")
|
||||
self.assertEqual(keys.LEFT, "left")
|
||||
self.assertEqual(keys.RIGHT, "right")
|
||||
self.assertEqual(keys.SPACE, "space")
|
||||
self.assertEqual(keys.TAB, "tab")
|
||||
self.assertEqual(keys.ESC, "esc")
|
||||
self.assertEqual(keys.ENTER, "enter")
|
||||
self.assertEqual(keys.CTRL_C, "ctrl_c")
|
||||
self.assertEqual(keys.BACKSPACE, "backspace")
|
||||
self.assertEqual(keys.DELETE, "delete")
|
||||
self.assertEqual(keys.HOME, "home")
|
||||
self.assertEqual(keys.END, "end")
|
||||
|
||||
def test_platform_tables_only_produce_known_names(self):
|
||||
tables = {
|
||||
"windows 扩展键": keys._WIN_EXTENDED,
|
||||
"windows 控制键": keys._WIN_CTRL,
|
||||
"posix 控制键": keys._POSIX_CTRL,
|
||||
"posix 转义序列": keys._POSIX_CSI,
|
||||
}
|
||||
for label, table in tables.items():
|
||||
for raw, name in table.items():
|
||||
with self.subTest(table=label, raw=raw):
|
||||
self.assertIn(name, keys.NAMED_KEYS, f"{label} 产出了未登记的键名 {name!r}")
|
||||
self.assertEqual(name, name.lower())
|
||||
|
||||
def test_both_platforms_produce_the_same_function_keys(self):
|
||||
"""这条断言就是那个 bug 的护栏：两端必须产出同一批方向键/功能键。"""
|
||||
windows = set(keys._WIN_EXTENDED.values())
|
||||
posix = set(keys._POSIX_CSI.values())
|
||||
self.assertEqual(windows, posix)
|
||||
self.assertEqual(
|
||||
windows, {keys.UP, keys.DOWN, keys.LEFT, keys.RIGHT, keys.HOME, keys.END, keys.DELETE}
|
||||
)
|
||||
|
||||
def test_both_platforms_produce_the_same_control_keys(self):
|
||||
windows = set(keys._WIN_CTRL.values())
|
||||
posix = set(keys._POSIX_CTRL.values())
|
||||
self.assertEqual(posix, windows - {keys.ESC}, "POSIX 的 ESC 走单独的转义序列分支")
|
||||
|
||||
|
||||
class KeyReaderAvailabilityTestCase(unittest.TestCase):
|
||||
def test_non_tty_stdin_degrades_with_a_reason(self):
|
||||
with mock.patch.object(sys, "stdin", None):
|
||||
reader = KeyReader()
|
||||
self.assertFalse(reader.available)
|
||||
self.assertIn("标准输入", reader.reason)
|
||||
|
||||
def test_piped_stdin_degrades_with_a_reason(self):
|
||||
class NotATty:
|
||||
def isatty(self) -> bool:
|
||||
return False
|
||||
|
||||
with mock.patch.object(sys, "stdin", NotATty()):
|
||||
reader = KeyReader()
|
||||
self.assertFalse(reader.available)
|
||||
self.assertNotEqual(reader.reason, "")
|
||||
|
||||
def test_unavailable_reader_returns_nothing(self):
|
||||
reader = KeyReader.__new__(KeyReader)
|
||||
reader._fd = None
|
||||
reader._saved_term = None
|
||||
reader._raw = b""
|
||||
reader.reason = "测试"
|
||||
reader.available = False
|
||||
self.assertIsNone(reader.poll())
|
||||
|
||||
def test_close_is_safe_without_a_terminal(self):
|
||||
reader = KeyReader.__new__(KeyReader)
|
||||
reader._fd = None
|
||||
reader._saved_term = None
|
||||
reader._raw = b"leftover"
|
||||
reader.reason = ""
|
||||
reader.available = False
|
||||
reader.close()
|
||||
self.assertEqual(reader._raw, b"")
|
||||
|
||||
|
||||
class PosixByteParsingTestCase(unittest.TestCase):
|
||||
"""POSIX 侧把原始字节流翻译成按键名——纯函数，不碰真终端。"""
|
||||
|
||||
def parse(self, data: bytes) -> str | None:
|
||||
return fake_reader(data)._parse_posix()
|
||||
|
||||
def drain(self, data: bytes, times: int = 8) -> list[str]:
|
||||
reader = fake_reader(data)
|
||||
out: list[str] = []
|
||||
for _ in range(times):
|
||||
key = reader._parse_posix()
|
||||
if key is None:
|
||||
break
|
||||
out.append(key)
|
||||
return out
|
||||
|
||||
def test_plain_characters_are_lowercased(self):
|
||||
self.assertEqual(self.parse(b"a"), "a")
|
||||
self.assertEqual(self.parse(b"A"), "a")
|
||||
self.assertEqual(self.parse(b"7"), "7")
|
||||
|
||||
def test_control_characters(self):
|
||||
self.assertEqual(self.parse(b"\x03"), keys.CTRL_C)
|
||||
self.assertEqual(self.parse(b"\r"), keys.ENTER)
|
||||
self.assertEqual(self.parse(b"\n"), keys.ENTER)
|
||||
self.assertEqual(self.parse(b"\t"), keys.TAB)
|
||||
self.assertEqual(self.parse(b"\x7f"), keys.BACKSPACE)
|
||||
self.assertEqual(self.parse(b" "), keys.SPACE)
|
||||
|
||||
def test_arrow_keys(self):
|
||||
self.assertEqual(self.parse(b"\x1b[A"), keys.UP)
|
||||
self.assertEqual(self.parse(b"\x1b[B"), keys.DOWN)
|
||||
self.assertEqual(self.parse(b"\x1b[C"), keys.RIGHT)
|
||||
self.assertEqual(self.parse(b"\x1b[D"), keys.LEFT)
|
||||
|
||||
def test_home_end_and_delete(self):
|
||||
self.assertEqual(self.parse(b"\x1b[H"), keys.HOME)
|
||||
self.assertEqual(self.parse(b"\x1b[F"), keys.END)
|
||||
self.assertEqual(self.parse(b"\x1b[1~"), keys.HOME)
|
||||
self.assertEqual(self.parse(b"\x1b[3~"), keys.DELETE)
|
||||
self.assertEqual(self.parse(b"\x1b[4~"), keys.END)
|
||||
|
||||
def test_application_mode_arrows(self):
|
||||
"""有些终端把方向键发成 ESC O A 而不是 ESC [ A。"""
|
||||
self.assertEqual(self.parse(b"\x1bOA"), keys.UP)
|
||||
self.assertEqual(self.parse(b"\x1bOD"), keys.LEFT)
|
||||
|
||||
def test_lone_escape_is_the_escape_key(self):
|
||||
self.assertEqual(self.parse(b"\x1b"), keys.ESC)
|
||||
|
||||
def test_escape_followed_by_a_plain_character(self):
|
||||
reader = fake_reader(b"\x1bX")
|
||||
self.assertEqual(reader._parse_posix(), keys.ESC)
|
||||
self.assertEqual(reader._raw, b"X", "多出来的字符要留给下一轮，不能吞掉")
|
||||
|
||||
def test_unknown_sequence_is_dropped_not_guessed(self):
|
||||
"""认不出来的功能键宁可丢掉，也不能退化成一个错误的动作。"""
|
||||
self.assertIsNone(self.parse(b"\x1b[99~"))
|
||||
|
||||
def test_utf8_character_is_decoded_as_a_whole(self):
|
||||
self.assertEqual(self.parse("中".encode()), "中")
|
||||
|
||||
def test_half_a_character_waits_for_the_rest(self):
|
||||
reader = fake_reader("中".encode()[:2])
|
||||
self.assertIsNone(reader._parse_posix())
|
||||
self.assertEqual(len(reader._raw), 2, "半个字符要留在缓冲区里等下一轮")
|
||||
reader._raw += "中".encode()[2:]
|
||||
self.assertEqual(reader._parse_posix(), "中")
|
||||
|
||||
def test_several_keys_arriving_at_once_are_returned_one_by_one(self):
|
||||
"""老实现只取第一个字节、把其余静默丢掉，快打或粘贴就会丢键。"""
|
||||
self.assertEqual(self.drain(b"abc"), ["a", "b", "c"])
|
||||
|
||||
def test_mixed_burst_keeps_order(self):
|
||||
self.assertEqual(self.drain(b"a\x1b[B"), ["a", keys.DOWN])
|
||||
|
||||
def test_backspace_then_letter(self):
|
||||
self.assertEqual(self.drain(b"\x7fp"), [keys.BACKSPACE, "p"])
|
||||
|
||||
def test_incomplete_sequence_falls_back_to_escape(self):
|
||||
"""序列读不全又不能干等：当作单按 ESC，总比把按键吞掉好。"""
|
||||
reader = fake_reader(b"\x1b[")
|
||||
self.assertEqual(reader._parse_posix(), keys.ESC)
|
||||
reader._raw = b"\x1b[A"
|
||||
self.assertEqual(reader._parse_posix(), keys.UP)
|
||||
|
||||
|
||||
class TextLayoutTestCase(unittest.TestCase):
|
||||
"""渲染用的纯函数：宽度算错的话，边框在中文环境下就会歪。"""
|
||||
|
||||
def test_ansi_codes_do_not_count_as_width(self):
|
||||
self.assertEqual(display_width("\x1b[1mhi\x1b[0m"), 2)
|
||||
|
||||
def test_cjk_characters_are_double_width(self):
|
||||
self.assertEqual(display_width("中文"), 4)
|
||||
self.assertEqual(display_width("a中"), 3)
|
||||
|
||||
def test_strip_ansi_removes_color_codes(self):
|
||||
self.assertEqual(strip_ansi("\x1b[31m红\x1b[0m"), "红")
|
||||
self.assertIsNone(ANSI_RE.search(strip_ansi("\x1b[31m红\x1b[0m")))
|
||||
|
||||
def test_pad_aligns_by_display_width(self):
|
||||
for align in ("left", "right", "center"):
|
||||
with self.subTest(align=align):
|
||||
self.assertEqual(display_width(pad("中文", 10, align)), 10)
|
||||
|
||||
def test_pad_truncates_overlong_content(self):
|
||||
self.assertEqual(display_width(pad("非常长的一段中文内容", 6)), 6)
|
||||
|
||||
def test_truncate_adds_an_ellipsis(self):
|
||||
text = truncate("非常长的一段中文内容", 7)
|
||||
self.assertLessEqual(display_width(text), 7)
|
||||
self.assertTrue(text.endswith("…"))
|
||||
|
||||
def test_truncate_keeps_short_text_intact(self):
|
||||
self.assertEqual(truncate("短的", 10), "短的")
|
||||
|
||||
def test_bar_renders_filled_and_empty_cells(self):
|
||||
self.assertEqual(bar(0, 10, 10), "░" * 10)
|
||||
self.assertEqual(bar(10, 10, 10), "█" * 10)
|
||||
self.assertEqual(len(bar(5, 10, 10)), 10)
|
||||
self.assertEqual(len(bar(1, 0, 4)), 4, "分母为 0 也不能崩")
|
||||
|
||||
def test_box_lines_are_exactly_the_requested_width(self):
|
||||
lines = draw_box("标题", ["内容", "另一行"], 30, use_color=False)
|
||||
for line in lines:
|
||||
with self.subTest(line=line):
|
||||
self.assertEqual(display_width(line), 30)
|
||||
|
||||
def test_hstack_pads_the_shorter_column(self):
|
||||
merged = hstack(["a", "b"], ["1"])
|
||||
self.assertEqual(len(merged), 2)
|
||||
self.assertTrue(merged[1].startswith("b"))
|
||||
|
||||
|
||||
class TuiNamedKeyBehaviourTestCase(unittest.TestCase):
|
||||
"""行为层：真的把键交给 TuiApp，看它有没有动。"""
|
||||
|
||||
def setUp(self):
|
||||
self.app = make_app()
|
||||
|
||||
def test_arrow_keys_move_the_plot_cursor(self):
|
||||
self.app.grid_cols = 3
|
||||
self.assertEqual(self.app.plot_cursor, 0)
|
||||
|
||||
self.app._handle_key(keys.RIGHT)
|
||||
self.assertEqual(self.app.plot_cursor, 1)
|
||||
self.app._handle_key(keys.DOWN)
|
||||
self.assertEqual(self.app.plot_cursor, 4)
|
||||
self.app._handle_key(keys.UP)
|
||||
self.assertEqual(self.app.plot_cursor, 1)
|
||||
self.app._handle_key(keys.LEFT)
|
||||
self.assertEqual(self.app.plot_cursor, 0)
|
||||
|
||||
def test_cursor_cannot_leave_the_grid(self):
|
||||
self.app.grid_cols = 3
|
||||
self.app._handle_key(keys.LEFT)
|
||||
self.assertEqual(self.app.plot_cursor, 0)
|
||||
self.app._handle_key(keys.UP)
|
||||
self.assertEqual(self.app.plot_cursor, 0)
|
||||
|
||||
def test_space_toggles_time(self):
|
||||
self.assertFalse(self.app.engine.paused)
|
||||
self.app._handle_key(keys.SPACE)
|
||||
self.assertTrue(self.app.engine.paused)
|
||||
self.app._handle_key(keys.SPACE)
|
||||
self.assertFalse(self.app.engine.paused)
|
||||
|
||||
def test_tab_cycles_through_animals(self):
|
||||
self.app._handle_key(keys.TAB)
|
||||
self.assertIn("还没有动物", self.app.message)
|
||||
|
||||
for _ in range(2):
|
||||
self.app.engine.buy_animal("chicken")
|
||||
self.app._handle_key(keys.TAB)
|
||||
self.assertEqual(self.app.animal_cursor, 1)
|
||||
self.app._handle_key(keys.TAB)
|
||||
self.assertEqual(self.app.animal_cursor, 0, "应当循环回第一只")
|
||||
|
||||
def test_ctrl_c_quits(self):
|
||||
self.assertTrue(self.app.running)
|
||||
self.app._handle_key(keys.CTRL_C)
|
||||
self.assertFalse(self.app.running)
|
||||
|
||||
def test_every_bound_named_key_does_something(self):
|
||||
"""这是那个 bug 的正面护栏：逐个具名键驱动真实 TuiApp。"""
|
||||
for key in NAMED_KEYS_WITH_MAIN_BINDINGS:
|
||||
with self.subTest(key=key):
|
||||
app = make_app()
|
||||
app.engine.buy_animal("chicken")
|
||||
app.engine.buy_animal("chicken")
|
||||
app.grid_cols = 3
|
||||
app.plot_cursor = 4 # 摆在中间，上下左右才都动得了
|
||||
before = (
|
||||
app.running,
|
||||
app.plot_cursor,
|
||||
app.animal_cursor,
|
||||
app.mode,
|
||||
app.engine.paused,
|
||||
)
|
||||
app._handle_key(key)
|
||||
after = (
|
||||
app.running,
|
||||
app.plot_cursor,
|
||||
app.animal_cursor,
|
||||
app.mode,
|
||||
app.engine.paused,
|
||||
)
|
||||
self.assertNotEqual(before, after, f"按键 {key!r} 没有任何反应")
|
||||
|
||||
def test_unbound_named_keys_are_ignored_without_crashing(self):
|
||||
for key in (keys.ENTER, keys.HOME, keys.END, keys.DELETE, keys.BACKSPACE):
|
||||
with self.subTest(key=key):
|
||||
app = make_app()
|
||||
app._handle_key(key)
|
||||
self.assertEqual(app.plot_cursor, 0)
|
||||
self.assertTrue(app.running)
|
||||
|
||||
def test_unknown_key_is_ignored(self):
|
||||
self.app._handle_key("zzz")
|
||||
self.assertEqual(self.app.plot_cursor, 0)
|
||||
self.assertTrue(self.app.running)
|
||||
|
||||
|
||||
class TuiLetterKeyBehaviourTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.app = make_app()
|
||||
self.engine = self.app.engine
|
||||
|
||||
def test_p_opens_the_crop_menu_and_a_digit_plants(self):
|
||||
self.app._handle_key("p")
|
||||
self.assertEqual(self.app.mode, "select_crop")
|
||||
self.assertTrue(self.app.menu)
|
||||
self.app._handle_key("1")
|
||||
self.assertEqual(self.app.mode, "main")
|
||||
self.assertIsNotNone(self.engine.state.plots[0].crop)
|
||||
|
||||
def test_escape_closes_the_menu(self):
|
||||
self.app._handle_key("p")
|
||||
self.app._handle_key(keys.ESC)
|
||||
self.assertEqual(self.app.mode, "main")
|
||||
self.assertIsNone(self.engine.state.plots[0].crop)
|
||||
|
||||
def test_menu_rejects_a_locked_choice(self):
|
||||
self.engine.state.level = 1
|
||||
self.app._handle_key("p")
|
||||
locked_index = next(
|
||||
i for i, item in enumerate(self.app.menu, start=1) if item["key"] == "strawberry"
|
||||
)
|
||||
self.app._handle_key(str(locked_index))
|
||||
self.assertIsNone(self.engine.state.plots[0].crop)
|
||||
self.assertIn("尚未解锁", self.app.message)
|
||||
|
||||
def test_menu_ignores_out_of_range_choices(self):
|
||||
self.app._handle_key("p")
|
||||
self.app._handle_key("0")
|
||||
self.assertIn("没有这个选项", self.app.message)
|
||||
self.app._handle_key("9")
|
||||
self.assertIn("没有这个选项", self.app.message)
|
||||
|
||||
def test_b_buys_an_animal_through_the_menu(self):
|
||||
self.app._handle_key("b")
|
||||
self.assertEqual(self.app.mode, "select_animal")
|
||||
self.app._handle_key("1")
|
||||
self.assertEqual(len(self.engine.state.animals), 1)
|
||||
|
||||
def test_w_waters_and_h_harvests_the_selected_plot(self):
|
||||
self.engine.plant(1, "wheat")
|
||||
self.engine.state.plots[0].moisture = 0
|
||||
self.app._handle_key("w")
|
||||
self.assertEqual(self.engine.state.plots[0].moisture, config.WATER_CAPACITY)
|
||||
|
||||
self.engine.state.plots[0].growth = config.CROPS["wheat"]["grow_days"]
|
||||
gold = self.engine.state.gold
|
||||
self.app._handle_key("h")
|
||||
self.assertGreater(self.engine.state.gold, gold)
|
||||
|
||||
def test_e_harvests_everything_ripe(self):
|
||||
for number in (1, 2, 3):
|
||||
self.engine.plant(number, "wheat")
|
||||
for index in (0, 1):
|
||||
self.engine.state.plots[index].growth = 2
|
||||
self.app._handle_key("e")
|
||||
self.assertTrue(self.engine.state.plots[0].is_empty)
|
||||
self.assertTrue(self.engine.state.plots[1].is_empty)
|
||||
self.assertFalse(self.engine.state.plots[2].is_empty, "没熟的留着")
|
||||
|
||||
def test_e_without_anything_ripe_says_so(self):
|
||||
self.app._handle_key("e")
|
||||
self.assertIn("没有成熟", self.app.message)
|
||||
|
||||
def test_f_feeds_the_focused_animal(self):
|
||||
self.engine.buy_animal("chicken")
|
||||
self.app._handle_key("f")
|
||||
self.assertTrue(self.engine.state.animals[0].fed)
|
||||
|
||||
def test_c_collects_the_focused_animal(self):
|
||||
self.engine.buy_animal("chicken")
|
||||
self.engine.state.animals[0].pending = 2
|
||||
gold = self.engine.state.gold
|
||||
self.app._handle_key("c")
|
||||
self.assertGreater(self.engine.state.gold, gold)
|
||||
|
||||
def test_l_buys_land_and_k_buys_feed(self):
|
||||
self.engine.state.level = 2
|
||||
plots = len(self.engine.state.plots)
|
||||
self.app._handle_key("l")
|
||||
self.assertEqual(len(self.engine.state.plots), plots + 1)
|
||||
|
||||
feed = self.engine.state.feed
|
||||
self.app._handle_key("k")
|
||||
self.assertEqual(self.engine.state.feed, feed + config.FEED_BULK)
|
||||
|
||||
def test_s_saves_the_game(self):
|
||||
self.app.autosave = False
|
||||
self.app._handle_key("s")
|
||||
self.assertTrue(os.path.exists(self.app.save_path))
|
||||
|
||||
def test_r_asks_before_resetting(self):
|
||||
self.engine.state.day = 5
|
||||
self.app._handle_key("r")
|
||||
self.assertEqual(self.app.mode, "confirm_reset")
|
||||
|
||||
self.app._handle_key("n")
|
||||
self.assertEqual(self.app.mode, "main")
|
||||
self.assertEqual(self.engine.state.day, 5, "没确认就不该重置")
|
||||
|
||||
def test_r_then_y_resets(self):
|
||||
self.engine.state.day = 5
|
||||
self.app._handle_key("r")
|
||||
self.app._handle_key("y")
|
||||
self.assertEqual(self.engine.state.day, 1)
|
||||
|
||||
def test_question_mark_opens_help_and_any_key_closes_it(self):
|
||||
self.app._handle_key("?")
|
||||
self.assertEqual(self.app.mode, "help")
|
||||
self.app._handle_key("x")
|
||||
self.assertEqual(self.app.mode, "main")
|
||||
|
||||
def test_q_quits(self):
|
||||
self.app._handle_key("q")
|
||||
self.assertFalse(self.app.running)
|
||||
|
||||
def test_failed_action_shows_the_reason_instead_of_crashing(self):
|
||||
self.app._handle_key("w") # 空地浇水
|
||||
self.assertIn("空地", self.app.message)
|
||||
self.assertTrue(self.app.running)
|
||||
|
||||
|
||||
class TuiRenderTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.app = make_app()
|
||||
# TuiApp 自己不驱动时钟，倒计时要靠 next_tick_at 才有值
|
||||
self.app.engine.next_tick_at = time.time() + config.DEFAULT_TICK_SECONDS
|
||||
|
||||
def capture(self) -> str:
|
||||
buffer = io.StringIO()
|
||||
with contextlib.redirect_stdout(buffer):
|
||||
self.app._render()
|
||||
return buffer.getvalue()
|
||||
|
||||
def test_first_frame_contains_every_panel(self):
|
||||
output = self.render_plain()
|
||||
for label in ("SimpleFarm", "农田", "牧场", "市场行情", "农事日志"):
|
||||
with self.subTest(panel=label):
|
||||
self.assertIn(label, output)
|
||||
|
||||
def render_plain(self) -> str:
|
||||
return strip_ansi(self.capture())
|
||||
|
||||
def test_frame_fits_the_terminal_width(self):
|
||||
output = strip_ansi(self.capture())
|
||||
for line in output.splitlines():
|
||||
with self.subTest(line=line[:20]):
|
||||
self.assertLessEqual(display_width(line.rstrip("\r")), 200)
|
||||
|
||||
def test_countdown_uses_the_human_format(self):
|
||||
output = self.render_plain()
|
||||
self.assertNotIn("s \n", output)
|
||||
self.assertRegex(output, r"\d+:\d{2}")
|
||||
|
||||
def test_paused_state_is_visible(self):
|
||||
self.app.engine.pause(True)
|
||||
self.assertIn("已暂停", self.render_plain())
|
||||
|
||||
def test_crop_state_is_visible(self):
|
||||
self.app.engine.plant(1, "wheat")
|
||||
self.assertIn("小麦", self.render_plain())
|
||||
|
||||
def test_market_panel_uses_the_china_convention(self):
|
||||
"""涨红跌绿是中国习惯，别跟着欧美反过来。"""
|
||||
tui_path = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "farm", "tui.py"
|
||||
)
|
||||
with open(tui_path, encoding="utf-8") as handle:
|
||||
source = handle.read()
|
||||
self.assertIn('trend, tone = "涨", C_RED', source)
|
||||
self.assertIn('trend, tone = "跌", C_GREEN', source)
|
||||
|
||||
def test_help_page_is_current(self):
|
||||
self.app._handle_key("?")
|
||||
output = self.render_plain()
|
||||
self.assertIn("分钟", output)
|
||||
self.assertNotIn("[N]休息", output)
|
||||
self.assertNotIn("15 秒", output)
|
||||
|
||||
def test_keyboard_notice_appears_when_input_is_unavailable(self):
|
||||
self.app.reader.available = False
|
||||
self.app.reader.reason = "标准输入不是终端（管道或重定向）"
|
||||
self.app.show_keyboard_notice = True
|
||||
self.assertIn("读不到按键", self.render_plain())
|
||||
|
||||
def test_keyboard_notice_can_be_suppressed_for_previews(self):
|
||||
self.app.reader.available = False
|
||||
self.app.show_keyboard_notice = False
|
||||
output = self.render_plain()
|
||||
self.assertNotIn("读不到按键", output)
|
||||
self.assertIn("↑↓←→", output)
|
||||
|
||||
def test_menu_frame_renders_choices(self):
|
||||
self.app._handle_key("p")
|
||||
output = self.render_plain()
|
||||
self.assertIn("选择要种在", output)
|
||||
self.assertIn("小麦", output)
|
||||
|
||||
def test_auto_save_only_runs_when_enabled(self):
|
||||
app = make_app(autosave=True)
|
||||
app._exit()
|
||||
self.assertTrue(os.path.exists(app.save_path))
|
||||
|
||||
|
||||
class TuiSnapshotCompatibilityTestCase(unittest.TestCase):
|
||||
"""TUI 要能同时吃本地引擎和远程客户端——两者的快照字段必须对齐。"""
|
||||
|
||||
def test_render_works_with_a_remote_shaped_snapshot(self):
|
||||
from farm.client import StateView
|
||||
|
||||
app = make_app()
|
||||
view = StateView(app.engine.snapshot())
|
||||
app.engine = view # 用一个"长得像引擎"的对象
|
||||
lines = app._render_status(view.to_dict(), 100)
|
||||
rendered = strip_ansi("\n".join(lines))
|
||||
self.assertIn("金币", rendered)
|
||||
|
||||
def test_log_entries_are_rendered(self):
|
||||
app = make_app()
|
||||
app.engine.next_tick_at = time.time() + config.DEFAULT_TICK_SECONDS
|
||||
app.engine.add_log("测试事件", "good")
|
||||
buffer = io.StringIO()
|
||||
with contextlib.redirect_stdout(buffer):
|
||||
app._render()
|
||||
self.assertIn("测试事件", strip_ansi(buffer.getvalue()))
|
||||
|
||||
def test_recent_messages_expire(self):
|
||||
app = make_app()
|
||||
app._flash("一闪而过", seconds=0.0)
|
||||
buffer = io.StringIO()
|
||||
with contextlib.redirect_stdout(buffer):
|
||||
app._render()
|
||||
self.assertNotIn("一闪而过", strip_ansi(buffer.getvalue()))
|
||||
|
||||
def test_json_round_trip_of_the_snapshot_is_renderable(self):
|
||||
"""远程客户端的快照是过了一遍 JSON 的，字段类型可能变。"""
|
||||
self.app = make_app()
|
||||
raw = json.loads(json.dumps(self.app.engine.snapshot(), ensure_ascii=False))
|
||||
self.app._render_status(raw, 100)
|
||||
self.app._render_farm(raw, 60)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Reference in new issue
Block a user