snowy-yang
15d290a60a
为什么：游戏的时间（40 分钟一天、只能自然流动）本身就是唯一的行动节奏 阀门，再叠一层"每天 10 点行动点"纯属重复惩罚——浇水还要体力既不拟真 也不好玩。删掉之后，稀缺的只剩时间本身，放置玩法的定位反而更纯粹。 改动范围（引擎到界面连根拔）： - config 删 START_AP/AP_COST；models 删 ap 字段与 max_ap，存档 payload 里的旧 ap 键被 from_dict 自然忽略； - engine 删 _spend_ap 与全部 8 个调用点、每日恢复逻辑与日志； available_actions 不再按行动点筛，cost 只剩金币； - storage 建表去掉 ap 列，老库在 connect() 里 ALTER TABLE DROP COLUMN 无痛迁移（实测 saves/farm.db 已自动迁完）； - bot 删 AP 分支（无 AP 后 available_actions 恒可用）；client/TUI/CLI/ README/帮助文本同步清理。 网页端顺带两件事： - "下一天"倒计时改为客户端本地走秒——SSE 只在有变化时推送，以前倒计时 只在推送瞬间跳变，现在每秒都在动（记基点 + 1s 定时器补齐）； - 字体优化：页面字体栈补齐 MiSans/HarmonyOS Sans/PingFang，正文字号 14→15，统计数字 tabular-nums；canvas 小字统一字体栈并加粗提对比度。 测试适配：删 2 个 AP 专属用例（无 AP 可测）、改名/收敛 5 个， 532 项全绿（skip 39），ruff/pyright 干净。真机验证：老库迁移、 统计条无行动点、倒计时逐秒递减（39:04→39:02）、动作不再报 no_ap。
989 lines
41 KiB
Python
989 lines
41 KiB
Python
"""远程客户端测试：视图语义、地址归一化、SSE 消息分发、以及 HTTP 传输层。
|
||
|
||
这里是「TUI 不需要知道自己在跟谁说话」这句承诺的兑现处：
|
||
``StateView`` 必须把服务器发来的两种东西区分开——**完整快照**（带 plots/animals）
|
||
要整体替换，**精简状态**（动作回执里那几个标量）只能做字段合并。搞反了的后果
|
||
很具体：播一次种，整片农田就从界面上消失了。
|
||
|
||
除 ``HttpTransportTestCase`` 会起一个纯标准库的回环桩服务外，其余全部离线；
|
||
桩服务只绑 127.0.0.1，不依赖 fastapi / httpx / uvicorn。
|
||
|
||
uv run python -m unittest tests.test_client -v
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import http.server
|
||
import socket
|
||
import threading
|
||
import unittest
|
||
import urllib.parse
|
||
import urllib.request
|
||
from typing import ClassVar
|
||
|
||
from farm import config, net
|
||
from farm.client import (
|
||
DEFAULT_ENDPOINT,
|
||
LOG_KEEP,
|
||
AnimalView,
|
||
PlotView,
|
||
RemoteBotClient,
|
||
RemoteFarm,
|
||
StateView,
|
||
_underlying_socket,
|
||
)
|
||
from farm.engine import GameError
|
||
|
||
|
||
# --------------------------------------------------------------------- 构造假数据
|
||
def plot_dict(**overrides):
|
||
"""一份「服务器会发过来的」地块 JSON。"""
|
||
data = {
|
||
"index": 0,
|
||
"position": 1,
|
||
"crop": "carrot",
|
||
"crop_name": "胡萝卜",
|
||
"growth": 1.0,
|
||
"grow_days": 3,
|
||
"days_left": 2,
|
||
"moisture": 3,
|
||
"is_ripe": False,
|
||
"is_empty": False,
|
||
"stage": "生长中",
|
||
}
|
||
data.update(overrides)
|
||
return data
|
||
|
||
|
||
def animal_dict(**overrides):
|
||
data = {
|
||
"id": 7,
|
||
"kind": "cow",
|
||
"name": "奶牛",
|
||
"product": "milk",
|
||
"fed": True,
|
||
"pending": 0,
|
||
"cycle": 3,
|
||
"progress": 1,
|
||
"starve": 0,
|
||
"days_to_product": 2,
|
||
"is_sick": False,
|
||
}
|
||
data.update(overrides)
|
||
return data
|
||
|
||
|
||
def offline_client(**kwargs):
|
||
"""不连网络的 RemoteFarm（autostart=False 是关键）。"""
|
||
kwargs.setdefault("autostart", False)
|
||
kwargs.setdefault("realtime", False)
|
||
return RemoteFarm(**kwargs)
|
||
|
||
|
||
def free_port() -> int:
|
||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||
sock.bind(("127.0.0.1", 0))
|
||
return int(sock.getsockname()[1])
|
||
|
||
|
||
class _StubHandler(http.server.BaseHTTPRequestHandler):
|
||
"""按路径吐预设响应的小桩，够本文件用。**
|
||
|
||
``payloads`` 由测试在 ``setUp`` 里填：path -> (status, body_bytes, content_type)。
|
||
"""
|
||
|
||
protocol_version = "HTTP/1.1"
|
||
payloads: ClassVar[dict[str, tuple[int, bytes, str]]] = {}
|
||
|
||
def do_GET(self) -> None:
|
||
self._reply()
|
||
|
||
def do_POST(self) -> None:
|
||
length = int(self.headers.get("Content-Length") or 0)
|
||
if length:
|
||
self.rfile.read(length)
|
||
self._reply()
|
||
|
||
def _reply(self) -> None:
|
||
path = urllib.parse.urlsplit(self.path).path
|
||
status, body, content_type = self.payloads.get(
|
||
path, (404, b'{"ok": false, "error": "not_found"}', "application/json")
|
||
)
|
||
self.send_response(status)
|
||
self.send_header("Content-Type", content_type)
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def log_message(self, *args) -> None:
|
||
"""别把访问日志刷到测试输出里。"""
|
||
|
||
|
||
# ===================================================================== 视图字段映射
|
||
class PlotViewTestCase(unittest.TestCase):
|
||
def test_fields_follow_the_server_payload(self):
|
||
view = PlotView(plot_dict())
|
||
self.assertEqual(view.index, 0)
|
||
self.assertEqual(view.position, 1)
|
||
self.assertEqual(view.crop, "carrot")
|
||
self.assertEqual(view.crop_name, "胡萝卜")
|
||
self.assertEqual(view.growth, 1.0)
|
||
self.assertEqual(view.grow_days, 3)
|
||
self.assertEqual(view.days_left, 2)
|
||
self.assertEqual(view.moisture, 3)
|
||
self.assertFalse(view.is_ripe)
|
||
self.assertFalse(view.is_empty)
|
||
self.assertEqual(view.stage, "生长中")
|
||
|
||
def test_missing_fields_fall_back_to_defaults(self):
|
||
view = PlotView({})
|
||
self.assertEqual(view.index, 0)
|
||
# position 缺省时由 index 推出（地块编号是给人看的，从 1 起）
|
||
self.assertEqual(view.position, 1)
|
||
self.assertIsNone(view.crop)
|
||
self.assertEqual(view.crop_name, "")
|
||
self.assertEqual(view.growth, 0.0)
|
||
self.assertEqual(view.moisture, 0)
|
||
self.assertEqual(view.stage, "")
|
||
|
||
def test_position_defaults_to_index_plus_one(self):
|
||
self.assertEqual(PlotView({"index": 4}).position, 5)
|
||
|
||
def test_empty_defaults_to_crop_is_none(self):
|
||
self.assertTrue(PlotView({"crop": None}).is_empty)
|
||
self.assertFalse(PlotView({"crop": "carrot"}).is_empty)
|
||
|
||
def test_explicit_is_empty_wins_over_crop(self):
|
||
self.assertTrue(PlotView({"crop": "carrot", "is_empty": True}).is_empty)
|
||
|
||
def test_numeric_strings_are_coerced(self):
|
||
# JSON 里数字有可能是字符串，视图不该把它漏给渲染层
|
||
view = PlotView({"growth": "2.5", "moisture": "3", "is_ripe": 1})
|
||
self.assertEqual(view.growth, 2.5)
|
||
self.assertEqual(view.moisture, 3)
|
||
self.assertIs(view.is_ripe, True)
|
||
|
||
def test_to_dict_is_a_copy(self):
|
||
view = PlotView(plot_dict())
|
||
dumped = view.to_dict()
|
||
dumped["crop"] = "corn"
|
||
self.assertEqual(view.crop, "carrot")
|
||
|
||
def test_view_is_read_only(self):
|
||
view = PlotView(plot_dict())
|
||
with self.assertRaises(AttributeError):
|
||
view.index = 9 # type: ignore[misc]
|
||
|
||
|
||
class AnimalViewTestCase(unittest.TestCase):
|
||
def test_fields_follow_the_server_payload(self):
|
||
view = AnimalView(animal_dict())
|
||
self.assertEqual(view.aid, 7)
|
||
self.assertEqual(view.kind, "cow")
|
||
self.assertEqual(view.name, "奶牛")
|
||
self.assertEqual(view.product, "milk")
|
||
self.assertTrue(view.fed)
|
||
self.assertEqual(view.pending, 0)
|
||
self.assertEqual(view.cycle, 3)
|
||
self.assertEqual(view.progress, 1)
|
||
self.assertEqual(view.days_to_product, 2)
|
||
self.assertFalse(view.is_sick)
|
||
|
||
def test_id_maps_to_aid(self):
|
||
# 服务器用 "id"，本地模型叫 .aid —— 同构靠的就是这一层改名
|
||
self.assertEqual(AnimalView({"id": 12}).aid, 12)
|
||
|
||
def test_missing_fields_fall_back_to_defaults(self):
|
||
view = AnimalView({})
|
||
self.assertEqual(view.aid, 0)
|
||
self.assertEqual(view.kind, "")
|
||
self.assertEqual(view.product, "")
|
||
self.assertFalse(view.fed)
|
||
self.assertEqual(view.cycle, 1) # 牲畜至少处在第 1 个周期
|
||
self.assertEqual(view.starve, 0)
|
||
self.assertFalse(view.is_sick)
|
||
|
||
def test_to_dict_is_a_copy(self):
|
||
view = AnimalView(animal_dict())
|
||
dumped = view.to_dict()
|
||
dumped["kind"] = "sheep"
|
||
self.assertEqual(view.kind, "cow")
|
||
|
||
|
||
# ===================================================================== 状态视图语义
|
||
class StateViewTestCase(unittest.TestCase):
|
||
def test_full_snapshot_replaces_everything(self):
|
||
view = StateView({"day": 1, "plots": [plot_dict()], "animals": [animal_dict()]})
|
||
self.assertEqual(len(view.plots), 1)
|
||
self.assertEqual(len(view.animals), 1)
|
||
|
||
view.apply({"day": 2, "plots": [], "animals": []})
|
||
self.assertEqual(view.day, 2)
|
||
self.assertEqual(view.plots, [])
|
||
self.assertEqual(view.animals, [])
|
||
|
||
def test_brief_state_merges_and_keeps_the_farm(self):
|
||
"""回归：动作回执里的精简 state 绝不能清空农田。"""
|
||
view = StateView(
|
||
{"day": 1, "gold": 100, "plots": [plot_dict(), plot_dict(index=1)], "animals": []}
|
||
)
|
||
view.apply({"gold": 60})
|
||
|
||
self.assertEqual(view.gold, 60)
|
||
self.assertEqual(view.day, 1, "没提到的字段要保留")
|
||
self.assertEqual(len(view.plots), 2, "一次播种不该让农田消失")
|
||
|
||
def test_brief_state_after_a_full_snapshot_keeps_views_usable(self):
|
||
view = StateView({"plots": [plot_dict(crop="corn")], "animals": [animal_dict()]})
|
||
view.apply({"gold": 1})
|
||
self.assertEqual(view.plots[0].crop, "corn")
|
||
self.assertEqual(len(view.plots), 1)
|
||
self.assertEqual(len(view.animals), 1)
|
||
|
||
def test_empty_data_is_ignored(self):
|
||
view = StateView({"day": 3, "gold": 50})
|
||
view.apply({})
|
||
self.assertEqual(view.day, 3)
|
||
self.assertEqual(view.gold, 50)
|
||
|
||
def test_initial_construction_is_tolerant(self):
|
||
view = StateView()
|
||
self.assertEqual(view.plots, [])
|
||
self.assertEqual(view.animals, [])
|
||
self.assertEqual(view.day, 1)
|
||
self.assertEqual(view.gold, 0)
|
||
self.assertEqual(view.level, 1)
|
||
self.assertEqual(view.market, {})
|
||
|
||
def test_an_empty_view_looks_like_an_empty_farm(self):
|
||
"""没收到任何快照时也必须能安全渲染。
|
||
|
||
服务器连不上、或客户端是 ``autostart=False`` 起的，界面拿到的是空视图；
|
||
这时读 ``.plots`` 必须是"空农田"，不能是 AttributeError。
|
||
"""
|
||
view = StateView()
|
||
self.assertTrue(hasattr(view, "plots"))
|
||
self.assertTrue(hasattr(view, "animals"))
|
||
self.assertEqual(len(view.plots), 0)
|
||
self.assertEqual(len(view.animals), 0)
|
||
self.assertEqual(view.to_dict(), {})
|
||
|
||
def test_offline_client_exposes_a_renderable_empty_state(self):
|
||
client = offline_client(endpoint="http://127.0.0.1:1/")
|
||
self.assertEqual(client.state.plots, [])
|
||
self.assertEqual(client.snapshot(), {})
|
||
|
||
def test_non_dict_entries_are_filtered_out(self):
|
||
view = StateView({"plots": [plot_dict(), "垃圾", 3], "animals": [None, animal_dict()]})
|
||
self.assertEqual(len(view.plots), 1)
|
||
self.assertEqual(len(view.animals), 1)
|
||
|
||
def test_scalar_accessors_read_through(self):
|
||
view = StateView(
|
||
{
|
||
"day": 5,
|
||
"gold": 320,
|
||
"level": 4,
|
||
"exp": 11,
|
||
"feed": 9,
|
||
"market": {"corn": 12},
|
||
}
|
||
)
|
||
self.assertEqual((view.day, view.gold, view.level), (5, 320, 4))
|
||
self.assertEqual((view.exp, view.feed), (11, 9))
|
||
self.assertEqual(view.market["corn"], 12)
|
||
|
||
def test_get_and_item_access(self):
|
||
view = StateView({"online": 3, "paused": False})
|
||
self.assertEqual(view.get("online"), 3)
|
||
self.assertEqual(view.get("nope", "兜底"), "兜底")
|
||
self.assertEqual(view["online"], 3)
|
||
with self.assertRaises(KeyError):
|
||
_ = view["nope"]
|
||
|
||
def test_to_dict_round_trips_the_raw_payload(self):
|
||
payload = {"day": 2, "gold": 5, "plots": [plot_dict()]}
|
||
self.assertEqual(StateView(payload).to_dict(), payload)
|
||
|
||
def test_to_dict_is_a_copy(self):
|
||
view = StateView({"gold": 5})
|
||
dumped = view.to_dict()
|
||
dumped["gold"] = 999
|
||
self.assertEqual(view.gold, 5)
|
||
|
||
|
||
# ===================================================================== 地址与 URL
|
||
class AddressTestCase(unittest.TestCase):
|
||
def test_normalize_adds_the_scheme(self):
|
||
self.assertEqual(RemoteFarm._normalize("127.0.0.1:8765"), "http://127.0.0.1:8765")
|
||
|
||
def test_normalize_maps_the_legacy_tcp_scheme(self):
|
||
self.assertEqual(RemoteFarm._normalize("tcp://127.0.0.1:8765"), "http://127.0.0.1:8765")
|
||
|
||
def test_normalize_strips_whitespace_and_trailing_slashes(self):
|
||
self.assertEqual(RemoteFarm._normalize(" http://host:9000/// "), "http://host:9000")
|
||
|
||
def test_normalize_keeps_https(self):
|
||
self.assertEqual(RemoteFarm._normalize("https://farm.example"), "https://farm.example")
|
||
|
||
def test_normalize_falls_back_when_empty(self):
|
||
self.assertEqual(RemoteFarm._normalize(""), DEFAULT_ENDPOINT)
|
||
self.assertEqual(RemoteFarm._normalize(" "), DEFAULT_ENDPOINT)
|
||
|
||
def test_normalize_falls_back_for_degenerate_endpoints(self):
|
||
"""畸形端点不能变成"没有主机名"的地址。
|
||
|
||
否则 host 是空串，``net.is_loopback`` 判成 False，请求会溜到系统代理上，
|
||
报回来一个和真实原因毫无关系的 502。回落到本机默认才是对的。
|
||
"""
|
||
for endpoint in ("", " ", "//", "http://", "https://", "tcp://"):
|
||
with self.subTest(endpoint=endpoint):
|
||
self.assertEqual(RemoteFarm._normalize(endpoint), DEFAULT_ENDPOINT)
|
||
|
||
def test_a_normalized_endpoint_always_has_a_host(self):
|
||
candidates = ("", " ", "http://", "127.0.0.1:1", "http://host:9000/", "tcp://h:1")
|
||
for endpoint in candidates:
|
||
with self.subTest(endpoint=endpoint):
|
||
normalized = RemoteFarm._normalize(endpoint)
|
||
self.assertTrue(urllib.parse.urlsplit(normalized).netloc, normalized)
|
||
|
||
def test_init_stores_a_normalized_endpoint(self):
|
||
client = offline_client(endpoint=" tcp://127.0.0.1:9999/ ")
|
||
self.assertEqual(client.endpoint, "http://127.0.0.1:9999")
|
||
self.assertEqual(client.host, "127.0.0.1:9999")
|
||
|
||
def test_farm_name_defaults_to_default(self):
|
||
self.assertEqual(offline_client(farm="").farm, "default")
|
||
self.assertEqual(offline_client(farm=None).farm, "default") # type: ignore[arg-type]
|
||
|
||
def test_host_falls_back_when_the_endpoint_has_none(self):
|
||
self.assertEqual(offline_client(endpoint="http://").host, "127.0.0.1:8765")
|
||
|
||
def test_url_always_carries_the_farm(self):
|
||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="阿黄")
|
||
parts = urllib.parse.urlsplit(client.url("/api/state"))
|
||
self.assertEqual(parts.path, "/api/state")
|
||
self.assertEqual(urllib.parse.parse_qs(parts.query), {"farm": ["阿黄"]})
|
||
|
||
def test_url_includes_the_token_when_present(self):
|
||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a", token="s3cret")
|
||
query = urllib.parse.parse_qs(urllib.parse.urlsplit(client.url("/api/state")).query)
|
||
self.assertEqual(query["token"], ["s3cret"])
|
||
|
||
def test_url_omits_the_token_when_absent(self):
|
||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a")
|
||
query = urllib.parse.parse_qs(urllib.parse.urlsplit(client.url("/api/state")).query)
|
||
self.assertNotIn("token", query)
|
||
|
||
def test_url_carries_extra_params(self):
|
||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a")
|
||
url = client.url("/api/log", limit=30)
|
||
query = urllib.parse.parse_qs(urllib.parse.urlsplit(url).query)
|
||
self.assertEqual(query["limit"], ["30"])
|
||
|
||
def test_url_drops_params_that_are_none(self):
|
||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a")
|
||
url = client.url("/api/log", limit=None, since=0)
|
||
query = urllib.parse.parse_qs(urllib.parse.urlsplit(url).query)
|
||
self.assertNotIn("limit", query)
|
||
self.assertEqual(query["since"], ["0"], "0 是有效值，不能当 None 丢掉")
|
||
|
||
def test_url_is_fully_qualified(self):
|
||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a")
|
||
self.assertTrue(client.url("/api/state").startswith("http://127.0.0.1:1/api/state?"))
|
||
|
||
def test_stream_url_carries_farm_and_name(self):
|
||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a", name="阿黄")
|
||
parts = urllib.parse.urlsplit(client.stream_url)
|
||
self.assertEqual(parts.path, "/api/stream")
|
||
query = urllib.parse.parse_qs(parts.query)
|
||
self.assertEqual(query["farm"], ["a"])
|
||
# 令牌只能走查询参数：EventSource / urllib 都没法给长连接加自定义头
|
||
self.assertEqual(query["name"], ["阿黄"])
|
||
|
||
def test_stream_url_skips_the_empty_name(self):
|
||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a", name="")
|
||
query = urllib.parse.parse_qs(urllib.parse.urlsplit(client.stream_url).query)
|
||
self.assertNotIn("name", query)
|
||
|
||
def test_stream_url_includes_the_token(self):
|
||
client = offline_client(endpoint="http://127.0.0.1:1/", farm="a", token="t")
|
||
query = urllib.parse.parse_qs(urllib.parse.urlsplit(client.stream_url).query)
|
||
self.assertEqual(query["token"], ["t"])
|
||
|
||
|
||
# ===================================================================== SSE 消息分发
|
||
class StreamMessageTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.client = offline_client(endpoint="http://127.0.0.1:1/", farm="a")
|
||
|
||
def push(self, message):
|
||
import json
|
||
|
||
self.client._handle_stream_message(json.dumps(message, ensure_ascii=False))
|
||
|
||
def test_welcome_installs_state_catalog_and_server_info(self):
|
||
self.push(
|
||
{
|
||
"type": "welcome",
|
||
"farm": "b",
|
||
"online": 4,
|
||
"server": {"version": "2.3"},
|
||
"catalog": {"crops": [{"key": "carrot"}]},
|
||
"state": {"day": 6, "gold": 88, "plots": [plot_dict()], "animals": []},
|
||
"log": [],
|
||
}
|
||
)
|
||
self.assertEqual(self.client.farm, "b")
|
||
self.assertEqual(self.client.server_info, {"version": "2.3"})
|
||
self.assertEqual(self.client.catalog()["crops"], [{"key": "carrot"}])
|
||
self.assertEqual(self.client.state.day, 6)
|
||
self.assertEqual(self.client.state.gold, 88)
|
||
self.assertEqual(self.client.state.get("online"), 4, "在线人数要合进状态视图")
|
||
self.assertEqual(len(self.client.state.plots), 1)
|
||
self.assertEqual(self.client.stats["pushes"], 1)
|
||
|
||
def test_state_and_sync_share_the_welcome_branch(self):
|
||
for kind in ("state", "sync"):
|
||
with self.subTest(kind=kind):
|
||
self.push({"type": kind, "state": {"gold": 7}, "plots": None})
|
||
self.assertEqual(self.client.state.gold, 7)
|
||
self.assertEqual(self.client.stats["pushes"], 2)
|
||
|
||
def test_brief_state_push_does_not_wipe_the_farm(self):
|
||
self.push({"type": "state", "state": {"plots": [plot_dict()], "animals": []}})
|
||
self.push({"type": "state", "state": {"gold": 3}})
|
||
self.assertEqual(len(self.client.state.plots), 1)
|
||
self.assertEqual(self.client.state.gold, 3)
|
||
|
||
def test_push_without_state_still_counts_and_keeps_old_data(self):
|
||
self.push({"type": "state", "state": {"day": 9}})
|
||
self.push({"type": "state"})
|
||
self.assertEqual(self.client.state.day, 9)
|
||
self.assertEqual(self.client.stats["pushes"], 2)
|
||
|
||
def test_push_ignores_a_bad_online_value(self):
|
||
self.push({"type": "state", "online": "很多", "state": {"day": 1}})
|
||
self.assertIsNone(self.client.state.get("online"))
|
||
|
||
def test_online_is_dropped_when_the_state_is_empty(self):
|
||
self.push({"type": "state", "online": 3})
|
||
self.assertIsNone(self.client.state.get("online"))
|
||
|
||
def test_push_merges_the_log(self):
|
||
self.push(
|
||
{
|
||
"type": "state",
|
||
"state": {"day": 1},
|
||
"log": [{"seq": 2, "text": "b"}, {"seq": 1, "text": "a"}],
|
||
}
|
||
)
|
||
entries = self.client.recent_log()
|
||
self.assertEqual([e["seq"] for e in entries], [1, 2], "日志要按 seq 排序")
|
||
self.assertEqual([e["text"] for e in entries], ["a", "b"])
|
||
|
||
def test_day_message_appends_a_header_and_its_events(self):
|
||
self.push({"type": "day", "day": 4, "events": ["奶牛生病了", "一块地旱了"]})
|
||
entries = self.client.recent_log()
|
||
self.assertEqual(len(entries), 3)
|
||
self.assertEqual(entries[0]["text"], "第 4 天开始")
|
||
self.assertEqual(entries[0]["kind"], "info")
|
||
self.assertEqual([e["text"] for e in entries[1:]], ["奶牛生病了", "一块地旱了"])
|
||
self.assertTrue(all(e["kind"] == "warn" for e in entries[1:]))
|
||
self.assertTrue(all(e["day"] == 4 for e in entries))
|
||
|
||
def test_day_message_continues_the_sequence(self):
|
||
self.push({"type": "state", "state": {"day": 1}, "log": [{"seq": 5, "text": "旧"}]})
|
||
self.push({"type": "day", "day": 2, "events": ["一"]})
|
||
self.assertEqual([e["seq"] for e in self.client.recent_log()], [5, 6, 7])
|
||
|
||
def test_day_message_survives_missing_events(self):
|
||
self.push({"type": "day", "day": 2})
|
||
entries = self.client.recent_log()
|
||
self.assertEqual(len(entries), 1)
|
||
self.assertEqual(entries[0]["text"], "第 2 天开始")
|
||
|
||
def test_error_message_is_recorded(self):
|
||
self.push({"type": "error", "message": "牧场不存在"})
|
||
self.assertEqual(self.client.last_error, "牧场不存在")
|
||
self.assertEqual(self.client.stats["errors"], 1)
|
||
|
||
def test_bye_disconnects_with_a_reason(self):
|
||
self.client.connected = True
|
||
self.push({"type": "bye", "message": "服务器正在停机"})
|
||
self.assertFalse(self.client.connected)
|
||
self.assertEqual(self.client.last_error, "服务器正在停机")
|
||
|
||
def test_bye_without_a_reason_uses_a_default(self):
|
||
self.client.connected = True
|
||
self.push({"type": "bye"})
|
||
self.assertFalse(self.client.connected)
|
||
self.assertEqual(self.client.last_error, "服务器已停机")
|
||
|
||
def test_malformed_payloads_are_ignored(self):
|
||
for payload in ("不是 JSON", "[1, 2, 3]", '"字符串"', "3", ""):
|
||
with self.subTest(payload=payload):
|
||
self.client._handle_stream_message(payload)
|
||
self.assertEqual(self.client.stats["pushes"], 0)
|
||
self.assertEqual(self.client.stats["errors"], 0)
|
||
self.assertEqual(self.client.last_error, "")
|
||
|
||
def test_unknown_message_types_are_ignored(self):
|
||
self.push({"type": "谁知道呢", "state": {"gold": 7}})
|
||
self.assertEqual(self.client.state.gold, 0)
|
||
self.assertEqual(self.client.stats["pushes"], 0)
|
||
|
||
|
||
# ===================================================================== 日志与回调
|
||
class LogAndCallbackTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.client = offline_client(endpoint="http://127.0.0.1:1/")
|
||
|
||
def test_merge_log_deduplicates_by_seq(self):
|
||
self.client._merge_log([{"seq": 1, "text": "a"}])
|
||
self.client._merge_log([{"seq": 1, "text": "a"}, {"seq": 2, "text": "b"}])
|
||
self.assertEqual([e["seq"] for e in self.client.recent_log()], [1, 2])
|
||
|
||
def test_merge_log_skips_entries_without_a_sequence(self):
|
||
self.client._merge_log([{"text": "无号"}, {"seq": 0, "text": "零号"}, {"seq": 1}])
|
||
self.assertEqual([e["seq"] for e in self.client.recent_log()], [1])
|
||
|
||
def test_merge_log_keeps_only_the_most_recent_entries(self):
|
||
self.client._merge_log([{"seq": n} for n in range(1, LOG_KEEP + 51)])
|
||
entries = self.client.recent_log(limit=0)
|
||
self.assertEqual(len(entries), LOG_KEEP)
|
||
self.assertEqual(entries[-1]["seq"], LOG_KEEP + 50)
|
||
|
||
def test_recent_log_filters_by_since(self):
|
||
self.client._merge_log([{"seq": n} for n in range(1, 6)])
|
||
self.assertEqual([e["seq"] for e in self.client.recent_log(since=3)], [4, 5])
|
||
|
||
def test_recent_log_limit_takes_the_tail(self):
|
||
self.client._merge_log([{"seq": n} for n in range(1, 6)])
|
||
self.assertEqual([e["seq"] for e in self.client.recent_log(limit=2)], [4, 5])
|
||
|
||
def test_recent_log_limit_zero_returns_everything(self):
|
||
self.client._merge_log([{"seq": n} for n in range(1, 6)])
|
||
self.assertEqual(len(self.client.recent_log(limit=0)), 5)
|
||
|
||
def test_apply_fires_the_change_callback(self):
|
||
hits = []
|
||
self.client.on_change = lambda: hits.append(1)
|
||
self.client._apply({"gold": 1})
|
||
self.client._apply({"gold": 2})
|
||
self.assertEqual(len(hits), 2)
|
||
|
||
def test_apply_does_not_fire_for_empty_state(self):
|
||
hits = []
|
||
self.client.on_change = lambda: hits.append(1)
|
||
self.client._apply({})
|
||
self.assertEqual(hits, [])
|
||
|
||
def test_a_broken_callback_cannot_break_data_sync(self):
|
||
def boom():
|
||
raise RuntimeError("界面炸了")
|
||
|
||
self.client.on_change = boom
|
||
self.client._apply({"gold": 42})
|
||
self.assertEqual(self.client.state.gold, 42)
|
||
|
||
def test_transport_label_reports_the_active_channel(self):
|
||
self.client.mode = "sse"
|
||
self.assertEqual(self.client.transport_label(), "SSE 实时")
|
||
self.client.mode = "poll"
|
||
self.assertEqual(self.client.transport_label(), "HTTP 轮询")
|
||
self.client.mode = "http"
|
||
self.assertEqual(self.client.transport_label(), "HTTP")
|
||
|
||
def test_catalog_starts_with_local_config_defaults(self):
|
||
catalog = self.client.catalog()
|
||
self.assertEqual(catalog["feed_price"], config.FEED_UNIT_PRICE)
|
||
self.assertEqual(catalog["water_capacity"], config.WATER_CAPACITY)
|
||
self.assertEqual(catalog["crops"], [])
|
||
|
||
|
||
# ===================================================================== 动作转发
|
||
class ActionForwardingTestCase(unittest.TestCase):
|
||
"""写操作统一走 POST /api/actions —— 用假的 _http 把请求原样抓下来核对。"""
|
||
|
||
def setUp(self):
|
||
self.client = offline_client(endpoint="http://127.0.0.1:1/")
|
||
self.requests: list[tuple[str, str, dict, dict]] = []
|
||
self.reply: dict = {}
|
||
self.client._http = self._fake_http # type: ignore[method-assign]
|
||
|
||
def _fake_http(self, method, path, payload=None, **params):
|
||
self.requests.append((method, path, payload or {}, params))
|
||
return self.reply
|
||
|
||
def last(self):
|
||
return self.requests[-1]
|
||
|
||
def test_plant_posts_the_right_payload(self):
|
||
self.client.plant(3, "carrot")
|
||
method, path, payload, params = self.last()
|
||
self.assertEqual(method, "POST")
|
||
self.assertEqual(path, "/api/actions")
|
||
self.assertEqual(payload, {"action": "plant", "params": {"plot": 3, "crop": "carrot"}})
|
||
self.assertEqual(params, {"full": 1}, "顺手要一份完整快照，省掉操作后再查一次")
|
||
|
||
def test_simple_actions_carry_their_arguments(self):
|
||
cases = [
|
||
(lambda: self.client.water(2), "water", {"plot": 2}),
|
||
(lambda: self.client.harvest(1), "harvest", {"plot": 1}),
|
||
(lambda: self.client.harvest_all(), "harvest_all", {}),
|
||
(lambda: self.client.buy_animal("cow"), "buy_animal", {"kind": "cow"}),
|
||
(lambda: self.client.collect(7), "collect", {"animal_id": 7}),
|
||
(lambda: self.client.buy_plot(), "buy_plot", {}),
|
||
]
|
||
for call, action, params in cases:
|
||
with self.subTest(action=action):
|
||
call()
|
||
self.assertEqual(self.last()[2], {"action": action, "params": params})
|
||
|
||
def test_feed_omits_quantity_when_not_given(self):
|
||
self.client.feed(7)
|
||
self.assertEqual(self.last()[2]["params"], {"animal_id": 7})
|
||
|
||
def test_feed_forwards_quantity(self):
|
||
self.client.feed(7, 3)
|
||
self.assertEqual(self.last()[2]["params"], {"animal_id": 7, "quantity": 3})
|
||
|
||
def test_buy_feed_defaults_to_bulk(self):
|
||
self.client.buy_feed()
|
||
self.assertEqual(self.last()[2]["params"], {"quantity": config.FEED_BULK})
|
||
|
||
def test_pause_without_argument_is_a_toggle(self):
|
||
self.client.pause()
|
||
self.assertEqual(self.last()[2], {"action": "pause", "params": {}})
|
||
|
||
def test_pause_forwards_an_explicit_value(self):
|
||
self.client.pause(True)
|
||
self.assertEqual(self.last()[2]["params"], {"paused": True})
|
||
|
||
def test_pause_forwards_an_explicit_false(self):
|
||
# False 与"没传"必须区分开，否则永远只能切不能停
|
||
self.client.pause(False)
|
||
self.assertEqual(self.last()[2]["params"], {"paused": False})
|
||
|
||
def test_save_forwards_to_the_server(self):
|
||
self.client.save()
|
||
self.assertEqual(self.last()[2], {"action": "save", "params": {}})
|
||
|
||
def test_batch_posts_a_single_request(self):
|
||
items = [{"action": "water", "params": {"plot": 1}}, {"action": "harvest", "params": {}}]
|
||
self.client.batch(items)
|
||
method, path, payload, params = self.last()
|
||
self.assertEqual(method, "POST")
|
||
self.assertEqual(path, "/api/batch")
|
||
self.assertEqual(payload, {"actions": items})
|
||
self.assertEqual(params, {"full": 1})
|
||
self.assertEqual(len(self.requests), 1, "批量就该是一个请求")
|
||
|
||
def test_action_applies_the_returned_snapshot(self):
|
||
self.reply = {"snapshot": {"plots": [plot_dict()], "gold": 10}, "log": [{"seq": 1}]}
|
||
self.client.water(1)
|
||
self.assertEqual(self.client.state.gold, 10)
|
||
self.assertEqual(len(self.client.state.plots), 1)
|
||
self.assertEqual(len(self.client.recent_log()), 1)
|
||
|
||
def test_action_falls_back_to_the_brief_state(self):
|
||
self.reply = {"state": {"gold": 5}}
|
||
self.client.water(1)
|
||
self.assertEqual(self.client.state.gold, 5)
|
||
|
||
def test_snapshot_wins_over_brief_state(self):
|
||
self.reply = {"snapshot": {"gold": 10}, "state": {"gold": 5}}
|
||
self.client.water(1)
|
||
self.assertEqual(self.client.state.gold, 10)
|
||
|
||
def test_reset_clears_the_log_and_refreshes(self):
|
||
self.client._merge_log([{"seq": 1, "text": "旧"}])
|
||
refreshed = []
|
||
original = self.client.refresh
|
||
self.client.refresh = lambda *a, **k: refreshed.append(1) or True # type: ignore[method-assign]
|
||
self.client.reset()
|
||
self.client.refresh = original # type: ignore[method-assign]
|
||
self.assertEqual(self.client.recent_log(), [])
|
||
self.assertEqual(self.last()[2], {"action": "reset", "params": {}})
|
||
self.assertEqual(refreshed, [1], "重置后要重新拉全量")
|
||
|
||
|
||
# ===================================================================== 实时通道
|
||
class RealtimeTestCase(unittest.TestCase):
|
||
def test_start_is_a_noop_when_realtime_is_off(self):
|
||
client = offline_client(endpoint="http://127.0.0.1:1/", realtime=False)
|
||
client.start()
|
||
self.assertIsNone(client._thread)
|
||
|
||
def test_start_spawns_only_one_thread(self):
|
||
client = offline_client(endpoint="http://127.0.0.1:1/", realtime=True)
|
||
client._realtime_loop = lambda: None # type: ignore[method-assign]
|
||
try:
|
||
client.start()
|
||
first = client._thread
|
||
client.start()
|
||
self.assertIs(client._thread, first, "重复 start 不该开第二条连接")
|
||
finally:
|
||
client.stop()
|
||
|
||
def test_stop_is_safe_without_a_connection(self):
|
||
client = offline_client(endpoint="http://127.0.0.1:1/")
|
||
client.stop() # 不该抛
|
||
self.assertTrue(client._stop.is_set())
|
||
|
||
def test_underlying_socket_walks_the_attribute_chain(self):
|
||
class _Raw:
|
||
_sock = "套接字"
|
||
|
||
class _Fp:
|
||
raw = _Raw()
|
||
|
||
class _Response:
|
||
fp = _Fp()
|
||
|
||
self.assertEqual(_underlying_socket(_Response()), "套接字")
|
||
|
||
def test_underlying_socket_degrades_to_none(self):
|
||
"""掏不到就返回 None，调用方会退化成 close() —— 允许变慢，不允许崩。"""
|
||
for fake in (object(), type("R", (), {"fp": None})(), type("R", (), {"fp": object()})()):
|
||
with self.subTest(fake=type(fake).__name__):
|
||
self.assertIsNone(_underlying_socket(fake))
|
||
|
||
def test_stop_shuts_down_the_stream_socket(self):
|
||
events = []
|
||
|
||
class _Sock:
|
||
def shutdown(self, how):
|
||
events.append(("shutdown", how))
|
||
|
||
class _Resp:
|
||
def close(self):
|
||
events.append(("close", None))
|
||
|
||
client = offline_client(endpoint="http://127.0.0.1:1/")
|
||
client._stream_socket = _Sock()
|
||
client._stream_response = _Resp()
|
||
client.stop()
|
||
|
||
self.assertIn(("shutdown", socket.SHUT_RDWR), events)
|
||
self.assertIn(("close", None), events)
|
||
# shutdown 必须在 close 之前：close 会等读线程的锁，最长拖到下一次心跳
|
||
self.assertLess(events.index(("shutdown", socket.SHUT_RDWR)), events.index(("close", None)))
|
||
|
||
def test_stop_tolerates_broken_stream_objects(self):
|
||
class _Sock:
|
||
def shutdown(self, how):
|
||
raise OSError("已经断了")
|
||
|
||
class _Resp:
|
||
def close(self):
|
||
raise RuntimeError("也炸了")
|
||
|
||
client = offline_client(endpoint="http://127.0.0.1:1/")
|
||
client._stream_socket = _Sock()
|
||
client._stream_response = _Resp()
|
||
client.stop() # 停机路径必须稳
|
||
|
||
|
||
class BotClientTestCase(unittest.TestCase):
|
||
def test_wait_online_returns_as_soon_as_refresh_works(self):
|
||
client = RemoteBotClient(endpoint="http://127.0.0.1:1/", autostart=False, realtime=False)
|
||
client.refresh = lambda *a, **k: True # type: ignore[method-assign]
|
||
self.assertTrue(client.wait_online(seconds=5))
|
||
|
||
def test_wait_online_gives_up_after_the_deadline(self):
|
||
client = RemoteBotClient(endpoint="http://127.0.0.1:1/", autostart=False, realtime=False)
|
||
calls = []
|
||
client.refresh = lambda *a, **k: calls.append(1) or False # type: ignore[method-assign]
|
||
self.assertFalse(client.wait_online(seconds=0))
|
||
self.assertEqual(calls, [], "时限已过就不该再去连")
|
||
|
||
|
||
# ===================================================================== HTTP 传输层
|
||
class HttpTransportTestCase(unittest.TestCase):
|
||
"""用纯标准库的回环桩服务验证 _http 的成功与各类失败分支。
|
||
|
||
这不是"外部服务器"：桩只绑 127.0.0.1，起不来就整类跳过。
|
||
顺带验证了 net.open_url 的回环直连——本机设了 HTTP_PROXY，
|
||
走代理的话这里会拿到 502，而不是桩真正返回的内容。
|
||
"""
|
||
|
||
server: http.server.ThreadingHTTPServer
|
||
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
payload = b'{"ok": true, "state": {"day": 3, "gold": 42}, "online": 2}'
|
||
_StubHandler.payloads = {
|
||
"/api/state": (200, payload, "application/json"),
|
||
"/api/catalog": (
|
||
200,
|
||
b'{"ok": true, "catalog": {"feed_price": 6}}',
|
||
"application/json",
|
||
),
|
||
"/api/log": (
|
||
200,
|
||
b'{"ok": true, "entries": [{"seq": 1, "text": "a"}]}',
|
||
"application/json",
|
||
),
|
||
"/api/bad": (200, "<html>不是 JSON</html>".encode(), "text/html"),
|
||
"/api/list": (200, b"[1, 2, 3]", "application/json"),
|
||
"/api/empty": (204, b"", "application/json"),
|
||
"/api/rejected": (
|
||
400,
|
||
'{"ok": false, "error": "locked", "message": "这块地还没解锁"}'.encode(),
|
||
"application/json",
|
||
),
|
||
"/api/explode": (500, "内部错误，没有 JSON".encode(), "text/plain"),
|
||
}
|
||
try:
|
||
cls.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _StubHandler)
|
||
except OSError as exc: # pragma: no cover - 环境不允许绑端口
|
||
raise unittest.SkipTest(f"起不了回环桩服务：{exc}") from exc
|
||
cls.port = int(cls.server.server_address[1])
|
||
cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
|
||
cls.thread.start()
|
||
|
||
@classmethod
|
||
def tearDownClass(cls):
|
||
cls.server.shutdown()
|
||
cls.server.server_close()
|
||
cls.thread.join(timeout=5)
|
||
|
||
def setUp(self):
|
||
self.client = offline_client(endpoint=f"127.0.0.1:{self.port}", farm="测试场")
|
||
|
||
def test_get_returns_the_parsed_body(self):
|
||
data = self.client._http("GET", "/api/state")
|
||
self.assertEqual(data["state"]["gold"], 42)
|
||
self.assertTrue(self.client.connected)
|
||
self.assertEqual(self.client.last_error, "")
|
||
self.assertEqual(self.client.stats["http"], 1)
|
||
|
||
def test_loopback_requests_do_not_go_through_the_proxy(self):
|
||
# 桩返回的是我们自己的 JSON；被系统代理截走的话这里会变成 502 页面
|
||
self.assertEqual(self.client._http("GET", "/api/catalog")["catalog"]["feed_price"], 6)
|
||
|
||
def test_open_url_accepts_a_plain_string(self):
|
||
url = f"http://127.0.0.1:{self.port}/api/state"
|
||
with net.open_url(url, timeout=5) as response:
|
||
self.assertEqual(response.status, 200)
|
||
self.assertIn("gold", response.read().decode("utf-8"))
|
||
|
||
def test_refresh_pulls_state_catalog_and_log(self):
|
||
self.assertTrue(self.client.refresh())
|
||
self.assertEqual(self.client.state.day, 3)
|
||
self.assertEqual(self.client.state.gold, 42)
|
||
self.assertEqual(self.client.state.get("online"), 2)
|
||
self.assertEqual(self.client.catalog()["feed_price"], 6)
|
||
self.assertEqual([e["seq"] for e in self.client.recent_log()], [1])
|
||
self.assertEqual(self.client.stats["http"], 3)
|
||
|
||
def test_refresh_swallows_errors_by_default(self):
|
||
client = offline_client(endpoint="127.0.0.1:1", farm="测试场") # 没人监听
|
||
self.assertFalse(client.refresh())
|
||
self.assertFalse(client.connected)
|
||
self.assertTrue(client.last_error)
|
||
|
||
def test_refresh_can_raise_on_error(self):
|
||
client = offline_client(endpoint="127.0.0.1:1", farm="测试场")
|
||
with self.assertRaises(GameError) as ctx:
|
||
client.refresh(raise_on_error=True)
|
||
self.assertEqual(ctx.exception.code, "network")
|
||
|
||
def test_empty_body_is_treated_as_an_empty_object(self):
|
||
self.assertEqual(self.client._http("GET", "/api/empty"), {})
|
||
|
||
def test_non_json_body_is_reported_as_a_bad_response(self):
|
||
with self.assertRaises(GameError) as ctx:
|
||
self.client._http("GET", "/api/bad")
|
||
self.assertEqual(ctx.exception.code, "bad_response")
|
||
self.assertEqual(ctx.exception.status, 502)
|
||
|
||
def test_json_that_is_not_an_object_is_rejected(self):
|
||
with self.assertRaises(GameError) as ctx:
|
||
self.client._http("GET", "/api/list")
|
||
self.assertEqual(ctx.exception.code, "bad_response")
|
||
self.assertIn("不是对象", ctx.exception.message)
|
||
|
||
def test_server_rejection_becomes_a_game_error(self):
|
||
with self.assertRaises(GameError) as ctx:
|
||
self.client._http("POST", "/api/rejected", {"action": "plant"})
|
||
self.assertEqual(ctx.exception.code, "locked")
|
||
self.assertEqual(ctx.exception.message, "这块地还没解锁")
|
||
self.assertEqual(ctx.exception.status, 400)
|
||
self.assertEqual(self.client.stats["errors"], 1)
|
||
|
||
def test_error_responses_that_are_not_json_say_so(self):
|
||
with self.assertRaises(GameError) as ctx:
|
||
self.client._http("GET", "/api/explode")
|
||
self.assertEqual(ctx.exception.code, "bad_response")
|
||
|
||
def test_a_dead_server_becomes_a_network_error_with_a_hint(self):
|
||
dead = offline_client(endpoint=f"127.0.0.1:{free_port()}", farm="测试场")
|
||
with self.assertRaises(GameError) as ctx:
|
||
dead._http("GET", "/api/state")
|
||
self.assertEqual(ctx.exception.code, "network")
|
||
self.assertEqual(ctx.exception.status, 503)
|
||
self.assertIn("main.py serve", ctx.exception.message, "错误信息要告诉用户怎么补救")
|
||
self.assertFalse(dead.connected)
|
||
self.assertEqual(dead.stats["errors"], 1)
|
||
|
||
def test_action_response_is_applied_end_to_end(self):
|
||
_StubHandler.payloads["/api/actions"] = (
|
||
200,
|
||
b'{"ok": true, "snapshot": {"gold": 7, "plots": []}, "log": [{"seq": 9}]}',
|
||
"application/json",
|
||
)
|
||
try:
|
||
self.client._action("harvest_all")
|
||
finally:
|
||
_StubHandler.payloads.pop("/api/actions", None)
|
||
self.assertEqual(self.client.state.gold, 7)
|
||
self.assertEqual([e["seq"] for e in self.client.recent_log()], [9])
|
||
|
||
|
||
# ===================================================================== 代理绕行
|
||
class NetTestCase(unittest.TestCase):
|
||
def test_loopback_hosts_are_recognized(self):
|
||
for host in ("127.0.0.1", "127.0.0.5", "localhost", "LocalHost", "::1", "[::1]", ""):
|
||
with self.subTest(host=host):
|
||
self.assertTrue(net.is_loopback(host))
|
||
|
||
def test_loopback_check_tolerates_surrounding_space(self):
|
||
self.assertTrue(net.is_loopback(" 127.0.0.1 "))
|
||
|
||
def test_remote_hosts_are_not_loopback(self):
|
||
for host in ("farm.example.com", "10.0.0.1", "192.168.1.9", "128.0.0.1", None):
|
||
with self.subTest(host=host):
|
||
self.assertFalse(net.is_loopback(host))
|
||
|
||
def test_host_of_extracts_from_a_string_or_a_request(self):
|
||
self.assertEqual(net.host_of("http://127.0.0.1:8765/api/state"), "127.0.0.1")
|
||
request = urllib.request.Request("http://example.com:9000/x")
|
||
self.assertEqual(net.host_of(request), "example.com")
|
||
|
||
def test_host_of_returns_none_when_unparseable(self):
|
||
self.assertIsNone(net.host_of("没有协议也没有主机"))
|
||
|
||
def test_endpoint_host_accepts_a_bare_host_port(self):
|
||
self.assertEqual(net.endpoint_host("127.0.0.1:8765"), "127.0.0.1")
|
||
|
||
def test_endpoint_host_falls_back_to_the_default(self):
|
||
self.assertEqual(net.endpoint_host("http://"), "127.0.0.1")
|
||
self.assertEqual(net.endpoint_host("", default="10.0.0.1"), "10.0.0.1")
|
||
|
||
def test_the_loopback_opener_registers_no_proxy_handler_at_all(self):
|
||
"""回环直连的机制：这个 opener 里压根没有代理处理器。
|
||
|
||
``ProxyHandler({})`` 不会注册任何 ``*_open`` 钩子，而 ``add_handler``
|
||
只在"至少挂上一个钩子"时才把 handler 记进 ``handlers``，所以它自己
|
||
反而没被收进去——结果是 http 请求直接进 ``HTTPHandler``，必然直连。
|
||
"""
|
||
handlers = net._no_proxy_opener.handlers
|
||
self.assertFalse([h for h in handlers if isinstance(h, urllib.request.ProxyHandler)])
|
||
|
||
|
||
if __name__ == "__main__": # pragma: no cover
|
||
unittest.main()
|