snowy-yang
ea346be484
顺着验收网页修复时发现：default 牧场根本没写进库，内存里每次 重启都是新档。根因是 session.dirty 一个标志位伺二主： * Publisher 推送循环每 ~250ms 就把 dirty 清掉（无人在线时直接清， "没人在看就不做序列化，省 CPU"）； * autosave() 每 60s 才看一眼 dirty。 dirty=True 的窗口只有约 250ms，自动存档撞上它的概率约 0.4%—— 等于永不存盘。进度只在手动存档/优雅停机时落盘，进程被强杀全丢。 修法：新增 session.unsaved 专管「有变化没落盘」，_on_change / load / touch / advance_day 置位，只有成功的 save() 清位；autosave() 改看 unsaved；dirty 继续只管推送节流，hub.py 的推送循环一字未动。 两只旧测试直接操纵 dirty 驱动 autosave，固化了被修掉的耦合，改成 unsaved 表达意图；新增回归测试：推送循环清了 dirty 也要照常落盘、 没有新变化不重复写库。 真机验证：0 客户端在线纯 HTTP 种一块地，70 秒内库行更新（金币 120→112，种子钱都对得上）。515 项测试 OK。
1316 lines
52 KiB
Python
1316 lines
52 KiB
Python
"""服务器层测试：牧场会话、调度时钟、SSE 订阅池、HTTP 接口、守护进程。
|
||
|
||
分成三档，缺依赖时自动降级而不是报错：
|
||
|
||
* 纯逻辑档（session / hub / daemon / 工具函数）——只需标准库；
|
||
* **TestClient 档**——用 FastAPI 的进程内客户端打接口，需要 ``httpx``；
|
||
* **真网络档**——真起 uvicorn、真开 socket、真读 SSE 流，只用标准库。
|
||
|
||
uv run python -m unittest tests.test_server -v
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import socket
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
import unittest
|
||
import urllib.request
|
||
|
||
from farm import config, net
|
||
from farm.engine import GameError
|
||
from farm.server.daemon import DaemonManager, ensure_parent_dir, pid_alive, port_busy
|
||
from farm.server.hub import (
|
||
CLOSE_MESSAGE,
|
||
Client,
|
||
ConnectionHub,
|
||
Publisher,
|
||
build_state_message,
|
||
next_seq,
|
||
sse_frame,
|
||
sse_heartbeat,
|
||
)
|
||
from farm.server.session import (
|
||
DEFAULT_FARM,
|
||
FarmClock,
|
||
FarmSession,
|
||
SessionManager,
|
||
validate_farm_id,
|
||
validate_owner,
|
||
)
|
||
from farm.storage import DB_NAME, farm_exists
|
||
|
||
try: # pragma: no cover - 环境相关
|
||
import fastapi # noqa: F401
|
||
|
||
from farm.server.app import (
|
||
BRIEF_KEYS,
|
||
ServerConfig,
|
||
ServerHandle,
|
||
_int,
|
||
as_asgi,
|
||
brief,
|
||
create_app,
|
||
rules_text,
|
||
)
|
||
|
||
HAS_FASTAPI = True
|
||
except ImportError: # pragma: no cover
|
||
HAS_FASTAPI = False
|
||
|
||
try: # pragma: no cover
|
||
from fastapi.testclient import TestClient
|
||
|
||
HAS_TESTCLIENT = True
|
||
except Exception: # pragma: no cover - 缺 httpx 时走这里
|
||
HAS_TESTCLIENT = False
|
||
|
||
try: # pragma: no cover
|
||
import uvicorn
|
||
|
||
HAS_UVICORN = True
|
||
except ImportError: # pragma: no cover
|
||
HAS_UVICORN = False
|
||
|
||
|
||
def free_port() -> int:
|
||
"""向系统要一个当前空闲的端口。"""
|
||
with socket.socket() as probe:
|
||
probe.bind(("127.0.0.1", 0))
|
||
return int(probe.getsockname()[1])
|
||
|
||
|
||
def wait_until(predicate, timeout: float = 6.0, interval: float = 0.05) -> bool:
|
||
deadline = time.time() + timeout
|
||
while time.time() < deadline:
|
||
if predicate():
|
||
return True
|
||
time.sleep(interval)
|
||
return bool(predicate())
|
||
|
||
|
||
class FakeLoop:
|
||
"""够用的假事件循环：直接同步执行，省掉为每条推送起真循环。"""
|
||
|
||
def __init__(self, broken: bool = False) -> None:
|
||
self.calls = 0
|
||
self.broken = broken
|
||
|
||
def call_soon_threadsafe(self, func, *args) -> None:
|
||
if self.broken:
|
||
raise RuntimeError("event loop is closed")
|
||
self.calls += 1
|
||
func(*args)
|
||
|
||
|
||
class FarmIdTestCase(unittest.TestCase):
|
||
"""牧场 ID 会变成存档文件名，所以它同时是安全边界。"""
|
||
|
||
def test_empty_falls_back_to_default(self):
|
||
self.assertEqual(validate_farm_id(None), DEFAULT_FARM)
|
||
self.assertEqual(validate_farm_id(""), DEFAULT_FARM)
|
||
# 只有 None 与空串才算没给；纯空白会被 strip 成空串，从而被正则挡下
|
||
with self.assertRaises(GameError) as ctx:
|
||
validate_farm_id(" ")
|
||
self.assertEqual(ctx.exception.code, "bad_farm")
|
||
|
||
def test_normal_ids_pass(self):
|
||
for raw in ("alice", "ai-1", "farm_01", "牧场一号", "A" * 24, "123"):
|
||
with self.subTest(raw=raw):
|
||
self.assertEqual(validate_farm_id(raw), raw)
|
||
|
||
def test_surrounding_whitespace_is_trimmed(self):
|
||
self.assertEqual(validate_farm_id(" alice "), "alice")
|
||
|
||
def test_numbers_are_stringified(self):
|
||
self.assertEqual(validate_farm_id(42), "42")
|
||
|
||
def test_path_traversal_is_refused(self):
|
||
"""这是这个校验存在的核心理由：ID 会被拼进文件路径。"""
|
||
for raw in ("../secret", "..\\secret", "a/b", "a\\b", "..", "."):
|
||
with self.subTest(raw=raw):
|
||
with self.assertRaises(GameError) as ctx:
|
||
validate_farm_id(raw)
|
||
self.assertEqual(ctx.exception.code, "bad_farm")
|
||
|
||
def test_other_illegal_shapes_are_refused(self):
|
||
for raw in ("a" * 25, "a b", "a.b", "a:b", "a*b", "牧场\n一号", "\x00"):
|
||
with self.subTest(raw=raw):
|
||
with self.assertRaises(GameError) as ctx:
|
||
validate_farm_id(raw)
|
||
self.assertEqual(ctx.exception.code, "bad_farm")
|
||
self.assertEqual(ctx.exception.status, 400)
|
||
|
||
def test_non_scalar_is_refused(self):
|
||
with self.assertRaises(GameError) as ctx:
|
||
validate_farm_id(["alice"])
|
||
self.assertEqual(ctx.exception.code, "bad_farm")
|
||
|
||
|
||
class OwnerTestCase(unittest.TestCase):
|
||
def test_missing_owner_is_empty_string(self):
|
||
self.assertEqual(validate_owner(None), "")
|
||
|
||
def test_scalars_are_accepted_and_trimmed(self):
|
||
self.assertEqual(validate_owner(" alice "), "alice")
|
||
self.assertEqual(validate_owner(42), "42")
|
||
|
||
def test_overlong_owner_is_refused(self):
|
||
with self.assertRaises(GameError) as ctx:
|
||
validate_owner("x" * 33)
|
||
self.assertEqual(ctx.exception.code, "bad_param")
|
||
|
||
def test_containers_are_refused(self):
|
||
with self.assertRaises(GameError):
|
||
validate_owner(["alice"])
|
||
|
||
|
||
class FarmSessionTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.dir = tempfile.mkdtemp(prefix="simplefarm-session-")
|
||
|
||
def test_clock_is_scheduled_immediately(self):
|
||
"""回归：不设 next_tick_at 的话，新建牧场头 0.2 秒里倒计时是 None，
|
||
客户端会把它显示成"已暂停 / 手动模式"。"""
|
||
session = FarmSession("alice", self.dir, 2400.0)
|
||
self.assertIsNotNone(session.engine.next_tick_at)
|
||
self.assertIsNotNone(session.snapshot()["seconds_to_tick"])
|
||
self.assertGreater(session.snapshot()["seconds_to_tick"], 0)
|
||
|
||
def test_snapshot_and_summary_carry_both_layers(self):
|
||
session = FarmSession("alice", self.dir, 2400.0, owner="爱丽丝")
|
||
snap = session.snapshot()
|
||
self.assertEqual(snap["farm"], "alice")
|
||
self.assertEqual(snap["owner"], "爱丽丝")
|
||
self.assertEqual(snap["tick_seconds"], 2400.0)
|
||
|
||
summary = session.summary(online=2)
|
||
self.assertEqual(summary["farm"], "alice")
|
||
self.assertEqual(summary["online"], 2)
|
||
self.assertIn("plots", summary)
|
||
self.assertIn("idle_seconds", summary)
|
||
|
||
def test_owner_defaults_to_the_farm_id(self):
|
||
session = FarmSession("alice", self.dir, 2400.0)
|
||
self.assertEqual(session.owner, "alice")
|
||
|
||
def test_new_farm_writes_a_welcome_log(self):
|
||
session = FarmSession("alice", self.dir, 2400.0)
|
||
texts = [entry["text"] for entry in session.engine.state.log]
|
||
self.assertTrue(any("alice" in text for text in texts))
|
||
|
||
def test_save_then_reload_keeps_progress(self):
|
||
session = FarmSession("alice", self.dir, 2400.0)
|
||
session.engine.state.day = 9
|
||
session.engine.state.gold = 777
|
||
path = session.save()
|
||
self.assertTrue(os.path.exists(path))
|
||
|
||
reopened = FarmSession("alice", self.dir, 2400.0)
|
||
self.assertEqual(reopened.engine.state.day, 9)
|
||
self.assertEqual(reopened.engine.state.gold, 777)
|
||
self.assertIn("第 9 天", reopened.load_message)
|
||
|
||
def test_autosave_survives_the_push_loop_clearing_dirty(self):
|
||
"""回归：dirty 是推送节流用的，Publisher ~250ms 就清一次；自动存档
|
||
如果也看 dirty，就等于永不存盘（进程被强杀进度全丢）。它必须看
|
||
只有成功的 save() 才清的 unsaved。"""
|
||
session = FarmSession("alice", self.dir, 2400.0)
|
||
session.touch()
|
||
session.dirty = False # 模拟 Publisher 清推送标记
|
||
self.assertTrue(session.autosave(), "推送循环清了 dirty 也要照常落盘")
|
||
reopened = FarmSession("alice", self.dir, 2400.0)
|
||
self.assertIn("已载入存档", reopened.load_message)
|
||
|
||
def test_autosave_skips_when_nothing_changed_since_last_save(self):
|
||
session = FarmSession("alice", self.dir, 2400.0)
|
||
session.touch()
|
||
self.assertTrue(session.autosave())
|
||
self.assertFalse(session.autosave(), "没有新变化就不该重复写库")
|
||
|
||
def test_broken_save_degrades_to_a_new_farm(self):
|
||
path = os.path.join(self.dir, DB_NAME)
|
||
with open(path, "w", encoding="utf-8") as handle:
|
||
handle.write("这不是 SQLite")
|
||
session = FarmSession("alice", self.dir, 2400.0)
|
||
self.assertEqual(session.engine.state.day, 1)
|
||
self.assertIn("已开新档", session.load_message)
|
||
|
||
def test_reset_clears_progress_and_counts_as_an_action(self):
|
||
session = FarmSession("alice", self.dir, 2400.0)
|
||
session.engine.state.day = 5
|
||
session.reset()
|
||
self.assertEqual(session.engine.state.day, 1)
|
||
self.assertEqual(session.actions_taken, 1)
|
||
|
||
def test_advance_day_remembers_the_events(self):
|
||
session = FarmSession("alice", self.dir, 2400.0)
|
||
session.engine.plant(1, "wheat")
|
||
session.advance_day("测试")
|
||
self.assertEqual(session.engine.state.day, 2)
|
||
self.assertIsInstance(session.last_events, list)
|
||
|
||
def test_touch_marks_dirty_and_counts_actions(self):
|
||
session = FarmSession("alice", self.dir, 2400.0)
|
||
session.dirty = False
|
||
session.touch(count_action=True)
|
||
self.assertTrue(session.dirty)
|
||
self.assertEqual(session.actions_taken, 1)
|
||
|
||
def test_autosave_skips_when_clean(self):
|
||
session = FarmSession("alice", self.dir, 2400.0)
|
||
session.unsaved = False
|
||
session.saved_at = 0.0
|
||
self.assertFalse(session.autosave())
|
||
|
||
def test_autosave_writes_when_unsaved(self):
|
||
session = FarmSession("alice", self.dir, 2400.0)
|
||
session.unsaved = True
|
||
session.saved_at = 0.0
|
||
self.assertTrue(session.autosave())
|
||
self.assertTrue(farm_exists(session.db_path, session.farm_id))
|
||
|
||
def test_autosave_waits_between_writes(self):
|
||
session = FarmSession("alice", self.dir, 2400.0)
|
||
session.dirty = True
|
||
session.saved_at = time.time()
|
||
self.assertFalse(session.autosave(), "刚存过就不用再存")
|
||
|
||
def test_state_message_shape(self):
|
||
session = FarmSession("alice", self.dir, 2400.0)
|
||
message = session.state_message(seq=7, note="给你")
|
||
self.assertEqual(message["type"], "state")
|
||
self.assertEqual(message["seq"], 7)
|
||
self.assertEqual(message["farm"], "alice")
|
||
self.assertEqual(message["note"], "给你")
|
||
self.assertIn("state", message)
|
||
self.assertIn("log", message)
|
||
|
||
|
||
class SessionManagerTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.dir = tempfile.mkdtemp(prefix="simplefarm-manager-")
|
||
|
||
def manager(self, **kwargs) -> SessionManager:
|
||
return SessionManager(save_dir=self.dir, tick_seconds=2400.0, **kwargs)
|
||
|
||
def test_get_creates_and_caches(self):
|
||
manager = self.manager()
|
||
first = manager.get("alice")
|
||
second = manager.get("alice")
|
||
self.assertIs(first, second)
|
||
self.assertEqual(manager.count(), 1)
|
||
self.assertEqual(manager.ids(), ["alice"])
|
||
|
||
def test_get_without_create_reports_404(self):
|
||
manager = self.manager()
|
||
with self.assertRaises(GameError) as ctx:
|
||
manager.get("ghost", create=False)
|
||
self.assertEqual(ctx.exception.code, "no_such_farm")
|
||
self.assertEqual(ctx.exception.status, 404)
|
||
|
||
def test_find_does_not_create(self):
|
||
manager = self.manager()
|
||
self.assertIsNone(manager.find("ghost"))
|
||
manager.get("alice")
|
||
self.assertIsNotNone(manager.find("alice"))
|
||
|
||
def test_capacity_limit_is_503(self):
|
||
manager = self.manager(max_sessions=2)
|
||
manager.get("a")
|
||
manager.get("b")
|
||
with self.assertRaises(GameError) as ctx:
|
||
manager.get("c")
|
||
self.assertEqual(ctx.exception.code, "too_many_farms")
|
||
self.assertEqual(ctx.exception.status, 503)
|
||
|
||
def test_farms_are_isolated_from_each_other(self):
|
||
manager = self.manager()
|
||
alice = manager.get("alice")
|
||
bob = manager.get("bob")
|
||
alice.engine.plant(1, "wheat")
|
||
alice.engine.state.gold = 999
|
||
self.assertTrue(bob.engine.state.plots[0].is_empty)
|
||
self.assertEqual(bob.engine.state.gold, config.START_GOLD)
|
||
|
||
def test_leaderboard_sorts_by_net_worth(self):
|
||
manager = self.manager()
|
||
rich = manager.get("rich")
|
||
manager.get("poor")
|
||
rich.engine.state.gold = 5000
|
||
rows = manager.leaderboard()
|
||
self.assertEqual(rows[0]["farm"], "rich")
|
||
self.assertEqual([row["rank"] for row in rows], [1, 2])
|
||
self.assertGreater(rows[0]["net_worth"], rows[1]["net_worth"])
|
||
self.assertGreaterEqual(rows[0]["net_worth"], 5000.0)
|
||
|
||
def test_leaderboard_shows_online_counts(self):
|
||
manager = self.manager()
|
||
manager.get("alice")
|
||
rows = manager.leaderboard({"alice": 3})
|
||
self.assertEqual(rows[0]["online"], 3)
|
||
|
||
def test_reap_drops_idle_farms_after_saving(self):
|
||
manager = self.manager(idle_timeout=0.01)
|
||
session = manager.get("alice")
|
||
session.engine.state.gold = 321
|
||
session.last_active = time.time() - 10
|
||
|
||
dropped = manager.reap()
|
||
self.assertEqual(dropped, ["alice"])
|
||
self.assertEqual(manager.count(), 0)
|
||
self.assertTrue(farm_exists(session.db_path, session.farm_id), "回收前必须先落盘")
|
||
|
||
def test_reap_keeps_farms_that_are_still_fresh(self):
|
||
manager = self.manager(idle_timeout=3600)
|
||
manager.get("alice")
|
||
self.assertEqual(manager.reap(), [])
|
||
self.assertEqual(manager.count(), 1)
|
||
|
||
def test_reap_keeps_farms_with_viewers(self):
|
||
manager = self.manager(idle_timeout=0.01)
|
||
manager.get("alice")
|
||
session = manager.get("alice")
|
||
session.last_active = time.time() - 10
|
||
self.assertEqual(manager.reap({"alice": 1}), [])
|
||
self.assertEqual(manager.count(), 1)
|
||
|
||
def test_reap_keeps_paused_farms(self):
|
||
"""暂停是人为操作，说明玩家还打算回来。"""
|
||
manager = self.manager(idle_timeout=0.01)
|
||
session = manager.get("alice")
|
||
session.engine.pause(True)
|
||
session.last_active = time.time() - 10
|
||
self.assertEqual(manager.reap(), [])
|
||
self.assertEqual(manager.count(), 1)
|
||
|
||
def test_autosave_all_reports_how_many_were_written(self):
|
||
manager = self.manager()
|
||
manager.get("alice")
|
||
bob = manager.get("bob")
|
||
alice = manager.get("alice")
|
||
alice.unsaved = True
|
||
bob.unsaved = False
|
||
alice.saved_at = bob.saved_at = 0.0
|
||
self.assertEqual(manager.autosave_all(), 1)
|
||
|
||
def test_shutdown_saves_every_farm(self):
|
||
manager = self.manager()
|
||
for name in ("alice", "bob"):
|
||
manager.get(name)
|
||
manager.shutdown()
|
||
for name in ("alice", "bob"):
|
||
self.assertTrue(farm_exists(os.path.join(self.dir, DB_NAME), name))
|
||
|
||
|
||
class FarmClockTestCase(unittest.TestCase):
|
||
"""多牧场共用一个时钟线程——不是每个牧场一个。"""
|
||
|
||
def setUp(self):
|
||
self.dir = tempfile.mkdtemp(prefix="simplefarm-clock-")
|
||
self.manager = SessionManager(save_dir=self.dir, tick_seconds=2400.0)
|
||
|
||
def test_one_thread_advances_every_farm(self):
|
||
alice = self.manager.get("alice")
|
||
bob = self.manager.get("bob")
|
||
for session in (alice, bob):
|
||
session.engine.tick_seconds = 0.05 # 绕过 clamp，把一天压到 50ms
|
||
session.engine.next_tick_at = time.time() + session.engine.tick_seconds
|
||
|
||
clock = FarmClock(self.manager, accuracy=0.02)
|
||
clock.start()
|
||
try:
|
||
advanced = wait_until(
|
||
lambda: alice.engine.state.day >= 2 and bob.engine.state.day >= 2, timeout=4.0
|
||
)
|
||
finally:
|
||
clock.stop()
|
||
self.assertTrue(advanced, "两个牧场都应该自己往前走")
|
||
self.assertGreater(clock.ticks, 0)
|
||
|
||
def test_paused_farm_stays_put_while_others_run(self):
|
||
frozen = self.manager.get("frozen")
|
||
running = self.manager.get("running")
|
||
frozen.engine.pause(True)
|
||
for session in (frozen, running):
|
||
session.engine.tick_seconds = 0.05
|
||
session.engine.next_tick_at = time.time() + 0.05
|
||
|
||
clock = FarmClock(self.manager, accuracy=0.02)
|
||
clock.start()
|
||
try:
|
||
advanced = wait_until(lambda: running.engine.state.day >= 2, timeout=4.0)
|
||
finally:
|
||
clock.stop()
|
||
self.assertTrue(advanced)
|
||
self.assertEqual(frozen.engine.state.day, 1, "暂停的牧场不该被推进")
|
||
|
||
def test_stats_reports_ticks_and_accuracy(self):
|
||
clock = FarmClock(self.manager, accuracy=0.25)
|
||
self.assertEqual(clock.stats(), (0, 0.25))
|
||
|
||
|
||
class HubTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.hub = ConnectionHub()
|
||
|
||
def client(self, farm: str = "alice", broken: bool = False) -> Client:
|
||
return Client(farm, FakeLoop(broken=broken), name="tester")
|
||
|
||
def test_register_and_online_counts(self):
|
||
alice = self.client("alice")
|
||
bob = self.client("bob")
|
||
self.hub.register(alice)
|
||
self.hub.register(bob)
|
||
self.assertEqual(self.hub.online("alice"), 1)
|
||
self.assertEqual(self.hub.total(), 2)
|
||
self.assertEqual(self.hub.online_map(), {"alice": 1, "bob": 1})
|
||
|
||
def test_unregister_clears_the_empty_bucket(self):
|
||
alice = self.client("alice")
|
||
self.hub.register(alice)
|
||
self.hub.unregister(alice)
|
||
self.assertEqual(self.hub.online("alice"), 0)
|
||
self.assertFalse(alice.alive)
|
||
self.assertEqual(self.hub.online_map(), {})
|
||
|
||
def test_broadcast_stays_inside_one_farm(self):
|
||
alice = self.client("alice")
|
||
bob = self.client("bob")
|
||
self.hub.register(alice)
|
||
self.hub.register(bob)
|
||
sent = self.hub.broadcast("alice", {"type": "state"})
|
||
self.assertEqual(sent, 1)
|
||
self.assertEqual(alice.queue.qsize(), 1)
|
||
self.assertEqual(bob.queue.qsize(), 0, "不能把 A 的状态推给 B")
|
||
|
||
def test_broadcast_can_exclude_the_sender(self):
|
||
alice = self.client("alice")
|
||
other = self.client("alice")
|
||
self.hub.register(alice)
|
||
self.hub.register(other)
|
||
self.hub.broadcast("alice", {"type": "notice"}, exclude=alice)
|
||
self.assertEqual(alice.queue.qsize(), 0)
|
||
self.assertEqual(other.queue.qsize(), 1)
|
||
|
||
def test_broadcast_all_reaches_every_farm(self):
|
||
self.hub.register(self.client("alice"))
|
||
self.hub.register(self.client("bob"))
|
||
self.assertEqual(self.hub.broadcast_all({"type": "notice"}), 2)
|
||
|
||
def test_close_all_pushes_the_bye_message(self):
|
||
alice = self.client("alice")
|
||
self.hub.register(alice)
|
||
self.assertEqual(self.hub.close_all(), 1)
|
||
self.assertEqual(alice.queue.get_nowait()["type"], "bye")
|
||
self.assertEqual(CLOSE_MESSAGE["type"], "bye")
|
||
|
||
def test_snapshot_lists_connection_stats(self):
|
||
client = self.client("alice")
|
||
self.hub.register(client)
|
||
rows = self.hub.snapshot()
|
||
self.assertEqual(len(rows), 1)
|
||
self.assertEqual(rows[0]["client_id"], client.cid)
|
||
self.assertEqual(rows[0]["farm"], "alice")
|
||
self.assertIn("online_seconds", rows[0])
|
||
|
||
def test_push_is_dropped_when_the_loop_is_gone(self):
|
||
client = self.client("alice", broken=True)
|
||
client.push({"type": "state"})
|
||
self.assertFalse(client.alive, "循环已关闭就该老实标记自己死了")
|
||
|
||
def test_push_after_unregister_is_a_no_op(self):
|
||
client = self.client("alice")
|
||
self.hub.register(client)
|
||
self.hub.unregister(client)
|
||
client.push({"type": "state"})
|
||
self.assertEqual(client.queue.qsize(), 0)
|
||
|
||
def test_full_queue_drops_the_oldest_message(self):
|
||
"""实时状态宁可少一帧，也不能把连接拖死。"""
|
||
from farm.server import hub as hub_mod
|
||
|
||
client = self.client("alice")
|
||
for i in range(hub_mod.QUEUE_SIZE + 5):
|
||
client._enqueue({"type": "state", "seq": i})
|
||
self.assertEqual(client.queue.qsize(), hub_mod.QUEUE_SIZE)
|
||
self.assertEqual(client.queue.get_nowait()["seq"], 5, "最旧的应被丢掉")
|
||
self.assertGreater(client.stats()["dropped"], 0)
|
||
|
||
|
||
class SseProtocolTestCase(unittest.TestCase):
|
||
def test_frame_is_single_line_json(self):
|
||
frame = sse_frame({"type": "state", "text": "你好"})
|
||
self.assertTrue(frame.startswith("data: "))
|
||
self.assertTrue(frame.endswith("\n\n"))
|
||
payload = json.loads(frame[len("data: ") :].strip())
|
||
self.assertEqual(payload["text"], "你好")
|
||
self.assertNotIn("\\u", frame, "中文不该被转义成 \\uXXXX")
|
||
|
||
def test_newlines_inside_values_stay_escaped(self):
|
||
"""SSE 靠空行分帧，所以 JSON 里绝不能出现真换行。"""
|
||
frame = sse_frame({"text": "第一行\n第二行"})
|
||
body = frame[len("data: ") : -2]
|
||
self.assertEqual(len(body.splitlines()), 1)
|
||
|
||
def test_heartbeat_is_a_comment_line(self):
|
||
self.assertEqual(sse_heartbeat(), ": keep-alive\n\n")
|
||
|
||
def test_sequence_numbers_only_go_up(self):
|
||
values = [next_seq() for _ in range(50)]
|
||
self.assertEqual(values, sorted(values))
|
||
self.assertEqual(len(set(values)), len(values))
|
||
|
||
def test_state_message_is_json_serializable(self):
|
||
session = FarmSession("alice", tempfile.mkdtemp(prefix="simplefarm-msg-"), 2400.0)
|
||
message = build_state_message(session, next_seq(), extra={"online": 1})
|
||
json.dumps(message, ensure_ascii=False)
|
||
self.assertEqual(message["online"], 1)
|
||
self.assertEqual(message["type"], "state")
|
||
|
||
|
||
class PublisherTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.dir = tempfile.mkdtemp(prefix="simplefarm-pub-")
|
||
self.manager = SessionManager(save_dir=self.dir, tick_seconds=2400.0)
|
||
self.hub = ConnectionHub()
|
||
self.publisher = Publisher(self.manager, self.hub, autosave_interval=3600.0)
|
||
self.session = self.manager.get("alice")
|
||
|
||
def attach(self) -> Client:
|
||
client = Client("alice", FakeLoop(), name="tui")
|
||
self.hub.register(client)
|
||
return client
|
||
|
||
def test_push_state_reaches_the_viewer(self):
|
||
client = self.attach()
|
||
self.publisher._push_state(self.session, time.time())
|
||
message = client.queue.get_nowait()
|
||
self.assertEqual(message["type"], "state")
|
||
self.assertEqual(message["online"], 1)
|
||
self.assertEqual(message["state"]["farm"], "alice")
|
||
self.assertEqual(self.publisher.pushes, 1)
|
||
|
||
def test_crossing_a_day_pushes_a_day_message_first(self):
|
||
client = self.attach()
|
||
self.session.engine.state.day = 2
|
||
self.publisher._push_state(self.session, time.time())
|
||
first = client.queue.get_nowait()
|
||
second = client.queue.get_nowait()
|
||
self.assertEqual(first["type"], "day")
|
||
self.assertEqual(first["day"], 2)
|
||
self.assertEqual(second["type"], "state")
|
||
|
||
def test_day_message_is_only_sent_once_per_day(self):
|
||
client = self.attach()
|
||
self.session.engine.state.day = 2
|
||
self.publisher._push_state(self.session, time.time())
|
||
while not client.queue.empty():
|
||
client.queue.get_nowait()
|
||
self.publisher._push_state(self.session, time.time())
|
||
self.assertEqual(client.queue.get_nowait()["type"], "state")
|
||
|
||
def test_errors_are_counted_instead_of_crashing(self):
|
||
"""推送线程绝不能因为单个牧场出错而退出。"""
|
||
|
||
class Exploding:
|
||
def all(self):
|
||
raise RuntimeError("牧场数据坏了")
|
||
|
||
publisher = Publisher(Exploding(), self.hub, autosave_interval=3600.0)
|
||
publisher.start()
|
||
try:
|
||
self.assertTrue(wait_until(lambda: publisher.errors > 0, timeout=3.0))
|
||
self.assertTrue(publisher.is_alive(), "出了错也要继续活着")
|
||
finally:
|
||
publisher.stop()
|
||
|
||
def test_uptime_is_a_number(self):
|
||
self.assertGreaterEqual(self.publisher.uptime(), 0.0)
|
||
|
||
|
||
class ServerHandleTestCase(unittest.TestCase):
|
||
def test_start_callbacks_all_run(self):
|
||
handle = ServerHandle()
|
||
seen: list[str] = []
|
||
handle.on_start(lambda: seen.append("a"))
|
||
handle.on_start(lambda: seen.append("b"))
|
||
handle.notify_start()
|
||
self.assertEqual(seen, ["a", "b"])
|
||
|
||
def test_request_stop_is_accepted_once(self):
|
||
handle = ServerHandle()
|
||
calls: list[str] = []
|
||
handle.on_stop(lambda: calls.append("stop"))
|
||
self.assertTrue(handle.request_stop())
|
||
self.assertEqual(calls, ["stop"])
|
||
self.assertFalse(handle.request_stop(), "重复请求要如实回答")
|
||
self.assertEqual(calls, ["stop"])
|
||
|
||
def test_a_failing_hook_does_not_block_the_others(self):
|
||
handle = ServerHandle()
|
||
calls: list[str] = []
|
||
|
||
def boom() -> None:
|
||
raise RuntimeError("钩子炸了")
|
||
|
||
handle.on_stop(boom)
|
||
handle.on_stop(lambda: calls.append("还是跑了"))
|
||
handle.on_start(boom)
|
||
handle.notify_start()
|
||
self.assertTrue(handle.request_stop())
|
||
self.assertEqual(calls, ["还是跑了"])
|
||
|
||
|
||
@unittest.skipUnless(HAS_FASTAPI, "未安装 FastAPI")
|
||
class BriefAndRulesTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.dir = tempfile.mkdtemp(prefix="simplefarm-brief-")
|
||
self.manager = SessionManager(save_dir=self.dir, tick_seconds=2400.0)
|
||
self.hub = ConnectionHub()
|
||
self.session = self.manager.get("alice")
|
||
|
||
def test_brief_keys_exist_in_the_brief_payload(self):
|
||
"""回归：brief_source 改字段名时，BRIEF_KEYS 必须跟着改，否则动作回执直接 500。
|
||
|
||
BRIEF_KEYS 的数据源是 ``FarmEngine.brief_source()``（精简状态），
|
||
**不是**完整快照 ``session.snapshot()``——后者包含 market/plots 等大块数据，
|
||
拿它来断言会把"brief 只读精简字段"这个设计意图测丢。
|
||
"""
|
||
data = brief(self.session, self.hub)
|
||
for key in BRIEF_KEYS:
|
||
self.assertIn(key, data, f"回执里没有 {key}")
|
||
|
||
def test_brief_exposes_the_to_do_lists(self):
|
||
self.session.engine.plant(1, "wheat")
|
||
self.session.engine.state.plots[0].growth = 2
|
||
self.session.engine.buy_animal("chicken")
|
||
data = brief(self.session, self.hub)
|
||
self.assertEqual(data["farm"], "alice")
|
||
self.assertEqual(data["ripe_plots"], [1])
|
||
self.assertEqual(len(data["hungry_animals"]), 1)
|
||
self.assertEqual(data["collectable_animals"], [])
|
||
self.assertEqual(data["online"], 0)
|
||
|
||
def test_brief_is_json_serializable(self):
|
||
json.dumps(brief(self.session, self.hub), ensure_ascii=False)
|
||
|
||
def test_rules_no_longer_mention_sleeping(self):
|
||
"""「休息」已经整体移除，给 AI 的说明书不能还留着它。"""
|
||
text = rules_text()
|
||
self.assertNotIn("sleep", text)
|
||
self.assertNotIn("休息", text)
|
||
self.assertNotIn("快进", text.replace("没有快进", ""))
|
||
self.assertIn("40 分钟", text)
|
||
|
||
def test_rules_point_at_the_recommended_loop(self):
|
||
text = rules_text()
|
||
self.assertIn("/api/actions", text)
|
||
self.assertIn("/api/stream", text)
|
||
|
||
|
||
@unittest.skipUnless(HAS_FASTAPI, "未安装 FastAPI")
|
||
class AppWiringTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.cfg = ServerConfig(
|
||
save_dir=tempfile.mkdtemp(prefix="simplefarm-app-"), tick_seconds=2400.0
|
||
)
|
||
self.app = create_app(self.cfg)
|
||
|
||
def test_app_state_is_fully_wired(self):
|
||
for name in ("manager", "hub", "publisher", "clock", "handle", "counters"):
|
||
self.assertTrue(hasattr(self.app.state, name), name)
|
||
|
||
def test_as_asgi_is_the_identity(self):
|
||
self.assertIs(as_asgi(self.app), self.app)
|
||
|
||
def test_default_config_matches_the_game_defaults(self):
|
||
cfg = ServerConfig()
|
||
self.assertEqual(cfg.host, "127.0.0.1")
|
||
self.assertEqual(cfg.tick_seconds, config.DEFAULT_TICK_SECONDS)
|
||
self.assertTrue(cfg.allow_remote_control)
|
||
|
||
def test_build_app_is_a_convenience_wrapper(self):
|
||
from farm.server.app import build_app
|
||
|
||
app = build_app(port=9999)
|
||
self.assertEqual(app.state.cfg.port, 9999)
|
||
self.assertEqual(app.state.cfg.host, "127.0.0.1")
|
||
|
||
def test_int_query_helper(self):
|
||
self.assertEqual(_int(None, 20, 1, 200), 20)
|
||
self.assertEqual(_int("", 20, 1, 200), 20)
|
||
self.assertEqual(_int("5", 20, 1, 200), 5)
|
||
self.assertEqual(_int("999", 20, 1, 200), 200)
|
||
self.assertEqual(_int("-5", 20, 1, 200), 1)
|
||
with self.assertRaises(GameError) as ctx:
|
||
_int("abc", 20, 1, 200)
|
||
self.assertEqual(ctx.exception.code, "bad_param")
|
||
|
||
|
||
class DaemonTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.dir = tempfile.mkdtemp(prefix="simplefarm-daemon-")
|
||
self.pid_file = os.path.join(self.dir, "server.pid")
|
||
self.log_file = os.path.join(self.dir, "server.log")
|
||
|
||
def manager(self, port: int) -> DaemonManager:
|
||
return DaemonManager(
|
||
host="127.0.0.1",
|
||
port=port,
|
||
save_dir=self.dir,
|
||
pid_file=self.pid_file,
|
||
log_file=self.log_file,
|
||
)
|
||
|
||
def test_own_process_is_alive_and_nonsense_pids_are_not(self):
|
||
self.assertTrue(pid_alive(os.getpid()))
|
||
self.assertFalse(pid_alive(0))
|
||
self.assertFalse(pid_alive(-1))
|
||
|
||
def test_port_busy_detects_a_listening_socket(self):
|
||
with socket.socket() as listener:
|
||
listener.bind(("127.0.0.1", 0))
|
||
listener.listen(16) # 至少要能容纳本用例的两连接
|
||
port = int(listener.getsockname()[1])
|
||
self.assertTrue(port_busy("127.0.0.1", port))
|
||
self.assertTrue(self.manager(port).port_busy())
|
||
self.assertFalse(port_busy("127.0.0.1", port), "监听关掉之后就该是空闲的")
|
||
|
||
def test_pid_file_round_trip(self):
|
||
manager = self.manager(free_port())
|
||
self.assertIsNone(manager.read_pid())
|
||
manager.write_pid(4321)
|
||
self.assertEqual(manager.read_pid(), 4321)
|
||
manager.clear_pid()
|
||
self.assertIsNone(manager.read_pid())
|
||
|
||
def test_stale_pid_file_is_cleaned_up(self):
|
||
manager = self.manager(free_port())
|
||
manager.write_pid(999_999_999) # 必定不存在的 pid
|
||
self.assertIsNone(manager.current())
|
||
self.assertIsNone(manager.read_pid(), "陈旧记录应当被顺手清掉")
|
||
|
||
def test_status_reports_not_running_for_a_dead_port(self):
|
||
info = self.manager(free_port()).status()
|
||
self.assertFalse(info["running"])
|
||
self.assertIsNone(info["pid"])
|
||
self.assertIsNone(info["health"])
|
||
|
||
def test_probe_and_world_return_none_without_a_server(self):
|
||
manager = self.manager(free_port())
|
||
self.assertIsNone(manager.probe())
|
||
self.assertIsNone(manager.world())
|
||
|
||
def test_command_passes_the_pid_file_through(self):
|
||
"""pid 必须由服务进程自己写：父进程拿到的 pid 在 Windows 上未必是监听者。"""
|
||
args = self.manager(8765).command()
|
||
self.assertIn("--pid-file", args)
|
||
self.assertEqual(args[args.index("--pid-file") + 1], self.pid_file)
|
||
|
||
def test_stop_on_idle_port_is_not_an_error(self):
|
||
ok, message = self.manager(free_port()).stop()
|
||
self.assertTrue(ok)
|
||
self.assertIn("没有运行", message)
|
||
|
||
def test_tail_returns_empty_for_a_missing_log(self):
|
||
self.assertEqual(self.manager(free_port()).tail(), [])
|
||
|
||
def test_tail_reads_the_last_lines(self):
|
||
with open(self.log_file, "w", encoding="utf-8") as handle:
|
||
handle.write("\n".join(f"第{i}行" for i in range(50)))
|
||
lines = self.manager(free_port()).tail(5)
|
||
self.assertEqual(len(lines), 5)
|
||
self.assertEqual(lines[-1], "第49行")
|
||
|
||
def test_ensure_parent_dir_tolerates_a_bare_file_name(self):
|
||
"""回归：路径没有目录部分时 dirname 返回空串，makedirs("") 直接崩。"""
|
||
ensure_parent_dir("server.pid") # 不该抛
|
||
ensure_parent_dir(os.path.join(self.dir, "a", "b", "server.pid"))
|
||
self.assertTrue(os.path.isdir(os.path.join(self.dir, "a", "b")))
|
||
|
||
def test_pid_file_without_a_directory_still_works(self):
|
||
"""``--pid-file server.pid`` 这种只给文件名的用法必须能起来。"""
|
||
manager = DaemonManager(
|
||
host="127.0.0.1",
|
||
port=free_port(),
|
||
save_dir=self.dir,
|
||
pid_file="server.pid",
|
||
log_file="server.log",
|
||
)
|
||
cwd = os.getcwd()
|
||
os.chdir(self.dir)
|
||
try:
|
||
manager.write_pid(4321)
|
||
self.assertEqual(manager.read_pid(), 4321)
|
||
manager.clear_pid()
|
||
self.assertIsNone(manager.read_pid())
|
||
finally:
|
||
os.chdir(cwd)
|
||
|
||
|
||
@unittest.skipUnless(HAS_TESTCLIENT, "未安装 httpx（TestClient 依赖它）")
|
||
class RestApiTestCase(unittest.TestCase):
|
||
"""进程内打接口：语义、鉴权、多牧场隔离。"""
|
||
|
||
def setUp(self):
|
||
self.dir = tempfile.mkdtemp(prefix="simplefarm-rest-")
|
||
self.app = create_app(ServerConfig(save_dir=self.dir, tick_seconds=2400.0))
|
||
self.client = TestClient(self.app)
|
||
|
||
def test_health_needs_no_token(self):
|
||
response = self.client.get("/api/health")
|
||
self.assertEqual(response.status_code, 200)
|
||
payload = response.json()
|
||
self.assertTrue(payload["ok"])
|
||
self.assertEqual(payload["service"], "SimpleFarm")
|
||
self.assertIn("clock_ticks", payload)
|
||
|
||
def test_state_and_catalog(self):
|
||
payload = self.client.get("/api/state?farm=alice").json()
|
||
self.assertEqual(payload["farm"], "alice")
|
||
self.assertEqual(payload["state"]["day"], 1)
|
||
self.assertEqual(payload["online"], 0)
|
||
|
||
catalog = self.client.get("/api/catalog?farm=alice").json()["catalog"]
|
||
self.assertEqual(len(catalog["crops"]), len(config.CROPS))
|
||
self.assertIn("ap_cost", catalog)
|
||
|
||
def test_action_then_read_it_back(self):
|
||
response = self.client.post(
|
||
"/api/actions?farm=alice",
|
||
json={"action": "plant", "params": {"plot": 1, "crop": "wheat"}},
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
payload = response.json()
|
||
self.assertTrue(payload["ok"])
|
||
self.assertEqual(payload["action"], "plant")
|
||
self.assertEqual(payload["state"]["plot_count"], config.BASE_PLOTS)
|
||
self.assertIn("seq", payload)
|
||
|
||
snap = self.client.get("/api/state?farm=alice").json()["state"]
|
||
self.assertEqual(snap["plots"][0]["crop"], "wheat")
|
||
|
||
def test_flat_parameters_are_accepted(self):
|
||
"""AI 常把参数平铺在顶层，接口对此要宽容。"""
|
||
payload = self.client.post(
|
||
"/api/actions?farm=alice", json={"action": "plant", "plot": 2, "crop": "wheat"}
|
||
).json()
|
||
self.assertTrue(payload["ok"])
|
||
|
||
def test_full_flag_returns_the_whole_snapshot(self):
|
||
payload = self.client.post(
|
||
"/api/actions?farm=alice&full=1", json={"action": "buy_feed", "params": {}}
|
||
).json()
|
||
self.assertIn("snapshot", payload)
|
||
self.assertIn("plots", payload["snapshot"])
|
||
|
||
def test_errors_are_machine_readable(self):
|
||
response = self.client.post("/api/actions?farm=alice", json={"action": "fly"})
|
||
self.assertEqual(response.status_code, 400)
|
||
payload = response.json()
|
||
self.assertFalse(payload["ok"])
|
||
self.assertEqual(payload["error"], "unknown_action")
|
||
|
||
def test_missing_parameter_is_a_400(self):
|
||
response = self.client.post(
|
||
"/api/actions?farm=alice", json={"action": "plant", "params": {"plot": 1}}
|
||
)
|
||
self.assertEqual(response.status_code, 400)
|
||
self.assertEqual(response.json()["error"], "missing_param")
|
||
|
||
def test_broken_json_body(self):
|
||
response = self.client.post(
|
||
"/api/actions?farm=alice",
|
||
content="{ not json",
|
||
headers={"Content-Type": "application/json"},
|
||
)
|
||
self.assertEqual(response.status_code, 400)
|
||
self.assertEqual(response.json()["error"], "bad_json")
|
||
|
||
def test_non_object_body(self):
|
||
response = self.client.post(
|
||
"/api/actions?farm=alice",
|
||
content="[1,2,3]",
|
||
headers={"Content-Type": "application/json"},
|
||
)
|
||
self.assertEqual(response.status_code, 400)
|
||
|
||
def test_farm_header_wins_over_the_default(self):
|
||
payload = self.client.get("/api/state", headers={"X-Farm-Id": "from-header"}).json()
|
||
self.assertEqual(payload["farm"], "from-header")
|
||
|
||
def test_bad_farm_id_is_refused(self):
|
||
response = self.client.get("/api/state?farm=../etc")
|
||
self.assertEqual(response.status_code, 400)
|
||
self.assertEqual(response.json()["error"], "bad_farm")
|
||
|
||
def test_batch_runs_every_action_and_reports_failures(self):
|
||
payload = self.client.post(
|
||
"/api/batch?farm=alice",
|
||
json={
|
||
"actions": [
|
||
{"action": "plant", "params": {"plot": 1, "crop": "wheat"}},
|
||
{"action": "plant", "params": {"plot": 1, "crop": "wheat"}}, # 这块地已占用
|
||
{"action": "buy_feed", "params": {"quantity": 5}},
|
||
]
|
||
},
|
||
).json()
|
||
self.assertTrue(payload["ok"])
|
||
self.assertEqual(payload["done"], 2)
|
||
self.assertEqual(payload["failed"], 1)
|
||
self.assertEqual(payload["results"][1]["error"], "plot_busy")
|
||
|
||
def test_batch_validates_its_container(self):
|
||
response = self.client.post("/api/batch?farm=alice", json={"actions": []})
|
||
self.assertEqual(response.status_code, 400)
|
||
response = self.client.post("/api/batch?farm=alice", json={"actions": "nope"})
|
||
self.assertEqual(response.status_code, 400)
|
||
|
||
def test_batch_has_an_upper_bound(self):
|
||
items = [{"action": "buy_feed", "params": {"quantity": 1}} for _ in range(51)]
|
||
response = self.client.post("/api/batch?farm=alice", json={"actions": items})
|
||
self.assertEqual(response.status_code, 400)
|
||
self.assertEqual(response.json()["error"], "too_many")
|
||
|
||
def test_generated_per_action_routes_work(self):
|
||
payload = self.client.post("/api/buy_feed?farm=alice", json={"quantity": 3}).json()
|
||
self.assertTrue(payload["ok"])
|
||
self.assertEqual(payload["action"], "buy_feed")
|
||
|
||
def test_log_supports_limits_and_incremental_reads(self):
|
||
self.client.post("/api/actions?farm=alice", json={"action": "buy_feed", "params": {}})
|
||
first = self.client.get("/api/log?farm=alice&limit=2").json()
|
||
self.assertLessEqual(first["count"], 2)
|
||
mark = first["last_seq"]
|
||
|
||
self.client.post("/api/actions?farm=alice", json={"action": "buy_feed", "params": {}})
|
||
newer = self.client.get(f"/api/log?farm=alice&since={mark}").json()
|
||
self.assertTrue(all(entry["seq"] > mark for entry in newer["entries"]))
|
||
|
||
def test_log_rejects_a_non_numeric_limit(self):
|
||
response = self.client.get("/api/log?farm=alice&limit=abc")
|
||
self.assertEqual(response.status_code, 400)
|
||
|
||
def test_farms_list_and_create(self):
|
||
created = self.client.post("/api/farms", json={"farm": "alice", "owner": "爱丽丝"}).json()
|
||
self.assertTrue(created["created"])
|
||
self.assertEqual(created["farm"], "alice")
|
||
self.assertEqual(created["state"]["farm"], "alice")
|
||
|
||
again = self.client.post("/api/farms", json={"farm": "alice"}).json()
|
||
self.assertFalse(again["created"])
|
||
|
||
rows = self.client.get("/api/farms").json()["farms"]
|
||
self.assertEqual(rows[0]["farm"], "alice")
|
||
|
||
def test_create_farm_can_tune_the_clock(self):
|
||
self.client.post("/api/farms", json={"farm": "fast", "tick": 5})
|
||
snap = self.client.get("/api/state?farm=fast").json()["state"]
|
||
self.assertEqual(snap["tick_seconds"], 5.0)
|
||
|
||
def test_create_farm_rejects_a_bad_tick(self):
|
||
response = self.client.post("/api/farms", json={"farm": "fast", "tick": "soon"})
|
||
self.assertEqual(response.status_code, 400)
|
||
|
||
def test_world_view(self):
|
||
self.client.post("/api/farms", json={"farm": "alice"})
|
||
payload = self.client.get("/api/world").json()
|
||
self.assertEqual(payload["farms"], 1)
|
||
self.assertIn("leaderboard", payload)
|
||
self.assertIn("connections", payload)
|
||
|
||
def test_help_describes_the_actions(self):
|
||
payload = self.client.get("/api/help").json()
|
||
names = {item["action"] for item in payload["actions"]}
|
||
self.assertIn("plant", names)
|
||
self.assertNotIn("sleep", names)
|
||
|
||
def test_farms_do_not_leak_into_each_other(self):
|
||
self.client.post(
|
||
"/api/actions?farm=alice",
|
||
json={"action": "plant", "params": {"plot": 1, "crop": "wheat"}},
|
||
)
|
||
bob = self.client.get("/api/state?farm=bob").json()["state"]
|
||
self.assertIsNone(bob["plots"][0]["crop"])
|
||
|
||
def test_save_goes_through_the_session_not_the_default_path(self):
|
||
"""存档位置由会话决定；直落 engine.save() 会写错地方、覆盖别人的档。"""
|
||
self.client.post("/api/actions?farm=alice", json={"action": "save", "params": {}})
|
||
self.assertTrue(farm_exists(os.path.join(self.dir, DB_NAME), "alice"))
|
||
self.assertFalse(
|
||
os.path.exists(os.path.join(self.dir, "save.json")), "别在会话目录外另起炉灶"
|
||
)
|
||
|
||
def test_reset_only_touches_its_own_farm(self):
|
||
self.client.post(
|
||
"/api/actions?farm=alice",
|
||
json={"action": "plant", "params": {"plot": 1, "crop": "wheat"}},
|
||
)
|
||
self.client.post(
|
||
"/api/actions?farm=bob",
|
||
json={"action": "plant", "params": {"plot": 1, "crop": "wheat"}},
|
||
)
|
||
self.client.post("/api/actions?farm=alice", json={"action": "reset", "params": {}})
|
||
|
||
alice = self.client.get("/api/state?farm=alice").json()["state"]
|
||
bob = self.client.get("/api/state?farm=bob").json()["state"]
|
||
self.assertIsNone(alice["plots"][0]["crop"])
|
||
self.assertEqual(bob["plots"][0]["crop"], "wheat")
|
||
|
||
def test_admin_save_writes_every_farm(self):
|
||
self.client.post("/api/farms", json={"farm": "alice"})
|
||
self.client.post("/api/farms", json={"farm": "bob"})
|
||
payload = self.client.post("/api/admin/save").json()
|
||
self.assertEqual(sorted(payload["farms"]), ["alice", "bob"])
|
||
|
||
def test_sleep_endpoint_is_gone(self):
|
||
"""整条"手动推进一天"的链路都不该存在。"""
|
||
response = self.client.post("/api/actions?farm=alice", json={"action": "sleep"})
|
||
self.assertEqual(response.status_code, 400)
|
||
self.assertEqual(response.json()["error"], "unknown_action")
|
||
|
||
def test_dashboard_pages_render(self):
|
||
self.assertEqual(self.client.get("/").status_code, 200)
|
||
text = self.client.get("/text").text
|
||
self.assertIn("SimpleFarm", text)
|
||
|
||
|
||
@unittest.skipUnless(HAS_TESTCLIENT, "未安装 httpx（TestClient 依赖它）")
|
||
class AuthTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.dir = tempfile.mkdtemp(prefix="simplefarm-auth-")
|
||
self.app = create_app(ServerConfig(save_dir=self.dir, token="s3cret", tick_seconds=2400.0))
|
||
self.client = TestClient(self.app)
|
||
|
||
def test_health_stays_open(self):
|
||
self.assertEqual(self.client.get("/api/health").status_code, 200)
|
||
|
||
def test_state_needs_a_token(self):
|
||
response = self.client.get("/api/state")
|
||
self.assertEqual(response.status_code, 401)
|
||
self.assertEqual(response.json()["error"], "unauthorized")
|
||
|
||
def test_wrong_token_is_refused(self):
|
||
response = self.client.get("/api/state?token=nope")
|
||
self.assertEqual(response.status_code, 401)
|
||
|
||
def test_query_token_works(self):
|
||
self.assertEqual(self.client.get("/api/state?token=s3cret").status_code, 200)
|
||
|
||
def test_header_token_works(self):
|
||
response = self.client.get("/api/state", headers={"X-API-Token": "s3cret"})
|
||
self.assertEqual(response.status_code, 200)
|
||
|
||
def test_writes_are_guarded_too(self):
|
||
response = self.client.post("/api/actions?farm=alice", json={"action": "buy_feed"})
|
||
self.assertEqual(response.status_code, 401)
|
||
|
||
def test_remote_shutdown_can_be_switched_off(self):
|
||
app = create_app(
|
||
ServerConfig(
|
||
save_dir=self.dir, token="s3cret", allow_remote_control=False, tick_seconds=2400.0
|
||
)
|
||
)
|
||
client = TestClient(app)
|
||
response = client.post("/api/admin/shutdown?token=s3cret")
|
||
self.assertEqual(response.status_code, 403)
|
||
|
||
|
||
@unittest.skipUnless(HAS_UVICORN and HAS_FASTAPI, "未安装 uvicorn / FastAPI")
|
||
class LiveServerTestCase(unittest.TestCase):
|
||
"""真起一个服务器进程内的 uvicorn，用真 socket 说话。"""
|
||
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
cls.dir = tempfile.mkdtemp(prefix="simplefarm-live-")
|
||
cls.port = free_port()
|
||
cls.cfg = ServerConfig(
|
||
host="127.0.0.1", port=cls.port, tick_seconds=config.MIN_TICK_SECONDS, save_dir=cls.dir
|
||
)
|
||
cls.app = create_app(cls.cfg)
|
||
config_obj = uvicorn.Config(
|
||
as_asgi(cls.app),
|
||
host="127.0.0.1",
|
||
port=cls.port,
|
||
log_level="warning",
|
||
http="zttp",
|
||
http2=True,
|
||
timeout_graceful_shutdown=3,
|
||
)
|
||
cls.server = uvicorn.Server(config_obj)
|
||
# __main__ 里做的那件事：停机请求要能把 should_exit 置位
|
||
cls.app.state.handle.on_stop(lambda: setattr(cls.server, "should_exit", True))
|
||
cls.thread = threading.Thread(target=cls.server.run, name="test-live-server", daemon=True)
|
||
cls.thread.start()
|
||
if not wait_until(cls.healthy, timeout=20.0):
|
||
raise unittest.SkipTest("服务器没能在 20 秒内起来")
|
||
|
||
@classmethod
|
||
def healthy(cls) -> bool:
|
||
try:
|
||
with net.open_url(f"http://127.0.0.1:{cls.port}/api/health", timeout=1.0) as response:
|
||
json.loads(response.read().decode("utf-8"))
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
@classmethod
|
||
def tearDownClass(cls):
|
||
cls.app.state.handle.request_stop()
|
||
cls.thread.join(timeout=8.0)
|
||
|
||
def endpoint(self) -> str:
|
||
return f"http://127.0.0.1:{self.port}"
|
||
|
||
def connect(self, farm: str, realtime: bool = False):
|
||
from farm.client import RemoteFarm
|
||
|
||
return RemoteFarm(endpoint=self.endpoint(), farm=farm, realtime=realtime, timeout=5.0)
|
||
|
||
def read_stream(self, farm: str, frames: int = 1, timeout: float = 8.0) -> list[dict]:
|
||
"""裸读 SSE 流，取到指定帧数就断开——顺便验证协议本身。"""
|
||
url = f"{self.endpoint()}/api/stream?farm={farm}"
|
||
request = urllib.request.Request(url, headers={"Accept": "text/event-stream"})
|
||
collected: list[dict] = []
|
||
with net.open_url(request, timeout=timeout) as response:
|
||
content_type = response.headers.get("content-type") or ""
|
||
self.assertTrue(content_type.startswith("text/event-stream"), content_type)
|
||
for raw in response:
|
||
line = raw.decode("utf-8", errors="replace").strip()
|
||
if not line.startswith("data:"):
|
||
continue
|
||
collected.append(json.loads(line[len("data:") :].strip()))
|
||
if len(collected) >= frames:
|
||
break
|
||
return collected
|
||
|
||
def test_http_round_trip(self):
|
||
client = self.connect("live-a")
|
||
self.assertTrue(client.connected, client.last_error)
|
||
before = client.state.gold
|
||
client.plant(1, "wheat")
|
||
self.assertEqual(client.state.plots[0].crop, "wheat")
|
||
self.assertLess(client.state.gold, before)
|
||
|
||
def test_two_clients_share_one_farm(self):
|
||
first = self.connect("live-shared")
|
||
second = self.connect("live-shared")
|
||
first.buy_feed(4)
|
||
self.assertTrue(wait_until(lambda: second.refresh() and second.state.feed >= 9))
|
||
self.assertEqual(second.state.feed, first.state.feed)
|
||
|
||
def test_other_farms_stay_separate(self):
|
||
alice = self.connect("live-alice")
|
||
bob = self.connect("live-bob")
|
||
alice.plant(1, "wheat")
|
||
bob.refresh()
|
||
self.assertIsNone(bob.state.plots[0].crop)
|
||
|
||
def test_sse_stream_opens_with_a_welcome_frame(self):
|
||
frames = self.read_stream("live-sse", frames=1)
|
||
welcome = frames[0]
|
||
self.assertEqual(welcome["type"], "welcome")
|
||
self.assertEqual(welcome["farm"], "live-sse")
|
||
self.assertIn("catalog", welcome)
|
||
self.assertIn("actions", welcome)
|
||
self.assertEqual(welcome["server"]["version"], self.app.version)
|
||
|
||
def test_sse_pushes_what_another_client_did(self):
|
||
listener = self.connect("live-push", realtime=True)
|
||
try:
|
||
self.assertTrue(
|
||
wait_until(lambda: listener.mode == "sse", timeout=8.0), listener.last_error
|
||
)
|
||
actor = self.connect("live-push")
|
||
actor.buy_feed(6)
|
||
self.assertTrue(
|
||
wait_until(lambda: listener.state.feed >= 9, timeout=8.0),
|
||
"另一个客户端的操作应当被推过来",
|
||
)
|
||
finally:
|
||
listener.stop()
|
||
|
||
def test_writes_keep_working_after_the_stream_is_cut(self):
|
||
client = self.connect("live-cut", realtime=True)
|
||
self.assertTrue(wait_until(lambda: client.mode == "sse", timeout=8.0))
|
||
client.stop()
|
||
client.buy_feed(2)
|
||
self.assertGreaterEqual(client.state.feed, 7)
|
||
|
||
def test_bot_can_run_purely_over_the_wire(self):
|
||
from farm.bot import RuleBot
|
||
|
||
client = self.connect("live-bot")
|
||
bot = RuleBot(client, verbose=False, pause=0.0, name="tester")
|
||
summary = bot.run(rounds=6, idle_limit=1)
|
||
self.assertGreaterEqual(summary["actions"], 1)
|
||
self.assertEqual(summary["name"], "tester")
|
||
|
||
def test_health_reports_a_running_world(self):
|
||
with net.open_url(f"{self.endpoint()}/api/health", timeout=3.0) as response:
|
||
payload = json.loads(response.read().decode("utf-8"))
|
||
self.assertTrue(payload["ok"])
|
||
self.assertGreaterEqual(payload["farms"], 1)
|
||
|
||
def test_second_instance_on_the_same_port_refuses_to_start(self):
|
||
"""端口自检必须用"能不能连上"判断，否则会静默顶掉已有实例。"""
|
||
result = subprocess.run(
|
||
[sys.executable, "-m", "farm.server", "--port", str(self.port)],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=60,
|
||
cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||
)
|
||
self.assertEqual(result.returncode, 3, result.stdout + result.stderr)
|
||
self.assertIn("已经有程序在监听", result.stdout + result.stderr)
|
||
|
||
def test_graceful_shutdown_writes_every_farm(self):
|
||
"""停机路径必须落盘。这里只验证"再启一个实例仍能读到存档"的前置条件。"""
|
||
client = self.connect("live-save")
|
||
client.plant(1, "wheat")
|
||
client.save()
|
||
self.assertTrue(farm_exists(os.path.join(self.dir, DB_NAME), "live-save"))
|
||
|
||
|
||
class SessionIsolationAcrossProcessesTestCase(unittest.TestCase):
|
||
"""守护进程的 pid 文件语义——不启动真服务，只验证文件层的约定。"""
|
||
|
||
def test_clear_pid_only_removes_our_own_record(self):
|
||
from farm.server.__main__ import clear_pid_file
|
||
|
||
directory = tempfile.mkdtemp(prefix="simplefarm-pid-")
|
||
path = os.path.join(directory, "server.pid")
|
||
|
||
with open(path, "w", encoding="utf-8") as handle:
|
||
handle.write("424242")
|
||
clear_pid_file(path)
|
||
self.assertTrue(os.path.exists(path), "不是自己写的 pid 文件不能删")
|
||
|
||
with open(path, "w", encoding="utf-8") as handle:
|
||
handle.write(str(os.getpid()))
|
||
clear_pid_file(path)
|
||
self.assertFalse(os.path.exists(path))
|
||
|
||
def test_clear_pid_tolerates_missing_files(self):
|
||
from farm.server.__main__ import clear_pid_file
|
||
|
||
clear_pid_file(None)
|
||
clear_pid_file(os.path.join(tempfile.mkdtemp(), "nothing-here.pid"))
|
||
|
||
def test_tick_is_clamped_by_the_entry_point(self):
|
||
from farm.server.__main__ import clamp_tick
|
||
|
||
self.assertEqual(clamp_tick(0), config.MIN_TICK_SECONDS)
|
||
self.assertEqual(clamp_tick(10**9), config.MAX_TICK_SECONDS)
|
||
self.assertEqual(clamp_tick(900), 900.0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|