snowy-yang
d7a7107b0a
pyright（basic 模式）从 15 errors / 10 warnings 降到 0/0： - pyproject: pythonPlatform="All"（termios/msvcrt 双平台分支不再误报）， reportUnsupportedDunderAll=false（__init__ 惰性导出属正常写法） - keys.py: 捕获 sys.stdin 局部变量 + 提前返回，消除 fileno() 的 None 收窄告警 - server/app.py: lifespan 改用 @asynccontextmanager 并标注 AsyncIterator[None] - server/actions.py: 显式标注 result: dict[str, Any]，消除 isinstance 冗余报错 - server/__main__.py: 用 importlib.util.find_spec 探测 websockets - tests/test_server.py: 同上，uvicorn 探测改为 find_spec - tools/preview_tui.py: TYPE_CHECKING 下引入 RemoteFarm，修正返回类型标注 - farm/tui.py: 取消未使用变量，统一 round/取整写法 其他： - README: 新增第 11 节「工程化：uv + ruff + pyright」，快速开始全部改用 uv sync / uv run；原第 11 节起顺延为 12 - requirements.txt: 由 uv export --no-dev --no-hashes 重新生成（锁定版本） - 清理临时日志文件 验收：ruff check 全通过 / ruff format 26 文件无改动 / pyright 0 错误 / unittest 97 项全部通过
619 lines
25 KiB
Python
619 lines
25 KiB
Python
"""服务器层测试：动作表校验、REST 接口、多牧场隔离、鉴权、WebSocket 端到端。
|
||
|
||
分三层
|
||
------
|
||
1. :class:`TestActionTable` / :class:`TestFarmId` —— 纯函数，不需要服务器；
|
||
2. :class:`TestRestAPI` —— 用 FastAPI 的 TestClient 在进程内打接口；
|
||
3. :class:`TestLiveServer` —— 真的起一个 uvicorn，再用 :class:`farm.client.RemoteFarm`
|
||
通过 HTTP + WebSocket 连上去。这一层才是"客户端到底行不行"的答案。
|
||
|
||
如果机器上没装 FastAPI/uvicorn，第 2、3 层会跳过而不是失败——
|
||
单机引擎的测试在 test_engine.py，那部分永远可跑。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import importlib.util
|
||
import json
|
||
import os
|
||
import shutil
|
||
import socket
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
import unittest
|
||
|
||
from farm import config
|
||
from farm.engine import FarmEngine, GameError
|
||
from farm.server.actions import ACTIONS, describe_actions, run_action
|
||
from farm.server.session import SessionManager, validate_farm_id
|
||
|
||
try:
|
||
# 能导入 TestClient 就说明 fastapi 在，不需要再单独 import 一次
|
||
from fastapi.testclient import TestClient
|
||
|
||
HAS_FASTAPI = True
|
||
except ImportError: # pragma: no cover - 取决于环境
|
||
HAS_FASTAPI = False
|
||
|
||
# uvicorn 只需要知道"在不在"，不必真的导入（导入它是 TestLiveServer 里的事）
|
||
HAS_UVICORN = importlib.util.find_spec("uvicorn") is not None
|
||
|
||
|
||
def free_port() -> int:
|
||
with socket.socket() as sock:
|
||
sock.bind(("127.0.0.1", 0))
|
||
return sock.getsockname()[1]
|
||
|
||
|
||
# ======================================================================= 纯逻辑
|
||
class TestActionTable(unittest.TestCase):
|
||
"""动作表是所有入口的唯一执行通道，参数校验必须严格。"""
|
||
|
||
def setUp(self) -> None:
|
||
self.engine = FarmEngine(tick_seconds=60)
|
||
|
||
def test_all_actions_documented(self) -> None:
|
||
described = describe_actions()
|
||
self.assertEqual(len(described), len(ACTIONS))
|
||
for item in described:
|
||
self.assertTrue(item["summary"])
|
||
self.assertIn(item["action"], ACTIONS)
|
||
|
||
def test_plant_and_harvest_via_action_table(self) -> None:
|
||
result = run_action(self.engine, "plant", {"plot": 1, "crop": "wheat"})
|
||
self.assertTrue(result["ok"])
|
||
self.assertEqual(result["action"], "plant")
|
||
state = self.engine.snapshot()
|
||
self.assertEqual(state["plots"][0]["crop"], "wheat")
|
||
|
||
self.engine.advance_day()
|
||
self.engine.advance_day()
|
||
harvested = run_action(self.engine, "harvest", {"plot": 1})
|
||
self.assertGreater(harvested["earned"], 0)
|
||
|
||
def test_string_number_is_coerced(self) -> None:
|
||
"""命令行/表单常把数字传成字符串，这里必须能识别。"""
|
||
result = run_action(self.engine, "plant", {"plot": "1", "crop": "wheat"})
|
||
self.assertTrue(result["ok"])
|
||
|
||
def test_unknown_action_rejected(self) -> None:
|
||
with self.assertRaises(GameError) as ctx:
|
||
run_action(self.engine, "fly_to_moon", {})
|
||
self.assertEqual(ctx.exception.code, "unknown_action")
|
||
self.assertEqual(ctx.exception.status, 400)
|
||
|
||
def test_missing_required_param(self) -> None:
|
||
with self.assertRaises(GameError) as ctx:
|
||
run_action(self.engine, "plant", {"plot": 1})
|
||
self.assertEqual(ctx.exception.code, "missing_param")
|
||
|
||
def test_unknown_param_rejected(self) -> None:
|
||
with self.assertRaises(GameError) as ctx:
|
||
run_action(self.engine, "plant", {"plot": 1, "crop": "wheat", "magic": 1})
|
||
self.assertEqual(ctx.exception.code, "bad_param")
|
||
self.assertIn("magic", ctx.exception.message)
|
||
|
||
def test_bad_type_rejected(self) -> None:
|
||
with self.assertRaises(GameError) as ctx:
|
||
run_action(self.engine, "plant", {"plot": "第一块地", "crop": "wheat"})
|
||
self.assertEqual(ctx.exception.code, "bad_param")
|
||
|
||
def test_bool_param_forms(self) -> None:
|
||
result = run_action(self.engine, "pause", {"paused": "true"})
|
||
self.assertTrue(self.engine.paused)
|
||
result = run_action(self.engine, "pause", {"paused": False})
|
||
self.assertFalse(self.engine.paused)
|
||
self.assertEqual(result["action"], "pause")
|
||
with self.assertRaises(GameError):
|
||
run_action(self.engine, "pause", {"paused": "也许"})
|
||
|
||
def test_harvest_all_with_nothing_ripe(self) -> None:
|
||
with self.assertRaises(GameError) as ctx:
|
||
run_action(self.engine, "harvest_all", {})
|
||
self.assertEqual(ctx.exception.code, "nothing_to_harvest")
|
||
|
||
def test_business_error_keeps_code(self) -> None:
|
||
"""业务错误要保留机器可读的 code，AI 靠它分支。"""
|
||
with self.assertRaises(GameError) as ctx:
|
||
run_action(self.engine, "plant", {"plot": 99, "crop": "wheat"})
|
||
self.assertEqual(ctx.exception.code, "no_such_plot")
|
||
with self.assertRaises(GameError) as ctx:
|
||
run_action(self.engine, "buy_animal", {"kind": "dragon"})
|
||
self.assertEqual(ctx.exception.code, "unknown_animal")
|
||
|
||
|
||
class TestFarmId(unittest.TestCase):
|
||
"""牧场 ID 会变成文件名，必须挡住路径穿越之类的输入。"""
|
||
|
||
def test_normal(self) -> None:
|
||
for name in ("default", "alice", "ai-1", "玩家甲", "farm_01"):
|
||
self.assertEqual(validate_farm_id(name), name)
|
||
|
||
def test_default_when_empty(self) -> None:
|
||
self.assertEqual(validate_farm_id(None), "default")
|
||
self.assertEqual(validate_farm_id(""), "default")
|
||
|
||
def test_rejects_path_traversal(self) -> None:
|
||
for bad in ("../etc/passwd", "a/b", "a\\b", "..", "a b", "x" * 40, "农场:1"):
|
||
with self.assertRaises(GameError):
|
||
validate_farm_id(bad)
|
||
|
||
|
||
class TestSessionManager(unittest.TestCase):
|
||
"""多牧场隔离与回收策略。"""
|
||
|
||
def setUp(self) -> None:
|
||
self.tmp = tempfile.mkdtemp(prefix="farm-session-")
|
||
self.manager = SessionManager(save_dir=self.tmp, tick_seconds=60)
|
||
|
||
def tearDown(self) -> None:
|
||
self.manager.shutdown()
|
||
shutil.rmtree(self.tmp, ignore_errors=True)
|
||
|
||
def test_farms_are_isolated(self) -> None:
|
||
alice = self.manager.get("alice")
|
||
bob = self.manager.get("bob")
|
||
alice.engine.plant(1, "wheat")
|
||
alice.engine.state.gold = 999
|
||
|
||
self.assertIsNone(bob.engine.state.plots[0].crop)
|
||
self.assertNotEqual(bob.engine.state.gold, 999)
|
||
self.assertEqual(self.manager.count(), 2)
|
||
self.assertIs(self.manager.get("alice"), alice)
|
||
|
||
def test_save_and_reload_roundtrip(self) -> None:
|
||
alice = self.manager.get("alice")
|
||
alice.engine.plant(1, "carrot")
|
||
alice.engine.state.gold = 777
|
||
alice.save()
|
||
|
||
fresh = SessionManager(save_dir=self.tmp, tick_seconds=60)
|
||
try:
|
||
again = fresh.get("alice")
|
||
self.assertEqual(again.engine.state.gold, 777)
|
||
self.assertEqual(again.engine.state.plots[0].crop, "carrot")
|
||
finally:
|
||
fresh.shutdown()
|
||
|
||
def test_leaderboard_sorted_by_net_worth(self) -> None:
|
||
poor = self.manager.get("poor")
|
||
rich = self.manager.get("rich")
|
||
poor.engine.state.gold = 10
|
||
rich.engine.state.gold = 5000
|
||
rows = self.manager.leaderboard()
|
||
self.assertEqual(rows[0]["farm"], "rich")
|
||
self.assertEqual(rows[0]["rank"], 1)
|
||
self.assertEqual(rows[1]["farm"], "poor")
|
||
|
||
def test_reap_only_drops_idle_offline_farms(self) -> None:
|
||
busy = self.manager.get("busy")
|
||
idle = self.manager.get("idle")
|
||
busy.last_active = time.time() - 9999
|
||
idle.last_active = time.time() - 9999
|
||
self.manager.idle_timeout = 10
|
||
|
||
dropped = self.manager.reap(online_map={"busy": 1})
|
||
self.assertEqual(dropped, ["idle"])
|
||
self.assertIsNotNone(self.manager.find("busy"))
|
||
self.assertIsNone(self.manager.find("idle"))
|
||
# 被回收的牧场一定已经落盘
|
||
self.assertTrue(os.path.exists(os.path.join(self.tmp, "idle.json")))
|
||
|
||
def test_paused_farm_is_not_reaped(self) -> None:
|
||
session = self.manager.get("thinking")
|
||
session.engine.pause(True)
|
||
session.last_active = time.time() - 9999
|
||
self.manager.idle_timeout = 10
|
||
self.assertEqual(self.manager.reap(), [])
|
||
|
||
def test_max_sessions_guard(self) -> None:
|
||
manager = SessionManager(save_dir=self.tmp, max_sessions=2)
|
||
try:
|
||
manager.get("a")
|
||
manager.get("b")
|
||
with self.assertRaises(GameError) as ctx:
|
||
manager.get("c")
|
||
self.assertEqual(ctx.exception.code, "too_many_farms")
|
||
finally:
|
||
manager.shutdown()
|
||
|
||
|
||
# ======================================================================= REST
|
||
@unittest.skipUnless(HAS_FASTAPI, "未安装 FastAPI")
|
||
class TestRestAPI(unittest.TestCase):
|
||
"""用 TestClient 在进程内验证 HTTP 语义。"""
|
||
|
||
def setUp(self) -> None:
|
||
from farm.server.app import ServerConfig, create_app
|
||
|
||
self.tmp = tempfile.mkdtemp(prefix="farm-api-")
|
||
self.cfg = ServerConfig(host="127.0.0.1", port=8765, save_dir=self.tmp, tick_seconds=60)
|
||
self.app = create_app(self.cfg)
|
||
self.client = TestClient(self.app)
|
||
self.client.__enter__()
|
||
|
||
def tearDown(self) -> None:
|
||
self.client.__exit__(None, None, None)
|
||
shutil.rmtree(self.tmp, ignore_errors=True)
|
||
|
||
# ------------------------------------------------------------ 基础
|
||
def test_health_without_token(self) -> None:
|
||
data = self.client.get("/api/health").json()
|
||
self.assertTrue(data["ok"])
|
||
self.assertIn("uptime", data)
|
||
|
||
def test_dashboard_and_text_index(self) -> None:
|
||
page = self.client.get("/")
|
||
self.assertEqual(page.status_code, 200)
|
||
self.assertIn("SimpleFarm", page.text)
|
||
self.assertIn("WebSocket", page.text)
|
||
text = self.client.get("/text").text
|
||
self.assertIn("/api/actions", text)
|
||
|
||
def test_help_lists_actions(self) -> None:
|
||
data = self.client.get("/api/help").json()
|
||
self.assertIn("rules", data)
|
||
names = {item["action"] for item in data["actions"]}
|
||
self.assertIn("plant", names)
|
||
self.assertIn("harvest_all", names)
|
||
|
||
def test_openapi_schema_available(self) -> None:
|
||
schema = self.client.get("/openapi.json").json()
|
||
self.assertIn("/api/actions", schema["paths"])
|
||
|
||
# ------------------------------------------------------------ 牧场读写
|
||
def test_state_creates_farm_lazily(self) -> None:
|
||
data = self.client.get("/api/state?farm=newbie").json()
|
||
self.assertEqual(data["state"]["day"], 1)
|
||
self.assertEqual(data["state"]["gold"], config.START_GOLD)
|
||
|
||
def test_generic_action_entry(self) -> None:
|
||
body = {"action": "plant", "params": {"plot": 1, "crop": "wheat"}}
|
||
data = self.client.post("/api/actions?farm=one", json=body).json()
|
||
self.assertTrue(data["ok"])
|
||
self.assertIn("已种下", data["message"])
|
||
self.assertEqual(
|
||
data["state"]["gold"], config.START_GOLD - config.CROPS["wheat"]["seed_cost"]
|
||
)
|
||
|
||
state = self.client.get("/api/state?farm=one").json()["state"]
|
||
self.assertEqual(state["plots"][0]["crop"], "wheat")
|
||
|
||
def test_flat_params_also_accepted(self) -> None:
|
||
"""{"action":"plant","plot":1,"crop":"wheat"} 这种平铺写法也要能用。"""
|
||
data = self.client.post(
|
||
"/api/actions?farm=flat", json={"action": "plant", "plot": 1, "crop": "wheat"}
|
||
).json()
|
||
self.assertTrue(data["ok"])
|
||
|
||
def test_convenience_route(self) -> None:
|
||
data = self.client.post("/api/plant?farm=short", json={"plot": 2, "crop": "carrot"}).json()
|
||
self.assertTrue(data["ok"])
|
||
self.assertEqual(data["action"], "plant")
|
||
|
||
def test_action_error_shape(self) -> None:
|
||
response = self.client.post(
|
||
"/api/actions?farm=err", json={"action": "plant", "params": {"plot": 1, "crop": "nope"}}
|
||
)
|
||
self.assertEqual(response.status_code, 400)
|
||
data = response.json()
|
||
self.assertFalse(data["ok"])
|
||
self.assertEqual(data["error"], "unknown_crop")
|
||
self.assertIn("message", data)
|
||
|
||
def test_out_of_range_plot(self) -> None:
|
||
response = self.client.post(
|
||
"/api/actions?farm=err", json={"action": "water", "params": {"plot": 999}}
|
||
)
|
||
self.assertEqual(response.status_code, 400)
|
||
self.assertEqual(response.json()["error"], "no_such_plot")
|
||
|
||
def test_bad_json_body(self) -> None:
|
||
response = self.client.post(
|
||
"/api/actions?farm=err", content=b"{oops", headers={"Content-Type": "application/json"}
|
||
)
|
||
self.assertEqual(response.status_code, 400)
|
||
self.assertEqual(response.json()["error"], "bad_json")
|
||
|
||
def test_actions_endpoint_matches_state(self) -> None:
|
||
data = self.client.get("/api/actions?farm=solo").json()
|
||
names = {item["action"] for item in data["actions"]}
|
||
self.assertIn("plant", names) # 有空地 + 有钱 → 可播种
|
||
self.assertNotIn("harvest", names) # 还没种东西 → 不可收获
|
||
|
||
def test_batch_endpoint(self) -> None:
|
||
items = {
|
||
"actions": [
|
||
{"action": "buy_feed", "params": {"quantity": 5}},
|
||
{"action": "plant", "params": {"plot": 1, "crop": "wheat"}},
|
||
{"action": "harvest", "params": {"plot": 1}}, # 必然失败
|
||
]
|
||
}
|
||
data = self.client.post("/api/batch?farm=multi", json=items).json()
|
||
self.assertEqual(data["done"], 2)
|
||
self.assertEqual(data["failed"], 1)
|
||
self.assertEqual(data["results"][2]["error"], "not_ripe")
|
||
|
||
def test_log_entries_and_since(self) -> None:
|
||
self.client.post(
|
||
"/api/actions?farm=logging",
|
||
json={"action": "plant", "params": {"plot": 1, "crop": "wheat"}},
|
||
)
|
||
entries = self.client.get("/api/log?farm=logging").json()["entries"]
|
||
self.assertTrue(entries)
|
||
last = entries[-1]["seq"]
|
||
newer = self.client.get(f"/api/log?farm=logging&since={last}").json()["entries"]
|
||
self.assertEqual(newer, [])
|
||
self.client.post("/api/actions?farm=logging", json={"action": "save"})
|
||
newer = self.client.get(f"/api/log?farm=logging&since={last}").json()["entries"]
|
||
self.assertTrue(newer)
|
||
|
||
def test_farms_list_and_isolation(self) -> None:
|
||
self.client.post(
|
||
"/api/actions?farm=alice", json={"action": "buy_feed", "params": {"quantity": 10}}
|
||
)
|
||
self.client.post("/api/actions?farm=bob", json={"action": "save"})
|
||
rows = self.client.get("/api/farms").json()["farms"]
|
||
by_name = {row["farm"]: row for row in rows}
|
||
self.assertIn("alice", by_name)
|
||
self.assertEqual(by_name["alice"]["gold"], config.START_GOLD - 10 * config.FEED_UNIT_PRICE)
|
||
|
||
def test_create_farm_reports_created_flag(self) -> None:
|
||
first = self.client.post("/api/farms", json={"farm": "zoe", "owner": "小Z"}).json()
|
||
self.assertTrue(first["created"])
|
||
self.assertEqual(first["summary"]["owner"], "小Z")
|
||
again = self.client.post("/api/farms", json={"farm": "zoe"}).json()
|
||
self.assertFalse(again["created"])
|
||
|
||
def test_bad_farm_id_rejected(self) -> None:
|
||
response = self.client.get("/api/state?farm=../evil")
|
||
self.assertEqual(response.status_code, 400)
|
||
self.assertEqual(response.json()["error"], "bad_farm")
|
||
|
||
def test_farm_via_header(self) -> None:
|
||
data = self.client.get("/api/state", headers={"X-Farm-Id": "header-farm"}).json()
|
||
self.assertEqual(data["farm"], "header-farm")
|
||
|
||
def test_admin_save(self) -> None:
|
||
self.client.post(
|
||
"/api/actions?farm=saver",
|
||
json={"action": "plant", "params": {"plot": 1, "crop": "wheat"}},
|
||
)
|
||
data = self.client.post("/api/admin/save").json()
|
||
self.assertIn("saver", data["farms"])
|
||
self.assertTrue(os.path.exists(os.path.join(self.tmp, "saver.json")))
|
||
|
||
def test_world_view(self) -> None:
|
||
self.client.post("/api/actions?farm=w1", json={"action": "save"})
|
||
data = self.client.get("/api/world").json()
|
||
self.assertEqual(data["farms"], 1)
|
||
self.assertEqual(data["leaderboard"][0]["farm"], "w1")
|
||
|
||
|
||
@unittest.skipUnless(HAS_FASTAPI, "未安装 FastAPI")
|
||
class TestAuth(unittest.TestCase):
|
||
"""令牌鉴权：没带令牌必须 401，带了才能操作。"""
|
||
|
||
def setUp(self) -> None:
|
||
from farm.server.app import ServerConfig, create_app
|
||
|
||
self.tmp = tempfile.mkdtemp(prefix="farm-auth-")
|
||
self.app = create_app(ServerConfig(save_dir=self.tmp, token="s3cret", tick_seconds=60))
|
||
self.client = TestClient(self.app)
|
||
self.client.__enter__()
|
||
|
||
def tearDown(self) -> None:
|
||
self.client.__exit__(None, None, None)
|
||
shutil.rmtree(self.tmp, ignore_errors=True)
|
||
|
||
def test_health_is_public(self) -> None:
|
||
self.assertEqual(self.client.get("/api/health").status_code, 200)
|
||
|
||
def test_state_requires_token(self) -> None:
|
||
response = self.client.get("/api/state")
|
||
self.assertEqual(response.status_code, 401)
|
||
self.assertEqual(response.json()["error"], "unauthorized")
|
||
|
||
def test_token_via_header(self) -> None:
|
||
response = self.client.get("/api/state", headers={"X-API-Token": "s3cret"})
|
||
self.assertEqual(response.status_code, 200)
|
||
|
||
def test_token_via_query(self) -> None:
|
||
response = self.client.get("/api/state?token=s3cret")
|
||
self.assertEqual(response.status_code, 200)
|
||
|
||
def test_wrong_token(self) -> None:
|
||
self.assertEqual(self.client.get("/api/state?token=nope").status_code, 401)
|
||
|
||
def test_action_blocked_without_token(self) -> None:
|
||
response = self.client.post("/api/actions", json={"action": "buy_plot"})
|
||
self.assertEqual(response.status_code, 401)
|
||
|
||
|
||
# ======================================================================= 真网络
|
||
@unittest.skipUnless(HAS_FASTAPI and HAS_UVICORN, "未安装 FastAPI/uvicorn")
|
||
class TestLiveServer(unittest.TestCase):
|
||
"""真的起 uvicorn，用真实客户端走 HTTP 与 WebSocket。"""
|
||
|
||
@classmethod
|
||
def setUpClass(cls) -> None:
|
||
from farm.server.app import ServerConfig, create_app
|
||
|
||
cls.tmp = tempfile.mkdtemp(prefix="farm-live-")
|
||
cls.port = free_port()
|
||
cls.url = f"http://127.0.0.1:{cls.port}"
|
||
app = create_app(
|
||
ServerConfig(
|
||
host="127.0.0.1",
|
||
port=cls.port,
|
||
save_dir=cls.tmp,
|
||
tick_seconds=600,
|
||
autosave_interval=3600,
|
||
)
|
||
)
|
||
import uvicorn
|
||
|
||
cls.server = uvicorn.Server(
|
||
uvicorn.Config(
|
||
app,
|
||
host="127.0.0.1",
|
||
port=cls.port,
|
||
log_level="warning",
|
||
access_log=False,
|
||
lifespan="on",
|
||
)
|
||
)
|
||
cls.thread = threading.Thread(target=cls.server.run, name="live-server", daemon=True)
|
||
cls.thread.start()
|
||
|
||
deadline = time.time() + 20
|
||
while time.time() < deadline:
|
||
try:
|
||
with socket.create_connection(("127.0.0.1", cls.port), timeout=0.5):
|
||
break
|
||
except OSError:
|
||
time.sleep(0.2)
|
||
else: # pragma: no cover
|
||
raise unittest.SkipTest("服务器没能在 20 秒内起来")
|
||
|
||
@classmethod
|
||
def tearDownClass(cls) -> None:
|
||
cls.server.should_exit = True
|
||
cls.thread.join(timeout=10)
|
||
shutil.rmtree(cls.tmp, ignore_errors=True)
|
||
|
||
def client(self, farm: str, realtime: bool = True):
|
||
from farm.client import RemoteFarm
|
||
|
||
return RemoteFarm(
|
||
endpoint=self.url, farm=farm, realtime=realtime, timeout=8.0, autostart=True
|
||
)
|
||
|
||
def test_http_roundtrip_through_real_client(self) -> None:
|
||
client = self.client("live-http", realtime=False)
|
||
try:
|
||
state = client.snapshot()
|
||
self.assertEqual(state["farm"], "live-http")
|
||
result = client.plant(1, "wheat")
|
||
self.assertTrue(result["ok"])
|
||
self.assertEqual(client.state.plots[0].crop, "wheat")
|
||
self.assertFalse(client.state.plots[0].is_ripe)
|
||
|
||
with self.assertRaises(GameError) as ctx:
|
||
client.plant(1, "carrot") # 地已经占用了
|
||
self.assertEqual(ctx.exception.code, "plot_busy")
|
||
|
||
actions = client.available_actions()
|
||
names = {item["action"] for item in actions}
|
||
self.assertIn("plant", names) # 还有空地可以种
|
||
self.assertNotIn("harvest", names) # 小麦还没成熟
|
||
|
||
client.water(1) # 播种后水分 2/3，可以补满
|
||
self.assertEqual(client.state.plots[0].moisture, config.WATER_CAPACITY)
|
||
with self.assertRaises(GameError) as ctx:
|
||
client.water(1) # 水分已满，再浇就是错的
|
||
self.assertEqual(ctx.exception.code, "already_watered")
|
||
|
||
catalog = client.catalog()
|
||
self.assertTrue(catalog["crops"])
|
||
self.assertTrue(client.leaderboard())
|
||
self.assertIn("规则速览", client.help_text())
|
||
finally:
|
||
client.stop()
|
||
|
||
def test_two_clients_talk_to_the_same_farm(self) -> None:
|
||
"""两个客户端改同一个牧场：一方操作，另一方从推送里看到。"""
|
||
watcher = self.client("shared")
|
||
actor = self.client("shared")
|
||
try:
|
||
deadline = time.time() + 8
|
||
while time.time() < deadline and len(watcher.state.plots) == 0:
|
||
time.sleep(0.1)
|
||
self.assertTrue(watcher.state.plots)
|
||
|
||
actor.plant(1, "carrot")
|
||
deadline = time.time() + 8
|
||
while time.time() < deadline and watcher.state.plots[0].crop != "carrot":
|
||
time.sleep(0.1)
|
||
self.assertEqual(
|
||
watcher.state.plots[0].crop, "carrot", "实时通道应该把另一方的操作推过来"
|
||
)
|
||
self.assertEqual(watcher.mode, "ws")
|
||
finally:
|
||
watcher.stop()
|
||
actor.stop()
|
||
|
||
def test_websocket_channel_is_really_used(self) -> None:
|
||
"""写了动作应该走 WS 通道，而不是退化到 HTTP。"""
|
||
client = self.client("ws-farm")
|
||
try:
|
||
deadline = time.time() + 8
|
||
while time.time() < deadline and client.mode != "ws":
|
||
time.sleep(0.1)
|
||
self.assertEqual(client.mode, "ws", "WebSocket 没能连上")
|
||
client.buy_feed(3)
|
||
self.assertEqual(client.stats["ws_actions"], 1)
|
||
self.assertEqual(client.stats["http"] > 0, True)
|
||
finally:
|
||
client.stop()
|
||
|
||
def test_farms_do_not_leak_into_each_other(self) -> None:
|
||
a = self.client("iso-a", realtime=False)
|
||
b = self.client("iso-b", realtime=False)
|
||
try:
|
||
a.plant(1, "wheat")
|
||
self.assertIsNone(b.state.plots[0].crop)
|
||
self.assertNotEqual(a.state.gold, b.state.gold)
|
||
finally:
|
||
a.stop()
|
||
b.stop()
|
||
|
||
def test_save_on_server_persists(self) -> None:
|
||
client = self.client("persist", realtime=False)
|
||
try:
|
||
client.plant(1, "wheat")
|
||
client.save()
|
||
path = os.path.join(self.tmp, "persist.json")
|
||
self.assertTrue(os.path.exists(path))
|
||
with open(path, encoding="utf-8") as handle:
|
||
payload = json.load(handle)
|
||
self.assertEqual(payload["state"]["plots"][0]["crop"], "wheat")
|
||
finally:
|
||
client.stop()
|
||
|
||
def test_bot_can_play_via_the_api(self) -> None:
|
||
"""规则机器人只调接口，也应该能真的把农场经营起来。"""
|
||
from farm.bot import RuleBot
|
||
|
||
client = self.client("bot-farm", realtime=False)
|
||
bot = RuleBot(client, verbose=False, pause=0.02, name="bot-farm")
|
||
try:
|
||
summary = bot.run(rounds=12, idle_limit=3)
|
||
self.assertGreaterEqual(summary["actions"], 1)
|
||
self.assertIn("net_worth", summary)
|
||
self.assertEqual(bot.failures, 0)
|
||
self.assertTrue(client.available_actions() is not None)
|
||
finally:
|
||
client.stop()
|
||
|
||
def test_rate_free_path_uses_ws_and_falls_back(self) -> None:
|
||
"""WS 断开后写操作要能自动退回 HTTP，而不是直接失败。"""
|
||
client = self.client("fallback")
|
||
try:
|
||
deadline = time.time() + 8
|
||
while time.time() < deadline and client.mode != "ws":
|
||
time.sleep(0.1)
|
||
self.assertEqual(client.mode, "ws")
|
||
# 强行掐断：让后续写操作只能走 HTTP
|
||
client._stop.set()
|
||
client._ws_socket = None
|
||
result = client.buy_feed(1)
|
||
self.assertTrue(result["ok"])
|
||
finally:
|
||
client.stop()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|