snowy-yang
f8e093c98f
三件事凑成一次不兼容变更：WebSocket 通道、Hypercorn 服务器、「手动推进一天」都消失了。 实时通道：WebSocket -> SSE - hub 改为 SSE 订阅池，GET /api/stream 单向推送 welcome/state/day/notice/bye， 250ms 节流 + 15s 心跳注释；写操作一律走 HTTP POST /api/actions，不再有 WS 回执。 - 停机三处一起修才干净：close_all() 发 bye + timeout_graceful_shutdown 兜底 + 客户端 stop() 直接 shutdown 底层 socket（光 close() 会等读线程的锁，最长卡到下次心跳）。 - 网页版改 EventSource + fetch；tools/ws_demo.py 换成 tools/sse_demo.py。 服务器：Hypercorn -> uvicorn + zttp 的 HTTP/2 - 同一端口同时说 HTTP/1.1 与 HTTP/2。uvicorn 0.53 不再内置 h2，必须装 zttp 并显式 http="zttp"，否则 Config.load() 直接 raise。 - 信号接管改成子类覆盖 Server.handle_exit（0.53 没有 install_signal_handlers 可覆盖）， 不调 super()、不写 _captured_signals，退出时不重放信号；第二次信号强制退出。 - 入口启动前用 daemon.port_busy() 自检端口：Windows 上 SO_REUSEADDR 会让第二个实例 「绑上」已监听端口并抢走新连接，所以判断「能不能连上」而不是「能不能 bind」。 - 修掉 clear_pid_file 的 Windows 真 bug：文件还开着就 os.remove 会失败且被 except OSError 吞掉，导致每次正常退出都留 stale pid 文件。 节奏：一天 15 秒 -> 40 分钟，并整体移除「休息」 - DEFAULT_TICK_SECONDS=2400，MAX_TICK_SECONDS 放宽到 86400（旧上限会把 40 分钟夹回去）。 - engine.sleep / SLEEP_COOLDOWN / /api/sleep / TUI 的 [N]休息 / 网页版按钮全部删掉， 时间只能自然流动；机器人改为 wait_for_next_day(timeout)，CLI 加 bot --wait。 - 时长显示统一走 models.format_duration()，不再出现 2399.6s 这种写法。 键盘：修复方向键/空格/Tab/ESC 全部静默失效 - keys.py 的具名键统一小写并成为唯一来源，tui.py 一律引用 keys.XXX 而不手写字符串 （历史 bug 就是 Windows 侧产大写 UP、TUI 里写小写 up，字母键正常所以很难发现）。 - 顺带修 POSIX 路径一次 os.read 只取首字节、其余按键静默丢弃的老毛病。 测试：重建 500 项 - test_engine 110 / test_actions 35 / test_server 131(skip 35) / test_tui_keys 75 / test_bot 36 / test_client 113。未安装的第三方依赖用 skipUnless 自动跳过，其余照跑， 所以是全绿而不是靠跳过伪装。 - 被测试逼出两处真隐患并修掉：RemoteFarm._normalize() 必须「先补 scheme 再削尾斜杠」 （顺序反了畸形端点会溜到系统代理、报一个和真实原因无关的 502）；StateView 空视图要 预置 plots/animals，否则连不上服务器时读 state.plots 直接 AttributeError。 优化：先量后改，依据与前后数字都在 tools/bench.py - 新增 engine.brief_source()：brief/summary/动作回执只要十几个标量，不再为它们序列化 整局；to_dict 里 net_worth 只算一次。brief -76%、summary -77%、leaderboard(32) -78%、 GET /api/world -31%、POST /api/actions?full=1 -16%。 - 拆掉三个巨型函数：create_app 475->32 行（_Services + _register_*）、 build_parser 与 available_actions 分组、__main__.main 160->7 行。 - 消除重复：等级门槛 5 处 -> _is_locked/_require_level/_newly_unlocked；hub 的 broadcast/broadcast_all/close_all -> _push_many；daemon.probe/world -> _get_json； main.py 的 _add_endpoint_args/_leaderboard_lines。 - tools/bench.py 作为长期性能守卫（含调用计数器），谁把轻量路径改回 snapshot() 会立刻报警。 门禁：ruff format/check 全绿、pyright 0 错误、500 项测试 OK（skip 35）。
913 lines
36 KiB
Python
913 lines
36 KiB
Python
"""引擎层测试：游戏规则、边界与存档。
|
|
|
|
这一层是"纯规则"，不依赖 FastAPI / uvicorn，所以能单独跑：
|
|
|
|
uv run python -m unittest tests.test_engine -v
|
|
|
|
设计原则：**只通过公开接口断言行为**。私有方法（``_gain_exp`` 之类）只在
|
|
"没有公开路径能触达"时才碰，且要说明为什么。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from farm import config
|
|
from farm.engine import FarmEngine, GameError, Scheduler, tick_decision
|
|
from farm.models import Animal, GameState, Plot, clamp, format_duration
|
|
from farm.server.actions import run_action
|
|
from farm.storage import SAVE_VERSION, delete_save, load_state, save_state
|
|
|
|
|
|
def make_engine(**kwargs) -> FarmEngine:
|
|
"""造一个测试引擎：默认给足金币，省得每个用例都先攒钱。"""
|
|
engine = FarmEngine(**kwargs)
|
|
engine.state.gold = 2000
|
|
return engine
|
|
|
|
|
|
class FormatDurationTestCase(unittest.TestCase):
|
|
"""时长格式化：界面与命令行都靠它，别让 2400 秒显示成 "2399.6s"。"""
|
|
|
|
def test_none_means_no_schedule(self):
|
|
self.assertEqual(format_duration(None), "—")
|
|
|
|
def test_seconds_below_a_minute(self):
|
|
self.assertEqual(format_duration(0), "0s")
|
|
self.assertEqual(format_duration(45), "45s")
|
|
self.assertEqual(format_duration(59.9), "59s")
|
|
|
|
def test_minutes_and_seconds_are_zero_padded(self):
|
|
self.assertEqual(format_duration(60), "1:00")
|
|
self.assertEqual(format_duration(2400), "40:00")
|
|
self.assertEqual(format_duration(2399.6), "39:59")
|
|
self.assertEqual(format_duration(65), "1:05")
|
|
|
|
def test_hours_are_shown_when_needed(self):
|
|
self.assertEqual(format_duration(3600), "1:00:00")
|
|
self.assertEqual(format_duration(86400), "24:00:00")
|
|
|
|
def test_negative_is_clamped_to_zero(self):
|
|
self.assertEqual(format_duration(-5), "0s")
|
|
|
|
|
|
class ClampTestCase(unittest.TestCase):
|
|
def test_clamp_keeps_value_inside_range(self):
|
|
self.assertEqual(clamp(5, 0, 10), 5)
|
|
self.assertEqual(clamp(-1, 0, 10), 0)
|
|
self.assertEqual(clamp(99, 0, 10), 10)
|
|
self.assertEqual(clamp(3.5, 0, 10), 3.5)
|
|
|
|
|
|
class PlotTestCase(unittest.TestCase):
|
|
def test_empty_plot_defaults(self):
|
|
plot = Plot(0)
|
|
self.assertTrue(plot.is_empty)
|
|
self.assertFalse(plot.is_ripe)
|
|
self.assertEqual(plot.days_left, 0)
|
|
self.assertEqual(plot.grow_days, 0)
|
|
self.assertEqual(plot.stage, "空地")
|
|
|
|
def test_growth_stages_follow_config_thresholds(self):
|
|
plot = Plot(0)
|
|
plot.crop = "corn" # 6 天成熟
|
|
plot.growth = 0
|
|
self.assertEqual(plot.stage, "已播种")
|
|
plot.growth = 2 # 33%
|
|
self.assertEqual(plot.stage, "幼苗")
|
|
plot.growth = 4 # 66%
|
|
self.assertEqual(plot.stage, "生长中")
|
|
plot.growth = 6
|
|
self.assertEqual(plot.stage, "已成熟")
|
|
|
|
def test_ripeness_and_days_left(self):
|
|
plot = Plot(2)
|
|
plot.crop = "carrot" # 3 天
|
|
plot.growth = 1
|
|
self.assertFalse(plot.is_ripe)
|
|
self.assertEqual(plot.days_left, 2)
|
|
plot.growth = 3
|
|
self.assertTrue(plot.is_ripe)
|
|
self.assertEqual(plot.days_left, 0)
|
|
|
|
def test_to_dict_exposes_position_which_is_one_based(self):
|
|
data = Plot(3).to_dict()
|
|
self.assertEqual(data["index"], 3)
|
|
self.assertEqual(data["position"], 4)
|
|
|
|
def test_from_dict_drops_crop_names_that_no_longer_exist(self):
|
|
"""旧存档里可能留着已经删掉的作物，读档不能 KeyError。"""
|
|
plot = Plot.from_dict({"index": 0, "crop": "古老的神秘作物", "growth": 3})
|
|
self.assertIsNone(plot.crop)
|
|
self.assertTrue(plot.is_empty)
|
|
|
|
def test_from_dict_survives_missing_fields(self):
|
|
plot = Plot.from_dict({})
|
|
self.assertEqual(plot.index, 0)
|
|
self.assertEqual(plot.growth, 0.0)
|
|
self.assertEqual(plot.moisture, 0)
|
|
|
|
|
|
class AnimalTestCase(unittest.TestCase):
|
|
def test_info_comes_from_config(self):
|
|
animal = Animal("chicken")
|
|
self.assertEqual(animal.name, "鸡")
|
|
self.assertEqual(animal.product, "鸡蛋")
|
|
self.assertEqual(animal.cycle, config.ANIMALS["chicken"]["cycle"])
|
|
|
|
def test_days_to_product_counts_down(self):
|
|
animal = Animal("sheep") # cycle 2
|
|
self.assertEqual(animal.days_to_product, 2)
|
|
animal.progress = 2
|
|
self.assertEqual(animal.days_to_product, 0)
|
|
|
|
def test_sick_after_two_days_without_food(self):
|
|
animal = Animal("chicken")
|
|
animal.starve = 1
|
|
self.assertFalse(animal.is_sick)
|
|
animal.starve = 2
|
|
self.assertTrue(animal.is_sick)
|
|
|
|
def test_from_dict_rejects_unknown_kind(self):
|
|
self.assertIsNone(Animal.from_dict({"kind": "恐龙"}))
|
|
|
|
def test_from_dict_bumps_id_counter_so_new_animals_do_not_collide(self):
|
|
restored = Animal.from_dict({"kind": "cow", "id": 500})
|
|
assert restored is not None
|
|
self.assertEqual(restored.aid, 500)
|
|
fresh = Animal("chicken")
|
|
self.assertGreater(fresh.aid, 500)
|
|
|
|
|
|
class GameStateTestCase(unittest.TestCase):
|
|
def test_initial_state_matches_config(self):
|
|
state = GameState()
|
|
self.assertEqual(state.day, 1)
|
|
self.assertEqual(state.gold, config.START_GOLD)
|
|
self.assertEqual(state.ap, config.START_AP)
|
|
self.assertEqual(len(state.plots), config.BASE_PLOTS)
|
|
self.assertEqual(len(state.animals), 0)
|
|
|
|
def test_market_is_seeded_with_every_crop_and_animal_product(self):
|
|
state = GameState()
|
|
for key in config.CROPS:
|
|
self.assertIn(key, state.market)
|
|
for key in config.ANIMALS:
|
|
self.assertIn(f"animal:{key}", state.market)
|
|
|
|
def test_max_plots_grows_with_level_and_is_capped(self):
|
|
state = GameState()
|
|
self.assertEqual(state.max_plots, config.BASE_PLOTS)
|
|
state.level = 2
|
|
self.assertEqual(state.max_plots, config.BASE_PLOTS + config.PLOTS_PER_LEVEL)
|
|
state.level = 99
|
|
self.assertEqual(state.max_plots, config.MAX_PLOTS)
|
|
|
|
def test_next_plot_price_grows_and_returns_minus_one_at_the_cap(self):
|
|
state = GameState()
|
|
state.level = 2 # 初始 6 块地刚好占满 Lv.1 的名额
|
|
first = state.next_plot_price
|
|
self.assertEqual(first, config.PLOT_BASE_PRICE)
|
|
state.plots_bought = 1
|
|
self.assertGreater(state.next_plot_price, first)
|
|
state.level = 99
|
|
state.plots = [Plot(i) for i in range(config.MAX_PLOTS)]
|
|
self.assertEqual(state.next_plot_price, -1)
|
|
|
|
def test_net_worth_counts_gold_land_livestock_and_inventory(self):
|
|
state = GameState()
|
|
state.gold = 100
|
|
state.feed = 0
|
|
self.assertEqual(state.net_worth, 100.0)
|
|
state.feed = 10
|
|
self.assertEqual(state.net_worth, 100.0 + 10 * config.FEED_UNIT_PRICE)
|
|
|
|
def test_goal_progress(self):
|
|
state = GameState()
|
|
state.gold = 0
|
|
state.feed = 0
|
|
self.assertEqual(state.goal_index, -1)
|
|
self.assertEqual(state.next_goal["amount"], config.GOALS[0][0])
|
|
state.gold = config.GOALS[-1][0] + 1
|
|
self.assertEqual(state.goal_index, len(config.GOALS) - 1)
|
|
self.assertIsNone(state.next_goal)
|
|
|
|
def test_from_dict_clamps_hostile_values(self):
|
|
"""存档是外部输入：负数金币、越界行动点都要被夹回来。"""
|
|
state = GameState.from_dict({"gold": -500, "ap": 9999, "level": 0, "day": -3})
|
|
self.assertEqual(state.gold, 0)
|
|
self.assertEqual(state.ap, state.max_ap)
|
|
self.assertEqual(state.level, 1)
|
|
self.assertEqual(state.day, 1)
|
|
|
|
def test_from_dict_pads_missing_plots_and_reindexes(self):
|
|
state = GameState.from_dict({"plots": [{"index": 7, "crop": "wheat"}]})
|
|
self.assertEqual(len(state.plots), config.BASE_PLOTS)
|
|
self.assertEqual([p.index for p in state.plots], list(range(config.BASE_PLOTS)))
|
|
self.assertEqual(state.plots[0].crop, "wheat")
|
|
|
|
def test_from_dict_drops_extra_animals_beyond_capacity(self):
|
|
raw = {"animals": [{"kind": "chicken", "id": i + 1} for i in range(config.MAX_ANIMALS + 5)]}
|
|
state = GameState.from_dict(raw)
|
|
self.assertEqual(len(state.animals), config.MAX_ANIMALS)
|
|
|
|
def test_from_dict_ignores_unparsable_market_prices(self):
|
|
state = GameState.from_dict({"market": {"wheat": "not a number"}})
|
|
self.assertEqual(state.market["wheat"], config.CROPS["wheat"]["base_price"])
|
|
|
|
|
|
class StorageTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.dir = tempfile.mkdtemp(prefix="simplefarm-test-")
|
|
self.path = os.path.join(self.dir, "save.json")
|
|
|
|
def tearDown(self):
|
|
delete_save(self.path)
|
|
|
|
def write_raw(self, payload) -> str:
|
|
with open(self.path, "w", encoding="utf-8") as handle:
|
|
json.dump(payload, handle, ensure_ascii=False)
|
|
return self.path
|
|
|
|
def test_round_trip_keeps_the_important_fields(self):
|
|
engine = make_engine()
|
|
engine.plant(1, "wheat")
|
|
engine.buy_animal("chicken")
|
|
engine.state.day = 7
|
|
engine.state.exp = 42
|
|
|
|
save_state(engine.state, self.path)
|
|
loaded, message = load_state(self.path)
|
|
|
|
assert loaded is not None, message
|
|
self.assertEqual(loaded.day, 7)
|
|
self.assertEqual(loaded.exp, 42)
|
|
self.assertEqual(loaded.plots[0].crop, "wheat")
|
|
self.assertEqual([a.kind for a in loaded.animals], ["chicken"])
|
|
self.assertIn("第 7 天", message)
|
|
|
|
def test_saved_file_is_readable_json_with_version(self):
|
|
engine = make_engine()
|
|
save_state(engine.state, self.path)
|
|
with open(self.path, encoding="utf-8") as handle:
|
|
payload = json.load(handle)
|
|
self.assertEqual(payload["version"], SAVE_VERSION)
|
|
self.assertIsInstance(payload["state"], dict)
|
|
|
|
def test_missing_file_is_not_an_error(self):
|
|
state, message = load_state(self.path)
|
|
self.assertIsNone(state)
|
|
self.assertIn("未找到存档", message)
|
|
|
|
def test_broken_json_degrades_instead_of_raising(self):
|
|
with open(self.path, "w", encoding="utf-8") as handle:
|
|
handle.write("{ this is not json")
|
|
state, message = load_state(self.path)
|
|
self.assertIsNone(state)
|
|
self.assertIn("无法解析", message)
|
|
|
|
def test_wrong_version_is_refused(self):
|
|
self.write_raw({"version": SAVE_VERSION + 1, "state": {}})
|
|
state, message = load_state(self.path)
|
|
self.assertIsNone(state)
|
|
self.assertIn("版本不兼容", message)
|
|
|
|
def test_root_must_be_an_object(self):
|
|
self.write_raw([1, 2, 3])
|
|
state, message = load_state(self.path)
|
|
self.assertIsNone(state)
|
|
self.assertIn("根节点不是对象", message)
|
|
|
|
def test_state_field_is_required(self):
|
|
self.write_raw({"version": SAVE_VERSION})
|
|
state, message = load_state(self.path)
|
|
self.assertIsNone(state)
|
|
self.assertIn("缺少 state", message)
|
|
|
|
def test_unexpected_state_shape_degrades(self):
|
|
self.write_raw({"version": SAVE_VERSION, "state": {"plots": "nonsense", "animals": 7}})
|
|
state, message = load_state(self.path)
|
|
self.assertIsNone(state)
|
|
self.assertIn("存档内容异常", message)
|
|
|
|
def test_atomic_write_leaves_no_temp_file(self):
|
|
engine = make_engine()
|
|
save_state(engine.state, self.path)
|
|
self.assertFalse(os.path.exists(self.path + ".tmp"))
|
|
|
|
def test_delete_save_reports_truthfully(self):
|
|
save_state(make_engine().state, self.path)
|
|
self.assertTrue(delete_save(self.path))
|
|
self.assertFalse(os.path.exists(self.path))
|
|
|
|
|
|
class ActionTestCase(unittest.TestCase):
|
|
"""各玩法动作的成功路径与错误码。"""
|
|
|
|
def setUp(self):
|
|
self.engine = make_engine()
|
|
|
|
def assert_code(self, code: str, func, *args, **kwargs):
|
|
with self.assertRaises(GameError) as ctx:
|
|
func(*args, **kwargs)
|
|
self.assertEqual(ctx.exception.code, code)
|
|
self.assertEqual(ctx.exception.status, 400)
|
|
|
|
# ------------------------------------------------------------------ 播种
|
|
def test_plant_costs_gold_and_action_point(self):
|
|
gold, ap = self.engine.state.gold, self.engine.state.ap
|
|
result = self.engine.plant(1, "wheat")
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual(self.engine.state.gold, gold - config.CROPS["wheat"]["seed_cost"])
|
|
self.assertEqual(self.engine.state.ap, ap - config.AP_COST["plant"])
|
|
|
|
def test_plant_sets_initial_moisture_and_planted_day(self):
|
|
self.engine.state.day = 4
|
|
self.engine.plant(1, "wheat")
|
|
plot = self.engine.state.plots[0]
|
|
self.assertEqual(plot.moisture, config.PLANT_INITIAL_MOISTURE)
|
|
self.assertEqual(plot.planted_day, 4)
|
|
self.assertEqual(plot.growth, 0.0)
|
|
|
|
def test_plant_rejects_unknown_crop(self):
|
|
self.assert_code("unknown_crop", self.engine.plant, 1, "榴莲")
|
|
|
|
def test_plant_rejects_level_locked_crop(self):
|
|
self.assert_code("locked", self.engine.plant, 1, "strawberry")
|
|
|
|
def test_plant_rejects_busy_plot(self):
|
|
self.engine.plant(1, "wheat")
|
|
self.assert_code("plot_busy", self.engine.plant, 1, "carrot")
|
|
|
|
def test_plot_number_is_one_based_and_bounded(self):
|
|
self.assert_code("no_such_plot", self.engine.plant, 0, "wheat")
|
|
self.assert_code("no_such_plot", self.engine.plant, 99, "wheat")
|
|
|
|
def test_non_numeric_plot_number_is_a_param_error(self):
|
|
self.assert_code("bad_param", self.engine.plant, "第一块", "wheat")
|
|
|
|
# ------------------------------------------------------------------ 浇水
|
|
def test_water_fills_moisture_to_capacity(self):
|
|
self.engine.plant(1, "wheat")
|
|
self.engine.state.plots[0].moisture = 0
|
|
self.engine.water(1)
|
|
self.assertEqual(self.engine.state.plots[0].moisture, config.WATER_CAPACITY)
|
|
|
|
def test_water_needs_ap(self):
|
|
self.engine.plant(1, "wheat")
|
|
self.engine.state.ap = 0
|
|
self.assert_code("no_ap", self.engine.water, 1)
|
|
|
|
def test_water_rejects_full_and_empty_plots(self):
|
|
self.assert_code("plot_empty", self.engine.water, 1)
|
|
self.engine.plant(2, "wheat")
|
|
self.engine.state.plots[1].moisture = config.WATER_CAPACITY
|
|
self.assert_code("already_watered", self.engine.water, 2)
|
|
|
|
# ------------------------------------------------------------------ 收获
|
|
def test_harvest_pays_the_market_price(self):
|
|
self.engine.plant(1, "wheat")
|
|
self.engine.state.plots[0].growth = config.CROPS["wheat"]["grow_days"]
|
|
gold = self.engine.state.gold
|
|
with mock.patch("farm.engine.random.random", return_value=0.99):
|
|
result = self.engine.harvest(1)
|
|
self.assertEqual(result["quantity"], 1)
|
|
self.assertEqual(result["earned"], config.CROPS["wheat"]["base_price"])
|
|
self.assertEqual(self.engine.state.gold, gold + result["earned"])
|
|
self.assertTrue(self.engine.state.plots[0].is_empty)
|
|
|
|
def test_lucky_harvest_doubles_the_yield(self):
|
|
self.engine.plant(1, "wheat")
|
|
self.engine.state.plots[0].growth = 2
|
|
with mock.patch("farm.engine.random.random", return_value=0.0):
|
|
result = self.engine.harvest(1)
|
|
self.assertEqual(result["quantity"], 2)
|
|
self.assertEqual(result["earned"], config.CROPS["wheat"]["base_price"] * 2)
|
|
|
|
def test_harvest_before_ripe_is_refused(self):
|
|
self.engine.plant(1, "carrot")
|
|
self.assert_code("not_ripe", self.engine.harvest, 1)
|
|
|
|
def test_harvest_on_empty_plot_is_refused(self):
|
|
self.assert_code("plot_empty", self.engine.harvest, 3)
|
|
|
|
def test_harvesting_clears_the_plot_completely(self):
|
|
self.engine.plant(1, "wheat")
|
|
self.engine.state.plots[0].growth = 2
|
|
self.engine.harvest(1)
|
|
plot = self.engine.state.plots[0]
|
|
self.assertIsNone(plot.crop)
|
|
self.assertEqual(plot.growth, 0)
|
|
self.assertEqual(plot.moisture, 0)
|
|
|
|
# ------------------------------------------------------------------ 牲畜
|
|
def test_buy_animal_takes_gold_and_lands_in_the_pasture(self):
|
|
result = self.engine.buy_animal("chicken")
|
|
self.assertEqual(len(self.engine.state.animals), 1)
|
|
self.assertEqual(result["animal"]["kind"], "chicken")
|
|
|
|
def test_buy_animal_checks_level_and_money_and_capacity(self):
|
|
self.assert_code("unknown_animal", self.engine.buy_animal, "独角兽")
|
|
self.assert_code("locked", self.engine.buy_animal, "cow")
|
|
self.engine.state.gold = 0
|
|
self.assert_code("not_enough_gold", self.engine.buy_animal, "chicken")
|
|
|
|
def test_pasture_has_a_capacity_limit(self):
|
|
self.engine.state.level = 5
|
|
for _ in range(config.MAX_ANIMALS):
|
|
self.engine.buy_animal("chicken")
|
|
self.assert_code("pasture_full", self.engine.buy_animal, "chicken")
|
|
|
|
def test_feed_consumes_inventory(self):
|
|
self.engine.buy_animal("chicken")
|
|
aid = self.engine.state.animals[0].aid
|
|
feed_before = self.engine.state.feed
|
|
self.engine.feed(aid)
|
|
self.assertEqual(self.engine.state.feed, feed_before - 1)
|
|
self.assertTrue(self.engine.state.animals[0].fed)
|
|
|
|
def test_feeding_twice_a_day_is_refused(self):
|
|
self.engine.buy_animal("chicken")
|
|
aid = self.engine.state.animals[0].aid
|
|
self.engine.feed(aid)
|
|
self.assert_code("already_fed", self.engine.feed, aid)
|
|
|
|
def test_short_feed_is_bought_at_the_emergency_price(self):
|
|
self.engine.buy_animal("chicken")
|
|
self.engine.state.feed = 0
|
|
gold = self.engine.state.gold
|
|
self.engine.feed(self.engine.state.animals[0].aid)
|
|
self.assertEqual(gold - self.engine.state.gold, config.FEED_EMERGENCY_PRICE)
|
|
self.assertEqual(self.engine.state.feed, 0)
|
|
|
|
def test_overfeeding_is_trimmed_to_the_actual_need(self):
|
|
self.engine.state.level = 3
|
|
self.engine.buy_animal("cow") # 每天 2 份
|
|
self.engine.state.feed = 10
|
|
self.engine.feed(self.engine.state.animals[0].aid, quantity=99)
|
|
self.assertEqual(self.engine.state.feed, 8)
|
|
|
|
def test_feed_rejects_bad_quantity(self):
|
|
self.engine.buy_animal("chicken")
|
|
aid = self.engine.state.animals[0].aid
|
|
self.assert_code("bad_param", self.engine.feed, aid, 0)
|
|
self.assert_code("bad_param", self.engine.feed, aid, "many")
|
|
|
|
def test_collect_pays_for_every_pending_product(self):
|
|
self.engine.buy_animal("chicken")
|
|
animal = self.engine.state.animals[0]
|
|
animal.pending = 3
|
|
gold = self.engine.state.gold
|
|
result = self.engine.collect(animal.aid)
|
|
self.assertEqual(result["quantity"], 3)
|
|
self.assertEqual(result["earned"], config.ANIMALS["chicken"]["base_price"] * 3)
|
|
self.assertEqual(self.engine.state.gold, gold + result["earned"])
|
|
self.assertEqual(animal.pending, 0)
|
|
|
|
def test_collect_without_output_is_refused(self):
|
|
self.engine.buy_animal("chicken")
|
|
self.assert_code(
|
|
"nothing_to_collect", self.engine.collect, self.engine.state.animals[0].aid
|
|
)
|
|
|
|
def test_unknown_animal_id(self):
|
|
self.assert_code("no_such_animal", self.engine.feed, 999)
|
|
|
|
# ------------------------------------------------------------------ 投资
|
|
def test_buy_plot_appends_land_and_raises_the_next_price(self):
|
|
self.engine.state.level = 2 # Lv.1 的 6 块地已经到手，升级后才有地可买
|
|
before = len(self.engine.state.plots)
|
|
price = self.engine.state.next_plot_price
|
|
gold = self.engine.state.gold
|
|
self.engine.buy_plot()
|
|
self.assertEqual(len(self.engine.state.plots), before + 1)
|
|
self.assertEqual(gold - self.engine.state.gold, price)
|
|
self.assertEqual(self.engine.state.plots[-1].index, before)
|
|
|
|
def test_buy_plot_respects_the_level_cap(self):
|
|
self.engine.state.level = 99
|
|
self.engine.state.gold = 10**9
|
|
while self.engine.state.next_plot_price > 0:
|
|
self.engine.state.ap = 10
|
|
self.engine.buy_plot()
|
|
self.assertEqual(len(self.engine.state.plots), config.MAX_PLOTS)
|
|
self.assert_code("plot_limit", self.engine.buy_plot)
|
|
|
|
def test_buy_feed_adds_inventory(self):
|
|
before = self.engine.state.feed
|
|
result = self.engine.buy_feed(20)
|
|
self.assertEqual(self.engine.state.feed, before + 20)
|
|
self.assertEqual(result["feed"], before + 20)
|
|
|
|
def test_buy_feed_bounds(self):
|
|
self.assert_code("bad_param", self.engine.buy_feed, 0)
|
|
self.assert_code("bad_param", self.engine.buy_feed, "一袋")
|
|
self.assert_code("too_many", self.engine.buy_feed, config.MAX_FEED_PURCHASE + 1)
|
|
|
|
def test_buy_feed_rejects_negative_amount_without_crediting_gold(self):
|
|
"""负数是"倒收钱"的经典漏洞，必须被金额校验挡住。"""
|
|
gold = self.engine.state.gold
|
|
with self.assertRaises(GameError):
|
|
self.engine.buy_feed(-5)
|
|
self.assertEqual(self.engine.state.gold, gold)
|
|
|
|
# ------------------------------------------------------------------ 杂项
|
|
def test_pause_toggles_and_can_be_explicit(self):
|
|
self.assertFalse(self.engine.paused)
|
|
self.assertTrue(self.engine.pause()["paused"], "无参调用应当反转")
|
|
self.assertFalse(self.engine.pause()["paused"], "再调一次就翻回来")
|
|
self.assertFalse(self.engine.pause(False)["paused"])
|
|
self.assertTrue(self.engine.pause(True)["paused"])
|
|
|
|
def test_reset_clears_progress_but_keeps_the_clock_setting(self):
|
|
engine = FarmEngine(tick_seconds=120)
|
|
engine.state.gold = 9999
|
|
engine.pause(True)
|
|
engine.reset()
|
|
self.assertEqual(engine.state.gold, config.START_GOLD)
|
|
self.assertEqual(engine.state.day, 1)
|
|
self.assertFalse(engine.paused)
|
|
self.assertEqual(engine.tick_seconds, 120)
|
|
|
|
def test_snapshot_carries_clock_fields(self):
|
|
snap = self.engine.snapshot()
|
|
self.assertIn("tick_seconds", snap)
|
|
self.assertIn("paused", snap)
|
|
self.assertIn("seconds_to_tick", snap)
|
|
|
|
def test_seconds_to_tick_is_none_without_a_scheduler(self):
|
|
"""没有调度器（单机暂停 / 尚未启动）时倒计时必须是 None，不是 0。"""
|
|
self.assertIsNone(self.engine.seconds_to_tick)
|
|
|
|
def test_seconds_to_tick_counts_down(self):
|
|
|
|
self.engine.next_tick_at = time.time() + 30
|
|
value = self.engine.seconds_to_tick
|
|
assert value is not None
|
|
self.assertGreater(value, 25)
|
|
self.assertLessEqual(value, 30)
|
|
|
|
|
|
class GrowthTestCase(unittest.TestCase):
|
|
"""水、生长、成熟这条链——放置游戏的核心节奏。"""
|
|
|
|
def setUp(self):
|
|
self.engine = make_engine()
|
|
|
|
def test_crops_grow_only_while_watered(self):
|
|
self.engine.plant(1, "carrot") # 3 天，刚好够观察缺水停摆
|
|
self.assertEqual(self.engine.state.plots[0].moisture, config.PLANT_INITIAL_MOISTURE)
|
|
|
|
self.engine.advance_day()
|
|
self.assertEqual(self.engine.state.plots[0].growth, 1)
|
|
|
|
self.engine.advance_day()
|
|
self.assertEqual(self.engine.state.plots[0].moisture, 0)
|
|
self.assertEqual(self.engine.state.plots[0].growth, 2)
|
|
|
|
events = self.engine.advance_day()
|
|
self.assertEqual(self.engine.state.plots[0].growth, 2, "没水就不该再长")
|
|
self.assertTrue(any("缺水" in text for text in events))
|
|
|
|
def test_watering_resumes_growth(self):
|
|
self.engine.plant(1, "carrot")
|
|
self.engine.advance_day()
|
|
self.engine.advance_day()
|
|
self.engine.advance_day() # 此刻缺水停摆
|
|
self.assertEqual(self.engine.state.plots[0].growth, 2, "断水期间不该生长")
|
|
self.engine.water(1) # 浇到 3 点水
|
|
self.engine.advance_day()
|
|
self.assertEqual(self.engine.state.plots[0].growth, 3, "浇水后又开始长了")
|
|
self.assertTrue(self.engine.state.plots[0].is_ripe)
|
|
|
|
def test_crop_reportedly_ripens_on_the_last_day(self):
|
|
self.engine.plant(1, "wheat") # 2 天
|
|
self.engine.advance_day()
|
|
self.engine.advance_day()
|
|
plot = self.engine.state.plots[0]
|
|
self.assertTrue(plot.is_ripe)
|
|
self.assertEqual(plot.growth, plot.grow_days)
|
|
|
|
def test_growth_never_exceeds_grow_days(self):
|
|
self.engine.plant(1, "wheat")
|
|
for _ in range(10):
|
|
self.engine.state.plots[0].moisture = config.WATER_CAPACITY
|
|
self.engine.advance_day()
|
|
plot = self.engine.state.plots[0]
|
|
self.assertEqual(plot.growth, plot.grow_days)
|
|
|
|
|
|
class AdvanceDayTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.engine = make_engine()
|
|
|
|
def test_day_increments_and_action_points_refill(self):
|
|
self.engine.state.ap = 0
|
|
self.engine.advance_day()
|
|
self.assertEqual(self.engine.state.day, 2)
|
|
self.assertEqual(self.engine.state.ap, config.START_AP)
|
|
|
|
def test_fed_animals_make_progress_and_get_hungry_again(self):
|
|
self.engine.buy_animal("chicken") # cycle 1
|
|
animal = self.engine.state.animals[0]
|
|
self.engine.feed(animal.aid)
|
|
self.engine.advance_day()
|
|
self.assertFalse(animal.fed, "新的一天要重新喂")
|
|
self.assertEqual(animal.pending, 1, "鸡每天下蛋")
|
|
|
|
def test_animal_dies_after_three_hungry_days(self):
|
|
self.engine.buy_animal("chicken")
|
|
animal = self.engine.state.animals[0]
|
|
aid = animal.aid
|
|
|
|
self.engine.advance_day()
|
|
self.assertEqual(animal.starve, 1)
|
|
events = self.engine.advance_day()
|
|
self.assertEqual(animal.starve, 2)
|
|
self.assertTrue(any("两天没吃东西" in text for text in events))
|
|
|
|
events = self.engine.advance_day()
|
|
self.assertEqual(len(self.engine.state.animals), 0)
|
|
self.assertTrue(any("离你而去" in text for text in events))
|
|
with self.assertRaises(GameError):
|
|
self.engine.feed(aid)
|
|
|
|
def test_feeding_resets_the_hunger_counter(self):
|
|
self.engine.buy_animal("chicken")
|
|
animal = self.engine.state.animals[0]
|
|
self.engine.advance_day()
|
|
self.engine.advance_day()
|
|
self.assertEqual(animal.starve, 2)
|
|
self.engine.state.feed = 5
|
|
self.engine.feed(animal.aid)
|
|
self.assertEqual(animal.starve, 0)
|
|
self.assertFalse(animal.is_sick)
|
|
|
|
def test_market_prices_stay_within_the_configured_band(self):
|
|
for _ in range(30):
|
|
self.engine.advance_day()
|
|
for key, info in config.CROPS.items():
|
|
base = info["base_price"]
|
|
price = self.engine.state.market[key]
|
|
self.assertGreaterEqual(price, round(base * 0.55, 2) - 0.01)
|
|
self.assertLessEqual(price, round(base * 1.55, 2) + 0.01)
|
|
for key, info in config.ANIMALS.items():
|
|
base = info["base_price"]
|
|
price = self.engine.state.market[f"animal:{key}"]
|
|
self.assertGreaterEqual(price, round(base * 0.55, 2) - 0.01)
|
|
self.assertLessEqual(price, round(base * 1.55, 2) + 0.01)
|
|
|
|
def test_progress_message_mentions_the_new_day(self):
|
|
self.engine.advance_day(reason="测试")
|
|
texts = [entry["text"] for entry in self.engine.state.log]
|
|
self.assertTrue(any("第 2 天" in text for text in texts))
|
|
|
|
|
|
class LevelUpTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.engine = make_engine()
|
|
|
|
def test_exp_accumulates_and_triggers_level_up(self):
|
|
needed = self.engine.state.exp_needed
|
|
self.engine.state.exp = needed - 1
|
|
self.engine.plant(1, "wheat")
|
|
self.engine.state.plots[0].growth = 2
|
|
self.engine.harvest(1)
|
|
|
|
self.assertEqual(self.engine.state.level, 2)
|
|
self.assertEqual(self.engine.state.exp, config.CROPS["wheat"]["exp"] - 1)
|
|
self.assertGreater(self.engine.state.gold, 2000 + config.CROPS["wheat"]["base_price"])
|
|
|
|
def test_big_exp_award_can_level_up_several_times(self):
|
|
self.engine.state.exp = 0
|
|
messages = self.engine._gain_exp(10**5)
|
|
self.assertGreater(len(messages), 2)
|
|
self.assertGreater(self.engine.state.level, 3)
|
|
|
|
def test_new_crop_unlocks_with_level(self):
|
|
with self.assertRaises(GameError):
|
|
self.engine.plant(1, "tomato")
|
|
self.engine.state.level = 2
|
|
self.engine.plant(1, "tomato")
|
|
self.assertEqual(self.engine.state.plots[0].crop, "tomato")
|
|
|
|
|
|
class AvailableActionsTestCase(unittest.TestCase):
|
|
"""给 AI 看的"当前合法动作"列表必须是**真的合法**。"""
|
|
|
|
def setUp(self):
|
|
self.engine = make_engine()
|
|
|
|
def names(self) -> set[str]:
|
|
return {item["action"] for item in self.engine.available_actions()}
|
|
|
|
def test_every_listed_action_can_actually_run(self):
|
|
"""这张表存在的意义：AI 照抄 params 就一定成功。
|
|
|
|
牲畜类动作的 animal_id 只对产生它的那个引擎有意义，
|
|
所以它们在别的用例里单独验证。
|
|
"""
|
|
actions = self.engine.available_actions()
|
|
self.assertTrue(actions)
|
|
for item in actions:
|
|
name = item["action"]
|
|
params = item.get("params") or {}
|
|
if "animal_id" in params:
|
|
continue
|
|
with self.subTest(action=name):
|
|
fresh = make_engine()
|
|
fresh.state.level = 5 # 解锁全部作物与牲畜
|
|
fresh.state.feed = 50
|
|
if name == "water":
|
|
plot = fresh.state.plots[params["plot"] - 1]
|
|
plot.crop, plot.moisture = "wheat", 0
|
|
elif name == "harvest":
|
|
plot = fresh.state.plots[params["plot"] - 1]
|
|
plot.crop, plot.growth = "wheat", 2
|
|
result = run_action(fresh, name, dict(params))
|
|
self.assertTrue(result["ok"])
|
|
|
|
def test_locked_crops_are_not_offered(self):
|
|
keys = {
|
|
item["params"].get("crop")
|
|
for item in self.engine.available_actions()
|
|
if item["action"] == "plant"
|
|
}
|
|
self.assertIn("wheat", keys)
|
|
self.assertNotIn("strawberry", keys, "Lv.5 的作物不该出现在 Lv.1")
|
|
|
|
def test_ripe_crop_offers_harvest_instead_of_water(self):
|
|
self.engine.plant(1, "wheat")
|
|
self.engine.state.plots[0].growth = 2
|
|
for_plot_1 = [
|
|
item["action"]
|
|
for item in self.engine.available_actions()
|
|
if item["params"].get("plot") == 1
|
|
]
|
|
self.assertIn("harvest", for_plot_1)
|
|
self.assertNotIn("water", for_plot_1)
|
|
|
|
def test_dry_plot_offers_water(self):
|
|
self.engine.plant(1, "carrot")
|
|
self.engine.state.plots[0].moisture = 0
|
|
self.assertIn("water", self.names())
|
|
|
|
def test_poor_player_is_not_offered_things_they_cannot_pay_for(self):
|
|
self.engine.state.gold = 0
|
|
self.assertNotIn("buy_plot", self.names())
|
|
self.assertNotIn("buy_animal", self.names())
|
|
self.assertNotIn("plant", self.names())
|
|
|
|
def test_feed_is_offered_only_while_inventory_is_low(self):
|
|
self.assertEqual(self.engine.state.feed, config.START_FEED)
|
|
self.assertNotIn("buy_feed", self.names(), "默认 5 份不算低")
|
|
self.engine.state.feed = 1
|
|
self.assertIn("buy_feed", self.names())
|
|
|
|
def test_hungry_animal_offers_feed_and_pending_offers_collect(self):
|
|
self.engine.buy_animal("chicken")
|
|
animal = self.engine.state.animals[0]
|
|
self.assertIn("feed", self.names())
|
|
animal.fed = True
|
|
animal.pending = 1
|
|
actions = [
|
|
item["action"]
|
|
for item in self.engine.available_actions()
|
|
if item["params"].get("animal_id") == animal.aid
|
|
]
|
|
self.assertIn("collect", actions)
|
|
self.assertNotIn("feed", actions)
|
|
|
|
def test_action_point_shortage_hides_action_point_actions(self):
|
|
self.engine.state.ap = 0
|
|
names = self.names()
|
|
self.assertNotIn("plant", names)
|
|
self.assertNotIn("water", names)
|
|
self.assertNotIn("feed", names)
|
|
|
|
def test_actions_explain_themselves(self):
|
|
for item in self.engine.available_actions():
|
|
with self.subTest(action=item["action"]):
|
|
self.assertTrue(item["reason"])
|
|
self.assertIn("gold", item["cost"])
|
|
self.assertIn("ap", item["cost"])
|
|
|
|
|
|
class LogTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.engine = make_engine()
|
|
|
|
def test_log_entries_carry_increasing_sequence_numbers(self):
|
|
self.engine.add_log("第一件事")
|
|
self.engine.add_log("第二件事")
|
|
seqs = [entry["seq"] for entry in self.engine.state.log]
|
|
self.assertEqual(seqs, sorted(seqs))
|
|
self.assertEqual(len(set(seqs)), len(seqs))
|
|
|
|
def test_recent_log_can_filter_by_sequence(self):
|
|
self.engine.add_log("老消息")
|
|
mark = self.engine.state.log[-1]["seq"]
|
|
self.engine.add_log("新消息")
|
|
entries = self.engine.recent_log(since=mark)
|
|
self.assertEqual([entry["text"] for entry in entries], ["新消息"])
|
|
|
|
def test_log_is_trimmed_to_the_configured_maximum(self):
|
|
for i in range(config.MAX_LOG_ENTRIES + 25):
|
|
self.engine.add_log(f"第 {i} 条")
|
|
self.assertEqual(len(self.engine.state.log), config.MAX_LOG_ENTRIES)
|
|
|
|
def test_listener_errors_do_not_break_the_game(self):
|
|
"""界面回调抛异常不能把玩法带走。"""
|
|
self.engine.subscribe(lambda: (_ for _ in ()).throw(RuntimeError("界面挂了")))
|
|
self.assertTrue(self.engine.plant(1, "wheat")["ok"])
|
|
|
|
|
|
class ClockTestCase(unittest.TestCase):
|
|
"""tick 配置、纯函数决策、以及单机调度线程。"""
|
|
|
|
def test_default_day_length_is_forty_minutes(self):
|
|
self.assertEqual(config.DEFAULT_TICK_SECONDS, 2400.0)
|
|
self.assertEqual(FarmEngine().tick_seconds, 2400.0)
|
|
|
|
def test_tick_is_clamped_on_both_ends(self):
|
|
self.assertEqual(FarmEngine(tick_seconds=0).tick_seconds, config.MIN_TICK_SECONDS)
|
|
self.assertEqual(FarmEngine(tick_seconds=10**9).tick_seconds, config.MAX_TICK_SECONDS)
|
|
self.assertEqual(FarmEngine(tick_seconds=999).tick_seconds, 999.0)
|
|
|
|
def test_the_old_ten_minute_ceiling_is_gone(self):
|
|
"""历史坑：上限曾是 600 秒，会把 40 分钟悄悄夹回 10 分钟。"""
|
|
self.assertGreater(config.MAX_TICK_SECONDS, config.DEFAULT_TICK_SECONDS)
|
|
|
|
def test_tick_decision_paused_defers(self):
|
|
advance, next_at = tick_decision(True, 100.0, 60.0, 500.0)
|
|
self.assertFalse(advance)
|
|
self.assertEqual(next_at, 560.0, "暂停时要顺延，保证恢复后有完整一轮")
|
|
|
|
def test_tick_decision_without_schedule_starts_counting(self):
|
|
advance, next_at = tick_decision(False, None, 60.0, 500.0)
|
|
self.assertFalse(advance)
|
|
self.assertEqual(next_at, 560.0)
|
|
|
|
def test_tick_decision_fires_when_due(self):
|
|
advance, next_at = tick_decision(False, 100.0, 60.0, 100.0)
|
|
self.assertTrue(advance)
|
|
self.assertEqual(next_at, 160.0)
|
|
|
|
def test_tick_decision_keeps_the_deadline_while_waiting(self):
|
|
advance, next_at = tick_decision(False, 100.0, 60.0, 40.0)
|
|
self.assertFalse(advance)
|
|
self.assertEqual(next_at, 100.0, "没到点不要重排，否则永远到不了")
|
|
|
|
def test_scheduler_sets_and_clears_the_deadline(self):
|
|
engine = make_engine()
|
|
scheduler = Scheduler(engine)
|
|
self.assertIsNone(engine.next_tick_at)
|
|
scheduler.start()
|
|
try:
|
|
self.assertIsNotNone(engine.next_tick_at)
|
|
finally:
|
|
scheduler.stop()
|
|
self.assertIsNone(engine.next_tick_at)
|
|
|
|
def test_scheduler_advances_days_on_its_own(self):
|
|
"""真的让它跑一小会儿：把 tick 手动压到 50ms。"""
|
|
engine = make_engine()
|
|
engine.tick_seconds = 0.05
|
|
scheduler = Scheduler(engine)
|
|
scheduler.start()
|
|
try:
|
|
deadline = time.time() + 3.0
|
|
while engine.state.day < 2 and time.time() < deadline:
|
|
time.sleep(0.05)
|
|
finally:
|
|
scheduler.stop()
|
|
self.assertGreaterEqual(engine.state.day, 2, "时间应当自己往前走")
|
|
|
|
def test_disabled_scheduler_does_nothing(self):
|
|
engine = make_engine()
|
|
scheduler = Scheduler(engine, enabled=False)
|
|
scheduler.start()
|
|
self.assertIsNone(engine.next_tick_at)
|
|
self.assertIsNone(scheduler._thread)
|
|
|
|
def test_paused_engine_does_not_advance(self):
|
|
engine = make_engine()
|
|
engine.tick_seconds = 0.05
|
|
engine.pause(True)
|
|
scheduler = Scheduler(engine)
|
|
scheduler.start()
|
|
try:
|
|
time.sleep(0.4)
|
|
finally:
|
|
scheduler.stop()
|
|
self.assertEqual(engine.state.day, 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|