snowy-yang
a7725f8812
在引入 uv 依赖管理、ruff 静态检查与格式化、pyright 类型检查之前， 先把当前可正常运行的状态固定下来，作为后续大规模自动改动的回滚点。 此时状态：97 项测试全绿，HTTP + WebSocket 多牧场服务器可用。
459 lines
18 KiB
Python
459 lines
18 KiB
Python
"""核心引擎的单元测试。
|
|
|
|
运行方式：
|
|
python -m unittest discover -s tests -v
|
|
或：
|
|
python tests/test_engine.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import random
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from farm import config # noqa: E402
|
|
from farm.engine import FarmEngine, GameError # noqa: E402
|
|
from farm.models import GameState, Plot # noqa: E402
|
|
from farm.storage import load_state, save_state # noqa: E402
|
|
|
|
|
|
class EngineTestCase(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
random.seed(20260918)
|
|
self.engine = FarmEngine(tick_seconds=5)
|
|
# 存档相关测试会写文件，统一放到临时目录
|
|
self.tmp = os.path.join(os.path.dirname(os.path.abspath(__file__)), "_tmp")
|
|
os.makedirs(self.tmp, exist_ok=True)
|
|
|
|
# ============================================================ 初始状态
|
|
def test_initial_state(self) -> None:
|
|
state = self.engine.state
|
|
self.assertEqual(state.day, 1)
|
|
self.assertEqual(state.gold, config.START_GOLD)
|
|
self.assertEqual(state.level, 1)
|
|
self.assertEqual(state.ap, config.START_AP)
|
|
self.assertEqual(len(state.plots), config.BASE_PLOTS)
|
|
self.assertEqual(state.animals, [])
|
|
self.assertEqual(state.exp_needed, config.exp_to_next(1))
|
|
|
|
# ============================================================ 种植
|
|
def test_plant_costs_gold_and_ap(self) -> None:
|
|
before_gold = self.engine.state.gold
|
|
before_ap = self.engine.state.ap
|
|
self.engine.plant(1, "wheat")
|
|
plot = self.engine.state.plots[0]
|
|
self.assertEqual(plot.crop, "wheat")
|
|
self.assertEqual(self.engine.state.gold, before_gold - config.CROPS["wheat"]["seed_cost"])
|
|
self.assertEqual(self.engine.state.ap, before_ap - config.AP_COST["plant"])
|
|
self.assertEqual(plot.moisture, config.PLANT_INITIAL_MOISTURE)
|
|
|
|
def test_plant_twice_on_same_plot_fails(self) -> None:
|
|
self.engine.plant(1, "wheat")
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.plant(1, "carrot")
|
|
self.assertEqual(ctx.exception.code, "plot_busy")
|
|
|
|
def test_plant_out_of_range_plot(self) -> None:
|
|
for bad in (0, -1, 99, "abc", None):
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.plant(bad, "wheat")
|
|
self.assertIn(ctx.exception.code, ("no_such_plot", "bad_param"))
|
|
|
|
def test_plant_unknown_crop(self) -> None:
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.plant(1, "banana")
|
|
self.assertEqual(ctx.exception.code, "unknown_crop")
|
|
with self.assertRaises(GameError) as ctx2:
|
|
self.engine.plant(1, 123)
|
|
self.assertEqual(ctx2.exception.code, "unknown_crop")
|
|
|
|
def test_plant_locked_crop(self) -> None:
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.plant(1, "pumpkin")
|
|
self.assertEqual(ctx.exception.code, "locked")
|
|
|
|
def test_plant_without_gold(self) -> None:
|
|
self.engine.state.gold = 1
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.plant(1, "wheat")
|
|
self.assertEqual(ctx.exception.code, "not_enough_gold")
|
|
|
|
def test_plant_without_ap(self) -> None:
|
|
self.engine.state.ap = 1
|
|
self.engine.plant(1, "wheat")
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.plant(2, "wheat")
|
|
self.assertEqual(ctx.exception.code, "no_ap")
|
|
|
|
# ============================================================ 生长与水分
|
|
def test_crop_needs_water_to_grow(self) -> None:
|
|
self.engine.plant(1, "wheat")
|
|
plot = self.engine.state.plots[0]
|
|
# 播种自带 2 点水分，推进 2 天刚好成熟
|
|
self.engine.advance_day()
|
|
self.engine.advance_day()
|
|
self.assertTrue(plot.is_ripe)
|
|
|
|
def test_crop_stops_growing_without_water(self) -> None:
|
|
self.engine.state.level = 5
|
|
self.engine.plant(1, "corn") # 需要 6 天
|
|
plot = self.engine.state.plots[0]
|
|
for _ in range(5):
|
|
self.engine.advance_day()
|
|
# 只有 2 点初始水分，所以最多长 2 天
|
|
self.assertEqual(plot.growth, 2)
|
|
self.assertEqual(plot.moisture, 0)
|
|
self.assertFalse(plot.is_ripe)
|
|
|
|
def test_water_restores_moisture(self) -> None:
|
|
self.engine.state.level = 5
|
|
self.engine.plant(1, "corn")
|
|
plot = self.engine.state.plots[0]
|
|
self.engine.state.ap = 10
|
|
self.engine.water(1)
|
|
self.assertEqual(plot.moisture, config.WATER_CAPACITY)
|
|
|
|
def test_water_empty_plot_fails(self) -> None:
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.water(1)
|
|
self.assertEqual(ctx.exception.code, "plot_empty")
|
|
|
|
def test_water_when_full_fails(self) -> None:
|
|
self.engine.plant(1, "wheat")
|
|
self.engine.water(1)
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.water(1)
|
|
self.assertEqual(ctx.exception.code, "already_watered")
|
|
|
|
# ============================================================ 收获
|
|
def test_harvest_not_ripe(self) -> None:
|
|
self.engine.plant(1, "wheat")
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.harvest(1)
|
|
self.assertEqual(ctx.exception.code, "not_ripe")
|
|
|
|
def test_harvest_empty_plot(self) -> None:
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.harvest(1)
|
|
self.assertEqual(ctx.exception.code, "plot_empty")
|
|
|
|
def test_harvest_gives_gold_and_exp(self) -> None:
|
|
self.engine.plant(1, "wheat")
|
|
self.engine.advance_day()
|
|
self.engine.advance_day()
|
|
gold = self.engine.state.gold
|
|
result = self.engine.harvest(1)
|
|
self.assertTrue(result["ok"])
|
|
self.assertGreater(result["earned"], 0)
|
|
self.assertEqual(self.engine.state.gold, gold + result["earned"])
|
|
self.assertEqual(self.engine.state.exp, config.CROPS["wheat"]["exp"])
|
|
self.assertIsNone(self.engine.state.plots[0].crop)
|
|
|
|
def test_exp_accumulates(self) -> None:
|
|
self.engine.plant(1, "wheat")
|
|
self.engine.advance_day()
|
|
self.engine.advance_day()
|
|
self.engine.harvest(1)
|
|
self.assertEqual(self.engine.state.exp, config.CROPS["wheat"]["exp"])
|
|
|
|
# ============================================================ 等级与土地
|
|
def test_level_up_unlocks_and_rewards(self) -> None:
|
|
state = self.engine.state
|
|
state.exp = state.exp_needed - 1
|
|
# 种一茬小麦拿经验，触发升级
|
|
self.engine.plant(1, "wheat")
|
|
self.engine.advance_day()
|
|
self.engine.advance_day()
|
|
gold_before = state.gold
|
|
self.engine.harvest(1)
|
|
self.assertEqual(state.level, 2)
|
|
self.assertGreaterEqual(state.gold, gold_before + config.level_up_bonus(2))
|
|
self.assertEqual(state.max_plots, config.BASE_PLOTS + config.PLOTS_PER_LEVEL)
|
|
|
|
def test_buy_plot_and_limit(self) -> None:
|
|
self.engine.state.gold = 99999
|
|
self.engine.state.level = 10
|
|
before = len(self.engine.state.plots)
|
|
self.engine.buy_plot()
|
|
self.assertEqual(len(self.engine.state.plots), before + 1)
|
|
|
|
while len(self.engine.state.plots) < config.MAX_PLOTS:
|
|
self.engine.buy_plot()
|
|
self.assertEqual(len(self.engine.state.plots), config.MAX_PLOTS)
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.buy_plot()
|
|
self.assertEqual(ctx.exception.code, "plot_limit")
|
|
|
|
def test_buy_plot_without_gold(self) -> None:
|
|
self.engine.state.level = 5 # 先解锁土地上限，才能走到"钱不够"这一步
|
|
self.engine.state.gold = 0
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.buy_plot()
|
|
self.assertEqual(ctx.exception.code, "not_enough_gold")
|
|
|
|
def test_buy_plot_blocked_at_level_cap(self) -> None:
|
|
self.engine.state.gold = 10 ** 6
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.buy_plot()
|
|
self.assertEqual(ctx.exception.code, "plot_limit")
|
|
|
|
# ============================================================ 牲畜
|
|
def test_buy_animal_and_produce(self) -> None:
|
|
self.engine.state.gold = 1000
|
|
result = self.engine.buy_animal("chicken")
|
|
self.assertEqual(result["animal"]["kind"], "chicken")
|
|
self.assertEqual(len(self.engine.state.animals), 1)
|
|
|
|
animal = self.engine.state.animals[0]
|
|
self.engine.feed(animal.aid)
|
|
self.assertTrue(animal.fed)
|
|
self.engine.advance_day()
|
|
self.assertEqual(animal.pending, 1)
|
|
|
|
gold = self.engine.state.gold
|
|
result = self.engine.collect(animal.aid)
|
|
self.assertEqual(result["quantity"], 1)
|
|
self.assertGreater(self.engine.state.gold, gold - 1)
|
|
self.assertEqual(animal.pending, 0)
|
|
|
|
def test_feed_twice_fails(self) -> None:
|
|
self.engine.state.gold = 1000
|
|
self.engine.buy_animal("chicken")
|
|
animal = self.engine.state.animals[0]
|
|
self.engine.feed(animal.aid)
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.feed(animal.aid)
|
|
self.assertEqual(ctx.exception.code, "already_fed")
|
|
|
|
def test_collect_without_product(self) -> None:
|
|
self.engine.state.gold = 1000
|
|
self.engine.buy_animal("chicken")
|
|
animal = self.engine.state.animals[0]
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.collect(animal.aid)
|
|
self.assertEqual(ctx.exception.code, "nothing_to_collect")
|
|
|
|
def test_animal_starves_to_death(self) -> None:
|
|
self.engine.state.gold = 1000
|
|
self.engine.buy_animal("chicken")
|
|
for _ in range(3):
|
|
self.engine.advance_day()
|
|
self.assertEqual(len(self.engine.state.animals), 0)
|
|
|
|
def test_animal_survives_if_fed(self) -> None:
|
|
self.engine.state.gold = 1000
|
|
self.engine.state.feed = 10
|
|
self.engine.buy_animal("chicken")
|
|
animal = self.engine.state.animals[0]
|
|
for _ in range(6):
|
|
self.engine.feed(animal.aid)
|
|
self.engine.advance_day()
|
|
self.assertEqual(len(self.engine.state.animals), 1)
|
|
self.assertGreater(animal.pending, 0)
|
|
|
|
def test_unknown_animal(self) -> None:
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.buy_animal("dragon")
|
|
self.assertEqual(ctx.exception.code, "unknown_animal")
|
|
|
|
def test_pasture_limit(self) -> None:
|
|
self.engine.state.gold = 10 ** 6
|
|
for _ in range(config.MAX_ANIMALS):
|
|
self.engine.buy_animal("chicken")
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.buy_animal("chicken")
|
|
self.assertEqual(ctx.exception.code, "pasture_full")
|
|
|
|
def test_emergency_feed_when_out_of_stock(self) -> None:
|
|
self.engine.state.gold = 1000
|
|
self.engine.buy_animal("chicken")
|
|
self.engine.state.feed = 0
|
|
animal = self.engine.state.animals[0]
|
|
gold = self.engine.state.gold
|
|
self.engine.feed(animal.aid)
|
|
self.assertTrue(animal.fed)
|
|
self.assertLess(self.engine.state.gold, gold) # 被扣了高价补货费
|
|
|
|
def test_buy_feed_validation(self) -> None:
|
|
self.engine.state.gold = 10 ** 6
|
|
with self.assertRaises(GameError):
|
|
self.engine.buy_feed(0)
|
|
with self.assertRaises(GameError):
|
|
self.engine.buy_feed(-5)
|
|
with self.assertRaises(GameError):
|
|
self.engine.buy_feed(config.MAX_FEED_PURCHASE + 1)
|
|
with self.assertRaises(GameError):
|
|
self.engine.buy_feed("十份")
|
|
self.engine.buy_feed(5)
|
|
self.assertEqual(self.engine.state.feed, config.START_FEED + 5)
|
|
|
|
# ============================================================ 时间与市场
|
|
def test_advance_day_resets_ap(self) -> None:
|
|
self.engine.state.ap = 0
|
|
self.engine.advance_day()
|
|
self.assertEqual(self.engine.state.ap, config.START_AP)
|
|
self.assertEqual(self.engine.state.day, 2)
|
|
|
|
def test_sleep_cooldown(self) -> None:
|
|
self.engine.sleep()
|
|
with self.assertRaises(GameError) as ctx:
|
|
self.engine.sleep()
|
|
self.assertEqual(ctx.exception.code, "cooldown")
|
|
|
|
def test_market_prices_stay_in_range(self) -> None:
|
|
for _ in range(60):
|
|
self.engine.advance_day()
|
|
for key, info in config.CROPS.items():
|
|
price = self.engine.market_price(key)
|
|
base = info["base_price"]
|
|
self.assertGreaterEqual(price, base * 0.5)
|
|
self.assertLessEqual(price, base * 1.6)
|
|
|
|
def test_pause_toggle(self) -> None:
|
|
self.engine.pause(True)
|
|
self.assertTrue(self.engine.paused)
|
|
self.engine.pause(False)
|
|
self.assertFalse(self.engine.paused)
|
|
result = self.engine.pause()
|
|
self.assertTrue(result["paused"])
|
|
self.engine.pause(False)
|
|
|
|
# ============================================================ 查询接口
|
|
def test_snapshot_is_serializable(self) -> None:
|
|
import json
|
|
|
|
self.engine.plant(1, "wheat")
|
|
snap = self.engine.snapshot()
|
|
text = json.dumps(snap, ensure_ascii=False)
|
|
self.assertIn("plots", text)
|
|
|
|
def test_available_actions_reference_valid_plots(self) -> None:
|
|
actions = self.engine.available_actions()
|
|
self.assertTrue(any(a["action"] == "plant" for a in actions))
|
|
for action in actions:
|
|
if action["action"] in ("plant", "water", "harvest"):
|
|
plot = action["params"]["plot"]
|
|
self.assertTrue(1 <= plot <= len(self.engine.state.plots))
|
|
|
|
def test_catalog_marks_locked(self) -> None:
|
|
catalog = self.engine.catalog()
|
|
by_key = {c["key"]: c for c in catalog["crops"]}
|
|
self.assertFalse(by_key["wheat"]["locked"])
|
|
self.assertTrue(by_key["pumpkin"]["locked"])
|
|
|
|
# ============================================================ 存档
|
|
def test_save_load_roundtrip(self) -> None:
|
|
self.engine.plant(1, "carrot")
|
|
self.engine.buy_animal("chicken")
|
|
self.engine.state.gold = 500 # 操作完成后再定金币，避免被扣费干扰断言
|
|
self.engine.state.feed = 7
|
|
self.engine.state.day = 13
|
|
path = os.path.join(self.tmp, "roundtrip.json")
|
|
|
|
save_state(self.engine.state, path)
|
|
loaded, message = load_state(path)
|
|
self.assertIsNotNone(loaded, message)
|
|
assert loaded is not None
|
|
self.assertEqual(loaded.gold, 500)
|
|
self.assertEqual(loaded.feed, 7)
|
|
self.assertEqual(loaded.day, 13)
|
|
self.assertEqual(loaded.plots[0].crop, "carrot")
|
|
self.assertEqual(loaded.animals[0].kind, "chicken")
|
|
|
|
def test_load_missing_file(self) -> None:
|
|
state, message = load_state(os.path.join(self.tmp, "nope.json"))
|
|
self.assertIsNone(state)
|
|
self.assertIn("未找到", message)
|
|
|
|
def test_load_corrupt_file(self) -> None:
|
|
path = os.path.join(self.tmp, "broken.json")
|
|
with open(path, "w", encoding="utf-8") as handle:
|
|
handle.write("{ not valid json")
|
|
state, message = load_state(path)
|
|
self.assertIsNone(state)
|
|
self.assertIn("解析", message)
|
|
|
|
def test_load_wrong_version(self) -> None:
|
|
import json
|
|
|
|
path = os.path.join(self.tmp, "old.json")
|
|
with open(path, "w", encoding="utf-8") as handle:
|
|
json.dump({"version": 99, "state": {}}, handle)
|
|
state, message = load_state(path)
|
|
self.assertIsNone(state)
|
|
self.assertIn("版本", message)
|
|
|
|
def test_state_from_dict_repairs_bad_data(self) -> None:
|
|
state = GameState.from_dict(
|
|
{
|
|
"day": -5,
|
|
"gold": -100,
|
|
"level": 0,
|
|
"plots": [{"index": 0, "crop": "不存在", "growth": 999, "moisture": 99}],
|
|
"animals": [{"kind": "独角兽"}],
|
|
"market": {"wheat": "not a number"},
|
|
}
|
|
)
|
|
self.assertEqual(state.day, 1)
|
|
self.assertEqual(state.gold, 0)
|
|
self.assertEqual(state.level, 1)
|
|
self.assertIsNone(state.plots[0].crop) # 非法作物被清空
|
|
self.assertEqual(state.animals, []) # 非法动物被丢弃
|
|
self.assertGreater(state.market["wheat"], 0) # 非法价格回落默认值
|
|
self.assertGreaterEqual(len(state.plots), config.BASE_PLOTS)
|
|
|
|
# ============================================================ 并发
|
|
def test_concurrent_operations_are_safe(self) -> None:
|
|
self.engine.state.gold = 10 ** 6
|
|
errors = []
|
|
|
|
def worker(index: int) -> None:
|
|
try:
|
|
for _ in range(30):
|
|
self.engine.snapshot()
|
|
self.engine.available_actions()
|
|
if index % 3 == 0:
|
|
try:
|
|
self.engine.buy_feed(1)
|
|
except GameError:
|
|
pass
|
|
except Exception as exc: # 记录任何非预期异常
|
|
errors.append(exc)
|
|
|
|
threads = [threading.Thread(target=worker, args=(i,)) for i in range(12)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
self.assertEqual(errors, [])
|
|
|
|
|
|
class PlotModelTestCase(unittest.TestCase):
|
|
def test_stage_progression(self) -> None:
|
|
plot = Plot(0)
|
|
self.assertEqual(plot.stage, "空地")
|
|
plot.crop = "wheat"
|
|
self.assertEqual(plot.stage, "已播种")
|
|
plot.growth = 0.7
|
|
self.assertEqual(plot.stage, "幼苗")
|
|
plot.growth = 1.5
|
|
self.assertEqual(plot.stage, "生长中")
|
|
plot.growth = 2
|
|
self.assertTrue(plot.is_ripe)
|
|
self.assertEqual(plot.stage, "已成熟")
|
|
|
|
def test_days_left_never_negative(self) -> None:
|
|
plot = Plot(0)
|
|
plot.crop = "wheat"
|
|
plot.growth = 99
|
|
self.assertEqual(plot.days_left, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|