snowy-yang
cf24cb43f1
动手前先修掉了两处「死机制」，新增的三个系统各自激活一个： * 天气：小麦 2 天成熟、初始水分 2、每天只掉 1 点，等于永远不需要浇水。 现在水分消耗由天气决定（晴 1 / 多云 0 / 干旱 2，雨天反补 3），而 `state.forecast` 把明天的天气提前告诉你——「今晚浇不浇」成了能算的决策。 5 种天气、权重合计 100、同种不连出；暴风雨会毁未熟作物，温室免疫。 * 谷仓：市场波动原本纯装饰（收获当场按市价变现，没有择时空间）。现在收获与 收取只进 `state.inventory`，靠新增的 `sell` 换金币，价格波动才真正参与决策。 仓满时 `_store()` 抛 `granary_full` 且不改状态，作物原样留在地里。 * 建筑：新增 `build`，买的是「以后不用天天手动做」——水井自动浇干地、 喂食器自动喂食（只吃库存饲料，绝不自动花钱）、温室、谷仓扩容。 每天在 `advance_day` 里先于生长执行，同类越造越贵（×1.6）。 存档升到 `SAVE_VERSION=2`，配 `MIN_SAVE_VERSION=1` 向后兼容，旧档缺的新字段 一律走默认值，不作废玩家手里的档。`advance_day(weather=...)` 是留给测试的 确定性入口。 顺带修两个真 bug： * `daemon.py` 三处 `makedirs(os.path.dirname(path))`：`--pid-file pid.txt` 这种只给文件名的用法会让 dirname 变成空串，启动阶段直接崩。抽出 `ensure_parent_dir()` 兜底。 * `RemoteFarm.snapshot()` 是本地缓存，非实时客户端轮询会一直读到旧数据， 看起来像服务器卡死。补 docstring 指明要自己调 `refresh()`。 质量门禁：ruff format/check 全绿、pyright 0 错误、503 项测试 OK（skip 35）、 bench 无回归（brief 19.5µs / snapshot 48.5µs / POST full=1 400µs）。 另起真实服务器（--tick 3 + 预置存档）跑了 38 项端到端验收，全过。
411 lines
15 KiB
Python
411 lines
15 KiB
Python
"""规则机器人测试：打分策略、单步执行、以及"没事可做时怎么办"。
|
|
|
|
机器人是「服务器把状态和合法动作摊开后，外面是谁在决策都不重要」这个设计的
|
|
活证明，所以它走的每一步都应当是**普通客户端**能走的。
|
|
|
|
全程使用鸭子类型的假客户端，不碰网络，也不依赖服务器。
|
|
|
|
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):
|
|
# AP 用完时，服务端的 available_actions 只会剩下 0 消耗的动作
|
|
# （卖货/建造/收获/收取），这里模拟"连这些都没有"的极端情况：
|
|
# 机器人应当直接收工，而不是空转或乱发请求。
|
|
client = FakeClient(
|
|
actions=[],
|
|
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_free_actions_still_run_without_action_points(self):
|
|
# 新设计：收获不再直接给钱，得靠 sell 把谷仓里的货变现；而 sell/build
|
|
# 是 0 消耗动作，AP=0 时也必须能执行，否则一整天只能干等着。
|
|
client = FakeClient(
|
|
actions=[{"action": "sell", "params": {}}],
|
|
state={**BASE_STATE, "ap": 0},
|
|
catalog=CATALOG,
|
|
)
|
|
bot = make_bot(client)
|
|
summary = bot.run(rounds=1)
|
|
self.assertEqual(summary["actions"], 1)
|
|
self.assertEqual(client.calls, [("sell", {})])
|
|
|
|
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)
|