fix(update): 迁移多应用接口并统一新域名
This commit is contained in:
1 parent
eac8a3a698
commit
04a7909dbf
13 files changed
+241
-264
No files matched your search
@@ -15,8 +15,9 @@ from pathlib import Path
|
||||
|
||||
|
||||
DEFAULT_ENDPOINTS = (
|
||||
"https://xstats-apps.12306.work/api/v1/releases/current",
|
||||
"https://x-stats.china.12306.work/api/v1/releases/current",
|
||||
"https://apps.12306.work/api/v1/apps/xstats/releases/current",
|
||||
"https://apps-api.xiai.me/api/v1/apps/xstats/releases/current",
|
||||
"https://apps.china.12306.work/api/v1/apps/xstats/releases/current",
|
||||
)
|
||||
|
||||
|
||||
@@ -27,9 +28,9 @@ MAXIMUM_FEED_BYTES = 256 * 1024
|
||||
|
||||
def sparkle_feed_url(endpoint: str, query: str = "") -> str:
|
||||
parts = urlsplit(endpoint)
|
||||
if not parts.path.endswith("/api/v1/releases/current"):
|
||||
if parts.path != "/api/v1/apps/xstats/releases/current":
|
||||
raise ValueError(f"无法推导 Sparkle 更新源:{endpoint}")
|
||||
return urlunsplit((parts.scheme, parts.netloc, parts.path.removesuffix("releases/current") + "update/appcast.xml", query, ""))
|
||||
return urlunsplit((parts.scheme, parts.netloc, "/api/v1/apps/xstats/update/appcast.xml", query, ""))
|
||||
|
||||
|
||||
def check_sparkle_endpoints(endpoints: list[str]) -> None:
|
||||
@@ -70,6 +71,9 @@ def publish(appcast: Path, endpoints: list[str], token: str) -> None:
|
||||
raise ValueError("XSTATS_RELEASE_TOKEN 不能为空")
|
||||
if not endpoints:
|
||||
raise ValueError("至少需要一个版本发布接口")
|
||||
# 先校验全部发布路径,避免后续地址配置错误时已切换部分区域的版本。
|
||||
for endpoint in endpoints:
|
||||
sparkle_feed_url(endpoint)
|
||||
body = appcast.read_bytes()
|
||||
manifest = json.loads(body)
|
||||
if not isinstance(manifest, dict) or not manifest.get("version"):
|
||||
|
||||
+46
-14
@@ -18,21 +18,37 @@ import publish_api
|
||||
|
||||
|
||||
class PublishAPITest(unittest.TestCase):
|
||||
def test_defaults_to_both_regional_endpoints(self) -> None:
|
||||
def test_defaults_to_all_regional_endpoints(self) -> None:
|
||||
self.assertEqual(publish_api.configured_endpoints({}), [
|
||||
"https://xstats-apps.12306.work/api/v1/releases/current",
|
||||
"https://x-stats.china.12306.work/api/v1/releases/current",
|
||||
"https://apps.12306.work/api/v1/apps/xstats/releases/current",
|
||||
"https://apps-api.xiai.me/api/v1/apps/xstats/releases/current",
|
||||
"https://apps.china.12306.work/api/v1/apps/xstats/releases/current",
|
||||
])
|
||||
|
||||
def test_sparkle_gate_checks_both_regions_without_installation_data(self) -> None:
|
||||
def test_sparkle_feed_uses_fixed_xstats_path(self) -> None:
|
||||
endpoint = "https://example.test/api/v1/apps/xstats/releases/current?ignored=1#fragment"
|
||||
self.assertEqual(publish_api.sparkle_feed_url(endpoint, "current_version=release-probe"),
|
||||
"https://example.test/api/v1/apps/xstats/update/appcast.xml?current_version=release-probe")
|
||||
|
||||
def test_sparkle_feed_rejects_unrecognized_release_paths(self) -> None:
|
||||
for path in ("/releases/current", "/api/v1/releases/current",
|
||||
"/api/v1/apps/another-app/releases/current", "/proxy/api/v1/apps/xstats/releases/current",
|
||||
"/api/v1/apps/releases/current",
|
||||
"/api/v1/apps/xstats/releases/current/", "/api/v1/apps/xstats/update/check",
|
||||
"/api/v1/apps/UPPER/releases/current", "/api/v1/apps/xstats/other/releases/current"):
|
||||
with self.subTest(path=path), self.assertRaises(ValueError):
|
||||
publish_api.sparkle_feed_url(f"https://example.test{path}")
|
||||
|
||||
def test_sparkle_gate_checks_all_endpoints_without_installation_data(self) -> None:
|
||||
errors = [urllib.error.HTTPError(url, 400, "invalid request", {"Cache-Control": "no-store"}, io.BytesIO(b""))
|
||||
for url in publish_api.DEFAULT_ENDPOINTS]
|
||||
with patch("publish_api.urllib.request.urlopen", side_effect=errors) as send:
|
||||
publish_api.check_sparkle_endpoints(list(publish_api.DEFAULT_ENDPOINTS))
|
||||
urls = [call.args[0].full_url for call in send.call_args_list]
|
||||
self.assertEqual(urls, [
|
||||
"https://xstats-apps.12306.work/api/v1/update/appcast.xml",
|
||||
"https://x-stats.china.12306.work/api/v1/update/appcast.xml",
|
||||
"https://apps.12306.work/api/v1/apps/xstats/update/appcast.xml",
|
||||
"https://apps-api.xiai.me/api/v1/apps/xstats/update/appcast.xml",
|
||||
"https://apps.china.12306.work/api/v1/apps/xstats/update/appcast.xml",
|
||||
])
|
||||
for call in send.call_args_list:
|
||||
self.assertNotIn("installation_id", call.args[0].full_url)
|
||||
@@ -49,7 +65,7 @@ class PublishAPITest(unittest.TestCase):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
feed = Path(directory) / "XStats-1.0.0-AppleSilicon.xml"
|
||||
feed.write_bytes(b"<rss>signed</rss>")
|
||||
served = [b"<rss>signed</rss>", b"<rss>signed</rss>"]
|
||||
served = [b"<rss>signed</rss>"] * 3
|
||||
responses = [contextlib.nullcontext(type("Response", (), {"status": 200, "read": lambda self, limit, body=body: body})())
|
||||
for body in served]
|
||||
with patch("publish_api.urllib.request.urlopen", side_effect=responses) as send, \
|
||||
@@ -57,8 +73,9 @@ class PublishAPITest(unittest.TestCase):
|
||||
publish_api.verify_sparkle_feeds(feed, list(publish_api.DEFAULT_ENDPOINTS))
|
||||
urls = [call.args[0].full_url for call in send.call_args_list]
|
||||
self.assertEqual(urls, [
|
||||
"https://xstats-apps.12306.work/api/v1/update/appcast.xml?current_version=release-probe&installation_id=" + "0" * 64,
|
||||
"https://x-stats.china.12306.work/api/v1/update/appcast.xml?current_version=release-probe&installation_id=" + "0" * 64,
|
||||
"https://apps.12306.work/api/v1/apps/xstats/update/appcast.xml?current_version=release-probe&installation_id=" + "0" * 64,
|
||||
"https://apps-api.xiai.me/api/v1/apps/xstats/update/appcast.xml?current_version=release-probe&installation_id=" + "0" * 64,
|
||||
"https://apps.china.12306.work/api/v1/apps/xstats/update/appcast.xml?current_version=release-probe&installation_id=" + "0" * 64,
|
||||
])
|
||||
for call in send.call_args_list:
|
||||
self.assertFalse(call.args[0].has_header("Authorization"))
|
||||
@@ -70,9 +87,9 @@ class PublishAPITest(unittest.TestCase):
|
||||
stale = type("Response", (), {"status": 200, "read": lambda self, limit: b"<rss>previous</rss>"})()
|
||||
ok = type("Response", (), {"status": 200, "read": lambda self, limit: b"<rss>signed</rss>"})()
|
||||
with patch("publish_api.urllib.request.urlopen",
|
||||
side_effect=[contextlib.nullcontext(ok), contextlib.nullcontext(stale)]), \
|
||||
side_effect=[contextlib.nullcontext(ok), contextlib.nullcontext(ok), contextlib.nullcontext(stale)]), \
|
||||
contextlib.redirect_stdout(io.StringIO()):
|
||||
with self.assertRaisesRegex(RuntimeError, "x-stats.china.12306.work"):
|
||||
with self.assertRaisesRegex(RuntimeError, "apps.china.12306.work"):
|
||||
publish_api.verify_sparkle_feeds(feed, list(publish_api.DEFAULT_ENDPOINTS))
|
||||
|
||||
def test_sparkle_gate_rejects_old_server_or_cached_validation(self) -> None:
|
||||
@@ -82,6 +99,20 @@ class PublishAPITest(unittest.TestCase):
|
||||
with self.assertRaises(RuntimeError):
|
||||
publish_api.check_sparkle_endpoints(list(publish_api.DEFAULT_ENDPOINTS))
|
||||
|
||||
def test_publish_validates_all_paths_before_sending_any_request(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = Path(directory) / "appcast.json"
|
||||
path.write_text(json.dumps({"version": "1.0.0"}), encoding="utf-8")
|
||||
valid = "https://example.test/api/v1/apps/xstats/releases/current"
|
||||
for invalid in ("/api/v1/releases/current", "/api/v1/apps/another-app/releases/current",
|
||||
"/proxy/api/v1/apps/xstats/releases/current"):
|
||||
with self.subTest(path=invalid), patch("publish_api.urllib.request.urlopen") as send, \
|
||||
contextlib.redirect_stdout(io.StringIO()):
|
||||
send.return_value.__enter__.return_value.status = 204
|
||||
with self.assertRaises(ValueError):
|
||||
publish_api.publish(path, [valid, f"https://example.test{invalid}"], "release-secret")
|
||||
send.assert_not_called()
|
||||
|
||||
def test_sends_manifest_with_bearer_token(self) -> None:
|
||||
received = []
|
||||
|
||||
@@ -117,15 +148,16 @@ class PublishAPITest(unittest.TestCase):
|
||||
}
|
||||
path = Path(directory) / "appcast.json"
|
||||
path.write_text(json.dumps(manifest), encoding="utf-8")
|
||||
base = f"http://127.0.0.1:{server.server_port}"
|
||||
endpoints = [f"http://{host}:{server.server_port}/api/v1/apps/xstats/releases/current"
|
||||
for host in ("127.0.0.1", "localhost", "127.0.0.1")]
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
publish_api.publish(path, [f"{base}/global", f"{base}/china"], "release-secret")
|
||||
publish_api.publish(path, endpoints, "release-secret")
|
||||
finally:
|
||||
server.shutdown()
|
||||
thread.join()
|
||||
server.server_close()
|
||||
|
||||
self.assertEqual([item["path"] for item in received], ["/global", "/china"])
|
||||
self.assertEqual([item["path"] for item in received], ["/api/v1/apps/xstats/releases/current"] * 3)
|
||||
for item in received:
|
||||
self.assertEqual(item["authorization"], "Bearer release-secret")
|
||||
self.assertEqual(item["content_type"], "application/json")
|
||||
|
||||
Reference in new issue
Block a user