"""服务器层测试：牧场会话、调度时钟、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, ) 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_broken_save_degrades_to_a_new_farm(self): path = os.path.join(self.dir, "alice.json") with open(path, "w", encoding="utf-8") as handle: handle.write("这不是 JSON") 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.dirty = False session.saved_at = 0.0 self.assertFalse(session.autosave()) def test_autosave_writes_when_dirty(self): session = FarmSession("alice", self.dir, 2400.0) session.dirty = True session.saved_at = 0.0 self.assertTrue(session.autosave()) self.assertTrue(os.path.exists(session.path)) 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(os.path.exists(session.path), "回收前必须先落盘") 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.dirty = True bob.dirty = 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(os.path.exists(os.path.join(self.dir, f"{name}.json"))) 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(os.path.exists(os.path.join(self.dir, "alice.json"))) 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(os.path.exists(os.path.join(self.dir, "live-save.json"))) 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)