503 lines
20 KiB
Python
503 lines
20 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
"""svfs_client 纯标准库层的登录认证测试（不 import Qt）。
|
|||
|
|
|
|||
|
|
用本地 http.server 起一个假服务端（行为可用类属性调整），覆盖
|
|||
|
|
auth.login / verify_token / logout 的成功与各类失败，以及 settings
|
|||
|
|
新增键（auth_token / auth_user）的持久化。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
import threading
|
|||
|
|
import unittest
|
|||
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|||
|
|
|
|||
|
|
from helpers import VFSTestCase
|
|||
|
|
|
|||
|
|
from svfs_client import auth, netutil
|
|||
|
|
from svfs_client import paths as client_paths
|
|||
|
|
from svfs_client import settings as client_settings
|
|||
|
|
|
|||
|
|
|
|||
|
|
class _AuthApiHandler(BaseHTTPRequestHandler):
|
|||
|
|
"""假服务端：/api/login、/api/me、/api/logout；``mode`` 可注入异常行为。"""
|
|||
|
|
|
|||
|
|
users = {"admin": "admin123"}
|
|||
|
|
sessions = {}
|
|||
|
|
mode = {}
|
|||
|
|
|
|||
|
|
def _bearer(self):
|
|||
|
|
header = self.headers.get("Authorization") or ""
|
|||
|
|
return header[7:].strip() if header.startswith("Bearer ") else ""
|
|||
|
|
|
|||
|
|
def _reply(self, code, payload):
|
|||
|
|
body = json.dumps(payload).encode("utf-8")
|
|||
|
|
self.send_response(code)
|
|||
|
|
self.send_header("Content-Type", "application/json")
|
|||
|
|
self.send_header("Content-Length", str(len(body)))
|
|||
|
|
self.end_headers()
|
|||
|
|
self.wfile.write(body)
|
|||
|
|
|
|||
|
|
def _read_json(self):
|
|||
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|||
|
|
try:
|
|||
|
|
return json.loads(self.rfile.read(length) or b"{}")
|
|||
|
|
except ValueError:
|
|||
|
|
return {}
|
|||
|
|
|
|||
|
|
def do_POST(self): # noqa: N802
|
|||
|
|
data = self._read_json()
|
|||
|
|
if self.path == "/api/login":
|
|||
|
|
if self.mode.get("login") == "badjson":
|
|||
|
|
raw = b"this is not json"
|
|||
|
|
self.send_response(200)
|
|||
|
|
self.send_header("Content-Length", str(len(raw)))
|
|||
|
|
self.end_headers()
|
|||
|
|
self.wfile.write(raw)
|
|||
|
|
return
|
|||
|
|
username = str(data.get("username") or "")
|
|||
|
|
if username and self.users.get(username) == data.get("password"):
|
|||
|
|
token = f"tok-{username}-{len(self.sessions)}"
|
|||
|
|
self.sessions[token] = {
|
|||
|
|
"username": username,
|
|||
|
|
"login_time": "2026-09-15T10:00:00",
|
|||
|
|
}
|
|||
|
|
if self.mode.get("login") == "notoken":
|
|||
|
|
self._reply(200, {"ok": True, "username": username})
|
|||
|
|
else:
|
|||
|
|
self._reply(200, {"ok": True, "token": token,
|
|||
|
|
"username": username,
|
|||
|
|
"login_time": "2026-09-15T10:00:00"})
|
|||
|
|
else:
|
|||
|
|
self._reply(401, {"detail": "用户名或密码不正确"})
|
|||
|
|
elif self.path == "/api/logout":
|
|||
|
|
self.sessions.pop(self._bearer(), None)
|
|||
|
|
self._reply(200, {"ok": True})
|
|||
|
|
else:
|
|||
|
|
self._reply(404, {"detail": "not found"})
|
|||
|
|
|
|||
|
|
def do_GET(self): # noqa: N802
|
|||
|
|
if self.path == "/api/me":
|
|||
|
|
session = self.sessions.get(self._bearer())
|
|||
|
|
if session:
|
|||
|
|
self._reply(200, {"ok": True, **session})
|
|||
|
|
else:
|
|||
|
|
self._reply(401, {"detail": "未登录或登录已过期"})
|
|||
|
|
else:
|
|||
|
|
self._reply(404, {"detail": "not found"})
|
|||
|
|
|
|||
|
|
def log_message(self, *args): # 静音
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
|
|||
|
|
class AuthApiTest(VFSTestCase):
|
|||
|
|
"""auth 模块对真实 HTTP 的行为（经本地假服务端）。"""
|
|||
|
|
|
|||
|
|
def setUp(self):
|
|||
|
|
super().setUp()
|
|||
|
|
_AuthApiHandler.sessions.clear()
|
|||
|
|
_AuthApiHandler.mode.clear()
|
|||
|
|
self.server = ThreadingHTTPServer(("127.0.0.1", 0), _AuthApiHandler)
|
|||
|
|
thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|||
|
|
thread.start()
|
|||
|
|
self.addCleanup(self.server.shutdown)
|
|||
|
|
self.addCleanup(self.server.server_close)
|
|||
|
|
self.base = f"http://127.0.0.1:{self.server.server_address[1]}"
|
|||
|
|
|
|||
|
|
def test_login_success_normalizes_base_url(self):
|
|||
|
|
session = auth.login(self.base + "/", "admin", "admin123")
|
|||
|
|
self.assertTrue(session["token"])
|
|||
|
|
self.assertEqual(session["username"], "admin")
|
|||
|
|
self.assertEqual(session["base_url"], self.base)
|
|||
|
|
self.assertEqual(session["login_time"], "2026-09-15T10:00:00")
|
|||
|
|
|
|||
|
|
def test_login_bad_password(self):
|
|||
|
|
with self.assertRaises(auth.AuthError) as caught:
|
|||
|
|
auth.login(self.base, "admin", "wrong")
|
|||
|
|
self.assertIn("不正确", str(caught.exception))
|
|||
|
|
|
|||
|
|
def test_login_requires_credentials(self):
|
|||
|
|
with self.assertRaises(auth.AuthError):
|
|||
|
|
auth.login(self.base, "", "admin123")
|
|||
|
|
with self.assertRaises(auth.AuthError):
|
|||
|
|
auth.login(self.base, "admin", "")
|
|||
|
|
|
|||
|
|
def test_login_rejects_invalid_base_url(self):
|
|||
|
|
with self.assertRaises(auth.AuthError):
|
|||
|
|
auth.login("ftp://nope", "admin", "admin123")
|
|||
|
|
|
|||
|
|
def test_login_invalid_response_body(self):
|
|||
|
|
_AuthApiHandler.mode["login"] = "badjson"
|
|||
|
|
with self.assertRaises(auth.AuthError) as caught:
|
|||
|
|
auth.login(self.base, "admin", "admin123")
|
|||
|
|
self.assertIn("响应无效", str(caught.exception))
|
|||
|
|
|
|||
|
|
def test_login_response_without_token(self):
|
|||
|
|
_AuthApiHandler.mode["login"] = "notoken"
|
|||
|
|
with self.assertRaises(auth.AuthError) as caught:
|
|||
|
|
auth.login(self.base, "admin", "admin123")
|
|||
|
|
self.assertIn("token", str(caught.exception))
|
|||
|
|
|
|||
|
|
def test_login_network_error(self):
|
|||
|
|
# 停掉服务器再请求：连接层失败 → 「无法连接」
|
|||
|
|
base = self.base
|
|||
|
|
self.server.shutdown()
|
|||
|
|
with self.assertRaises(auth.AuthError) as caught:
|
|||
|
|
auth.login(base, "admin", "admin123")
|
|||
|
|
self.assertIn("无法连接服务器", str(caught.exception))
|
|||
|
|
|
|||
|
|
def test_verify_token_roundtrip(self):
|
|||
|
|
token = auth.login(self.base, "admin", "admin123")["token"]
|
|||
|
|
result = auth.verify_token(self.base, token)
|
|||
|
|
self.assertEqual(result["username"], "admin")
|
|||
|
|
self.assertEqual(result["base_url"], self.base)
|
|||
|
|
self.assertEqual(result["login_time"], "2026-09-15T10:00:00")
|
|||
|
|
|
|||
|
|
def test_verify_token_rejects_garbage(self):
|
|||
|
|
with self.assertRaises(auth.AuthError):
|
|||
|
|
auth.verify_token(self.base, "not-a-token")
|
|||
|
|
with self.assertRaises(auth.AuthError) as caught:
|
|||
|
|
auth.verify_token(self.base, " ")
|
|||
|
|
self.assertIn("凭据", str(caught.exception))
|
|||
|
|
|
|||
|
|
def test_logout_invalidates_token(self):
|
|||
|
|
token = auth.login(self.base, "admin", "admin123")["token"]
|
|||
|
|
self.assertTrue(auth.verify_token(self.base, token))
|
|||
|
|
auth.logout(self.base, token)
|
|||
|
|
with self.assertRaises(auth.AuthError):
|
|||
|
|
auth.verify_token(self.base, token)
|
|||
|
|
|
|||
|
|
def test_logout_is_best_effort(self):
|
|||
|
|
# 服务器已停 / token 无效都不抛
|
|||
|
|
self.server.shutdown()
|
|||
|
|
auth.logout(self.base, "whatever")
|
|||
|
|
auth.logout(self.base, "")
|
|||
|
|
|
|||
|
|
|
|||
|
|
class SettingsAuthTest(VFSTestCase):
|
|||
|
|
"""settings 新增键的缺省与持久化。"""
|
|||
|
|
|
|||
|
|
def test_defaults_include_auth_keys(self):
|
|||
|
|
base = client_paths.ensure_layout(self.path("data"))
|
|||
|
|
data = client_settings.load(base)
|
|||
|
|
self.assertEqual(data["auth_token"], "")
|
|||
|
|
self.assertEqual(data["auth_user"], "")
|
|||
|
|
|
|||
|
|
def test_auth_keys_roundtrip(self):
|
|||
|
|
base = client_paths.ensure_layout(self.path("data"))
|
|||
|
|
data = client_settings.load(base)
|
|||
|
|
data.update({"base_url": "http://10.0.0.5:9000",
|
|||
|
|
"auth_token": "tok", "auth_user": "admin"})
|
|||
|
|
client_settings.save(base, data)
|
|||
|
|
loaded = client_settings.load(base)
|
|||
|
|
self.assertEqual(loaded["auth_token"], "tok")
|
|||
|
|
self.assertEqual(loaded["auth_user"], "admin")
|
|||
|
|
|
|||
|
|
|
|||
|
|
class NetUtilTest(VFSTestCase):
|
|||
|
|
"""loopback 请求的代理绕过：系统代理黑掉 localhost 是"接口一直在
|
|||
|
|
请求中"的经典原因。"""
|
|||
|
|
|
|||
|
|
class _OkHandler(BaseHTTPRequestHandler):
|
|||
|
|
def do_GET(self): # noqa: N802
|
|||
|
|
body = b'{"ok": true}'
|
|||
|
|
self.send_response(200)
|
|||
|
|
self.send_header("Content-Length", str(len(body)))
|
|||
|
|
self.end_headers()
|
|||
|
|
self.wfile.write(body)
|
|||
|
|
|
|||
|
|
def log_message(self, *args):
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
def setUp(self):
|
|||
|
|
super().setUp()
|
|||
|
|
# 全局策略与 opener 缓存跨用例残留，每个用例复位
|
|||
|
|
self.addCleanup(netutil.configure, "", "")
|
|||
|
|
netutil.configure("", "")
|
|||
|
|
|
|||
|
|
def test_is_loopback(self):
|
|||
|
|
for url in ("http://localhost:8000/a", "http://127.0.0.1:8000/",
|
|||
|
|
"http://[::1]:8000/", "http://api.localhost/x",
|
|||
|
|
"HTTP://LOCALHOST/x"):
|
|||
|
|
self.assertTrue(netutil.is_loopback(url), url)
|
|||
|
|
for url in ("http://example.com/", "http://192.168.1.5:8000/",
|
|||
|
|
"not a url"):
|
|||
|
|
self.assertFalse(netutil.is_loopback(url), url)
|
|||
|
|
|
|||
|
|
def test_loopback_bypasses_system_proxy(self):
|
|||
|
|
import urllib.error
|
|||
|
|
import urllib.request
|
|||
|
|
|
|||
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), self._OkHandler)
|
|||
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|||
|
|
thread.start()
|
|||
|
|
self.addCleanup(server.shutdown)
|
|||
|
|
self.addCleanup(server.server_close)
|
|||
|
|
base = f"http://127.0.0.1:{server.server_address[1]}"
|
|||
|
|
|
|||
|
|
# 模拟系统代理指向一个死端口：默认 urlopen 会失败（走代理），
|
|||
|
|
# netutil 对 loopback 强制直连 → 成功
|
|||
|
|
old_http = os.environ.get("http_proxy")
|
|||
|
|
old_https = os.environ.get("https_proxy")
|
|||
|
|
os.environ["http_proxy"] = "http://127.0.0.1:9"
|
|||
|
|
os.environ["https_proxy"] = "http://127.0.0.1:9"
|
|||
|
|
try:
|
|||
|
|
# 代理环境变量已生效（默认 urllib 会把请求发往该代理）。
|
|||
|
|
# 注意：这里只查 getproxies()，绝不能真用默认 urlopen 发请求——
|
|||
|
|
# 那会把"含死代理的 opener"缓存进全局，殃及之后所有测试模块。
|
|||
|
|
proxies = urllib.request.getproxies()
|
|||
|
|
self.assertEqual(proxies.get("http"), "http://127.0.0.1:9")
|
|||
|
|
# netutil 对 loopback 强制直连 → 不受代理影响
|
|||
|
|
fresh = urllib.request.Request(base + "/ok")
|
|||
|
|
with netutil.urlopen(fresh, timeout=3) as response:
|
|||
|
|
self.assertEqual(response.status, 200)
|
|||
|
|
finally:
|
|||
|
|
if old_http is None:
|
|||
|
|
os.environ.pop("http_proxy", None)
|
|||
|
|
else:
|
|||
|
|
os.environ["http_proxy"] = old_http
|
|||
|
|
if old_https is None:
|
|||
|
|
os.environ.pop("https_proxy", None)
|
|||
|
|
else:
|
|||
|
|
os.environ["https_proxy"] = old_https
|
|||
|
|
|
|||
|
|
|
|||
|
|
class _FlakyHandler(BaseHTTPRequestHandler):
|
|||
|
|
"""前 ``fail_times`` 次请求回 503，之后回 200；记录收到的每个请求。
|
|||
|
|
|
|||
|
|
模块级而不是塞在测试类里：嵌套类的方法体里看不到外层类作用域，
|
|||
|
|
引用它得写全限定名，很别扭。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
fail_times = 0
|
|||
|
|
hits = []
|
|||
|
|
|
|||
|
|
def _handle(self):
|
|||
|
|
# 先把请求体读完再应答：不读就回错误码会让客户端在写 body 时
|
|||
|
|
# 撞上连接被重置（表现为 ConnectionResetError 而不是 HTTPError）
|
|||
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|||
|
|
if length:
|
|||
|
|
self.rfile.read(length)
|
|||
|
|
_FlakyHandler.hits.append((self.command, self.path))
|
|||
|
|
if len(_FlakyHandler.hits) <= _FlakyHandler.fail_times:
|
|||
|
|
self.send_response(503)
|
|||
|
|
self.send_header("Content-Length", "0")
|
|||
|
|
self.end_headers()
|
|||
|
|
return
|
|||
|
|
body = b'{"ok": true}'
|
|||
|
|
self.send_response(200)
|
|||
|
|
self.send_header("Content-Length", str(len(body)))
|
|||
|
|
self.end_headers()
|
|||
|
|
self.wfile.write(body)
|
|||
|
|
|
|||
|
|
do_GET = _handle
|
|||
|
|
do_HEAD = _handle
|
|||
|
|
do_POST = _handle
|
|||
|
|
|
|||
|
|
def log_message(self, *args):
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
|
|||
|
|
class NetPolicyTest(VFSTestCase):
|
|||
|
|
"""代理/CA 配置、幂等重试与设置落点（P1-5 / P1-7）。"""
|
|||
|
|
|
|||
|
|
def setUp(self):
|
|||
|
|
super().setUp()
|
|||
|
|
self.addCleanup(netutil.configure, "", "")
|
|||
|
|
netutil.configure("", "")
|
|||
|
|
_FlakyHandler.hits = []
|
|||
|
|
_FlakyHandler.fail_times = 0
|
|||
|
|
|
|||
|
|
def _serve(self, handler=None):
|
|||
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), handler or _FlakyHandler)
|
|||
|
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
|||
|
|
self.addCleanup(server.shutdown)
|
|||
|
|
self.addCleanup(server.server_close)
|
|||
|
|
return server, f"http://127.0.0.1:{server.server_address[1]}"
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _noop_sleep(_seconds):
|
|||
|
|
"""不真等退避时间（真实退避从 0.3s 起，两次重试就该等 0.9s）。"""
|
|||
|
|
|
|||
|
|
# ---- 重试策略 ---------------------------------------------------
|
|||
|
|
|
|||
|
|
def test_idempotent_get_retries_until_success(self):
|
|||
|
|
import urllib.request
|
|||
|
|
|
|||
|
|
_server, base = self._serve()
|
|||
|
|
_FlakyHandler.fail_times = 2 # 503、503、200
|
|||
|
|
request = urllib.request.Request(base + "/api/x")
|
|||
|
|
with netutil.open_with_retry(request, timeout=5, retries=2,
|
|||
|
|
sleep=self._noop_sleep) as response:
|
|||
|
|
self.assertEqual(response.status, 200)
|
|||
|
|
self.assertEqual(len(_FlakyHandler.hits), 3)
|
|||
|
|
|
|||
|
|
def test_post_is_never_retried(self):
|
|||
|
|
import urllib.error
|
|||
|
|
import urllib.request
|
|||
|
|
|
|||
|
|
_server, base = self._serve()
|
|||
|
|
_FlakyHandler.fail_times = 99
|
|||
|
|
request = urllib.request.Request(base + "/api/order", data=b"{}",
|
|||
|
|
method="POST")
|
|||
|
|
with self.assertRaises(urllib.error.HTTPError) as caught:
|
|||
|
|
netutil.open_with_retry(request, timeout=5, retries=3,
|
|||
|
|
sleep=self._noop_sleep)
|
|||
|
|
self.assertEqual(caught.exception.code, 503)
|
|||
|
|
# 自动重发 POST 可能把"下单"变成两单：只允许发一次
|
|||
|
|
self.assertEqual(len(_FlakyHandler.hits), 1)
|
|||
|
|
|
|||
|
|
def test_retries_exhausted_raises_last_error(self):
|
|||
|
|
import urllib.error
|
|||
|
|
import urllib.request
|
|||
|
|
|
|||
|
|
_server, base = self._serve()
|
|||
|
|
_FlakyHandler.fail_times = 99
|
|||
|
|
request = urllib.request.Request(base + "/api/x")
|
|||
|
|
with self.assertRaises(urllib.error.HTTPError):
|
|||
|
|
netutil.open_with_retry(request, timeout=5, retries=2,
|
|||
|
|
sleep=self._noop_sleep)
|
|||
|
|
self.assertEqual(len(_FlakyHandler.hits), 3) # 1 次 + 2 次重试
|
|||
|
|
|
|||
|
|
def test_non_retryable_status_is_not_retried(self):
|
|||
|
|
import urllib.error
|
|||
|
|
import urllib.request
|
|||
|
|
|
|||
|
|
class _NotFound(BaseHTTPRequestHandler):
|
|||
|
|
hits = 0
|
|||
|
|
|
|||
|
|
def do_GET(self): # noqa: N802
|
|||
|
|
_NotFound.hits += 1
|
|||
|
|
self.send_response(404)
|
|||
|
|
self.send_header("Content-Length", "0")
|
|||
|
|
self.end_headers()
|
|||
|
|
|
|||
|
|
def log_message(self, *args):
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
_server, base = self._serve(_NotFound)
|
|||
|
|
request = urllib.request.Request(base + "/api/x")
|
|||
|
|
with self.assertRaises(urllib.error.HTTPError):
|
|||
|
|
netutil.open_with_retry(request, timeout=5, retries=3,
|
|||
|
|
sleep=self._noop_sleep)
|
|||
|
|
self.assertEqual(_NotFound.hits, 1) # 404 重试没有意义
|
|||
|
|
|
|||
|
|
def test_transport_error_is_retried(self):
|
|||
|
|
"""连接层失败（拒绝连接/DNS/超时）也重试——服务器刚起来那几秒。"""
|
|||
|
|
import urllib.error
|
|||
|
|
import urllib.request
|
|||
|
|
|
|||
|
|
attempts = []
|
|||
|
|
|
|||
|
|
class _Response:
|
|||
|
|
status = 200
|
|||
|
|
|
|||
|
|
def __enter__(self):
|
|||
|
|
return self
|
|||
|
|
|
|||
|
|
def __exit__(self, *exc):
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
def flaky(request, *, timeout):
|
|||
|
|
attempts.append(1)
|
|||
|
|
if len(attempts) < 3:
|
|||
|
|
raise urllib.error.URLError("connection refused")
|
|||
|
|
return _Response()
|
|||
|
|
|
|||
|
|
original = netutil.urlopen
|
|||
|
|
netutil.urlopen = flaky
|
|||
|
|
self.addCleanup(setattr, netutil, "urlopen", original)
|
|||
|
|
request = urllib.request.Request("http://127.0.0.1:1/api/x")
|
|||
|
|
with netutil.open_with_retry(request, timeout=5, retries=2,
|
|||
|
|
sleep=self._noop_sleep) as response:
|
|||
|
|
self.assertEqual(response.status, 200)
|
|||
|
|
self.assertEqual(len(attempts), 3)
|
|||
|
|
|
|||
|
|
# ---- 代理与 CA ---------------------------------------------------
|
|||
|
|
|
|||
|
|
def test_explicit_proxy_for_remote_and_loopback_stays_direct(self):
|
|||
|
|
netutil.configure("http://proxy.corp:3128")
|
|||
|
|
remote = {type(h).__name__: h for h in netutil._opener(False).handlers}
|
|||
|
|
self.assertEqual(
|
|||
|
|
remote["ProxyHandler"].proxies,
|
|||
|
|
{"http": "http://proxy.corp:3128",
|
|||
|
|
"https": "http://proxy.corp:3128"},
|
|||
|
|
)
|
|||
|
|
# loopback 即使配了代理也直连：opener 里干脆没有代理handler
|
|||
|
|
# （ProxyHandler({}) 不生成 *_open 方法，build_opener 会丢掉它）
|
|||
|
|
local = {type(h).__name__ for h in netutil._opener(True).handlers}
|
|||
|
|
self.assertNotIn("ProxyHandler", local)
|
|||
|
|
|
|||
|
|
def test_missing_ca_file_is_reported_and_not_applied(self):
|
|||
|
|
error = netutil.configure("", self.path("nope.pem"))
|
|||
|
|
self.assertIsNotNone(error)
|
|||
|
|
self.assertIn("不存在", error)
|
|||
|
|
self.assertEqual(netutil.current(), ("", "")) # 没生效（不假装配上了）
|
|||
|
|
|
|||
|
|
def test_ca_bundle_valid_pem_installs_https_handler(self):
|
|||
|
|
import ssl
|
|||
|
|
|
|||
|
|
source = ssl.get_default_verify_paths().cafile
|
|||
|
|
if not source or not os.path.isfile(source):
|
|||
|
|
self.skipTest("本机没有可复制的默认 CA 文件（Windows 无 cafile）")
|
|||
|
|
bundle = self.path("ca.pem")
|
|||
|
|
with open(source, "rb") as src, open(bundle, "wb") as dst:
|
|||
|
|
dst.write(src.read())
|
|||
|
|
self.assertIsNone(netutil.configure("", bundle))
|
|||
|
|
handlers = netutil._opener(False).handlers
|
|||
|
|
self.assertTrue(any(type(h).__name__ == "HTTPSHandler" for h in handlers))
|
|||
|
|
|
|||
|
|
def test_bad_ca_content_degrades_without_crashing(self):
|
|||
|
|
"""CA 文件在但内容是垃圾：退回系统信任库，不能把客户端搞挂。"""
|
|||
|
|
from svfs_client import diagnostics
|
|||
|
|
|
|||
|
|
bundle = self.path("junk.pem")
|
|||
|
|
with open(bundle, "w", encoding="utf-8") as handle:
|
|||
|
|
handle.write("这不是一个 PEM 证书\n")
|
|||
|
|
self.assertIsNone(netutil.configure("", bundle)) # 文件存在 → 配置接受
|
|||
|
|
netutil._opener(False) # 构建 opener 不抛（退回系统信任库）
|
|||
|
|
self.assertIn("CA 证书无法加载", "\n".join(diagnostics.recent()))
|
|||
|
|
|
|||
|
|
def test_apply_settings_reads_settings_json(self):
|
|||
|
|
base = client_paths.ensure_layout(self.path("data"))
|
|||
|
|
data = client_settings.load(base)
|
|||
|
|
data["proxy"] = "10.0.0.9:8080" # 只写 host:port 也认
|
|||
|
|
client_settings.save(base, data)
|
|||
|
|
self.assertIsNone(netutil.apply_settings(base))
|
|||
|
|
self.assertEqual(netutil.current()[0], "http://10.0.0.9:8080")
|
|||
|
|
|
|||
|
|
def test_settings_net_options_fall_back_on_bad_values(self):
|
|||
|
|
base = client_paths.ensure_layout(self.path("data"))
|
|||
|
|
self.assertEqual(client_settings.net_options(base),
|
|||
|
|
(client_settings.NET_TIMEOUT_DEFAULT,
|
|||
|
|
client_settings.NET_RETRIES_DEFAULT))
|
|||
|
|
data = client_settings.load(base)
|
|||
|
|
data.update({"net_timeout": "很快", "net_retries": -3})
|
|||
|
|
client_settings.save(base, data)
|
|||
|
|
# 坏配置不该让网络不可用：一律落回默认值
|
|||
|
|
self.assertEqual(client_settings.net_options(base),
|
|||
|
|
(client_settings.NET_TIMEOUT_DEFAULT,
|
|||
|
|
client_settings.NET_RETRIES_DEFAULT))
|
|||
|
|
data.update({"net_timeout": 0, "net_retries": 0})
|
|||
|
|
client_settings.save(base, data)
|
|||
|
|
# 超时 0 没意义（会把工作线程钉死）→ 用默认；重试 0 = 明确不重试
|
|||
|
|
self.assertEqual(client_settings.net_options(base),
|
|||
|
|
(client_settings.NET_TIMEOUT_DEFAULT, 0))
|
|||
|
|
|
|||
|
|
def test_bridge_reads_net_options_from_provider(self):
|
|||
|
|
"""桥现读设置（provider 抛错时退回默认值，坏回调不该让桥瘫痪）。"""
|
|||
|
|
from svfs_client.bridge import NET_TIMEOUT, SvfsBridge
|
|||
|
|
|
|||
|
|
bridge = SvfsBridge("com.example.demo", ["net"],
|
|||
|
|
self.path("data/storage.json"),
|
|||
|
|
base_url_provider=lambda: "http://127.0.0.1:1",
|
|||
|
|
net_options_provider=lambda: (3, 2))
|
|||
|
|
self.addCleanup(bridge.shutdown)
|
|||
|
|
self.assertEqual(bridge._net_options(), (3, 2))
|
|||
|
|
bridge._net_options_provider = lambda: 1 / 0
|
|||
|
|
self.assertEqual(bridge._net_options(), (NET_TIMEOUT, 0))
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
unittest.main()
|