zijunchiang
87aa979abd
版本号：
- 1.5.11 越界（规则 3 规定 patch ≤ 10）→ 修正为 1.6.0（进位到 minor、patch 归 0）
- versionCode 43 不变：43 从未对外发布，本次修正只是把命名拉回约定
合并支线（原在工作区根目录、App 仓库之外，DESIGN.md 自称"支线"）：
- selfhost-source/server/ 自建歌曲源服务器（纯 Python 标准库、零 pip 依赖）
- selfhost-source/deploy/、DESIGN.md、LIBRARY.md、build_inventory.py
- selfhost-source/inventory/ 小清单（playlists.csv、song-stats.md 等）
- ⚠️ inventory 生成物（songs-*.csv / singers-top.csv，共 84MB）不入库，
已写入 .gitignore；需要时用 build_inventory.py 依据 muse.db 重新生成
同时入库的 App 侧支线改动（此前为工作树未提交状态）：
- Whoami.kt（whoami 设备心跳）、DeviceIdentity.kt（设备身份伪装）
- PushDownload.kt / PushDownloadClient.kt（点歌推送）
- AdVideoFilter 带 query/fragment 的地址闸门；KTV_BRIDGE_API 11 → 13
- 修 ktv_api.js 的 global.KtvBridgeApiVersion（12 → 13，与注释及 Kotlin 两处校验一致）
- 对应单测：WhoamiTest / DeviceIdentityTest / PushDownloadTest / AdVideoFilterTest（共 39 例）
文档：
- DESIGN.md 的"在 App 仓库之外"约定改为"已并入仓库统一版本化"
- AGENTS.md 版本号行更新为 1.6.0/43，并新增支线并入说明
验证：assembleRelease + testReleaseUnitTest 全过（39 例）
165 lines
7.0 KiB
Python
165 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
||
"""接口级自测：直接调用路由处理函数（**不监听端口、不启动服务**），验证 App 契约。
|
||
|
||
覆盖：
|
||
* POST /api.php → status=success，且 trial_times=0（App 不会因额度限制源数量）
|
||
* GET /i.php → code=200 + token
|
||
* GET /music/do.php → 已就绪：code=200 且 data 以 /<编号>.ts 结尾 + 域名（非裸 IP）
|
||
未下载：code=404（**绝不能用 403**——那会被 App 当成源站风控）
|
||
* GET /media/<编号>.ts → 200 与 Range/206
|
||
"""
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from ktv import service
|
||
from ktv.config import Config
|
||
from ktv.store import Store
|
||
from ktv.web import Request
|
||
|
||
|
||
def req(method, path, query=None, body=b"", headers=None):
|
||
return Request(method, path, query or {}, headers or {}, body)
|
||
|
||
|
||
class TestAppContract(unittest.TestCase):
|
||
def tearDown(self):
|
||
# 显式关连接：否则 ResourceWarning（unclosed database）会污染验收输出
|
||
try:
|
||
self.store.conn.close()
|
||
except Exception:
|
||
pass
|
||
|
||
def setUp(self):
|
||
self.tmp = tempfile.mkdtemp()
|
||
self.cfg = Config(os.path.join(self.tmp, "c.json"))
|
||
self.media = os.path.join(self.tmp, "media")
|
||
os.makedirs(os.path.join(self.media, "tmp"), exist_ok=True)
|
||
self.cfg.update({"public_media_base": "http://ktv.example.com:6677/media",
|
||
"media_dir": self.media, "data_dir": self.tmp})
|
||
self.store = Store(os.path.join(self.tmp, "t.db"))
|
||
self.store.import_library([(7009126, "一帘好梦", "国语", "", "胡云鹏", "7009126.ts", 1060, "")],
|
||
"test")
|
||
self.router = service.build_router(self.cfg, self.store, None)
|
||
|
||
def call(self, r):
|
||
return self.router.dispatch(r)
|
||
|
||
def test_api_php(self):
|
||
resp = self.call(req("POST", "/api.php", {"action": "get_config"}, b"action=get_config&device_id=abc"))
|
||
self.assertEqual(resp.status, 200)
|
||
import json
|
||
data = json.loads(resp.body)
|
||
self.assertEqual(data["status"], "success")
|
||
self.assertEqual(data["config"]["trial_times"], 0)
|
||
self.assertFalse(data["config"]["network_verification"])
|
||
|
||
def test_i_php(self):
|
||
import json
|
||
resp = self.call(req("GET", "/i.php", {"appid": "x", "mac": "800b510bf4fd_2d7bb4a7d7f7"}))
|
||
data = json.loads(resp.body)
|
||
self.assertEqual(data["code"], 200)
|
||
self.assertTrue(data["token"])
|
||
|
||
def test_do_php_ready(self):
|
||
import json
|
||
# 先登记为已下载完成
|
||
self.store.enqueue([7009126])
|
||
self.store.claim_one(7009126)
|
||
self.store.finish(7009126, True, bytes_=7 << 20, path=os.path.join(self.media, "7009126.ts"))
|
||
resp = self.call(req("GET", "/music/do.php", {"musicno": "7009126", "ls": "1", "resolution": "720"}))
|
||
data = json.loads(resp.body)
|
||
self.assertEqual(data["code"], 200)
|
||
self.assertEqual(data["msg"], "SUCCESS")
|
||
url = data["data"]
|
||
self.assertTrue(url.endswith("/media/7009126.ts"), url)
|
||
self.assertFalse(url.split("//", 1)[1].split("/")[0].replace(":", "").replace(".", "").isdigit(),
|
||
"媒体地址不能是裸 IP（App 会拒收）")
|
||
|
||
def test_do_php_not_ready_returns_404_not_403(self):
|
||
import json
|
||
resp = self.call(req("GET", "/music/do.php", {"musicno": "7009126"}))
|
||
data = json.loads(resp.body)
|
||
self.assertEqual(data["code"], 404) # 关键：不能是 403（403=源站风控语义）
|
||
self.assertNotEqual(data["code"], 403)
|
||
|
||
def test_do_php_unknown_tid(self):
|
||
import json
|
||
resp = self.call(req("GET", "/music/do.php", {"musicno": "9999999"}))
|
||
self.assertEqual(json.loads(resp.body)["code"], 404)
|
||
|
||
def test_media_range(self):
|
||
path = os.path.join(self.media, "7009126.ts")
|
||
with open(path, "wb") as fh:
|
||
fh.write(b"\x47" + b"0" * 375 + b"\x47" + b"0" * 100)
|
||
full = self.call(req("GET", "/media/7009126.ts"))
|
||
self.assertEqual(full.status, 200)
|
||
self.assertEqual(full.headers.get("Accept-Ranges"), "bytes")
|
||
# 媒体改为**流式发送**：断言长度/偏移字段而不是 body（避免 75MB 整块读进内存）
|
||
part = self.call(req("GET", "/media/7009126.ts", headers={"range": "bytes=0-99"}))
|
||
self.assertEqual(part.status, 206)
|
||
self.assertEqual(part.file_len, 100)
|
||
self.assertEqual(part.file_start, 0)
|
||
self.assertTrue(part.file_path.endswith("7009126.ts"))
|
||
self.assertIn("bytes 0-99/", part.headers.get("Content-Range", ""))
|
||
# 请求头必须大小写不敏感（否则 App 的 Range 断点续传会退化成全量下载）
|
||
upper = self.call(req("GET", "/media/7009126.ts", headers={"Range": "bytes=10-19"}))
|
||
self.assertEqual(upper.status, 206)
|
||
self.assertEqual(upper.file_start, 10)
|
||
self.assertEqual(upper.file_len, 10)
|
||
|
||
def test_media_missing(self):
|
||
self.assertEqual(self.call(req("GET", "/media/1234567.ts")).status, 404)
|
||
|
||
|
||
class TestAdminContract(unittest.TestCase):
|
||
def tearDown(self):
|
||
# 显式关连接：否则 ResourceWarning（unclosed database）会污染验收输出
|
||
try:
|
||
self.store.conn.close()
|
||
except Exception:
|
||
pass
|
||
|
||
def setUp(self):
|
||
from ktv import admin
|
||
self.tmp = tempfile.mkdtemp()
|
||
self.cfg = Config(os.path.join(self.tmp, "c.json"))
|
||
self.cfg.update({"media_dir": os.path.join(self.tmp, "media"), "data_dir": self.tmp})
|
||
self.store = Store(os.path.join(self.tmp, "t.db"))
|
||
self.store.import_library([(1, "A", "国语", "", "s", "1.ts", 5, ""),
|
||
(2, "B", "粤语", "", "s", "2.ts", 9, "")], "v")
|
||
self.router = admin.build_router(self.cfg, self.store, None)
|
||
|
||
def test_stats(self):
|
||
import json
|
||
d = json.loads(self.router.dispatch(req("GET", "/api/stats")).body)
|
||
self.assertEqual(d["songs"], 2)
|
||
|
||
def test_songs_list_and_filter(self):
|
||
import json
|
||
d = json.loads(self.router.dispatch(req("GET", "/api/songs", {"order": "tid", "desc": "0"})).body)
|
||
self.assertEqual(d["total"], 2)
|
||
self.assertEqual(d["items"][0]["tid"], 1)
|
||
d2 = json.loads(self.router.dispatch(req("GET", "/api/songs", {"lang": "粤语"})).body)
|
||
self.assertEqual(d2["total"], 1)
|
||
|
||
def test_enqueue_rejects_unknown(self):
|
||
import json
|
||
r = json.loads(self.router.dispatch(req("POST", "/api/download/enqueue")).body)
|
||
self.assertFalse(r["ok"])
|
||
|
||
def test_settings_concurrency_cap(self):
|
||
import json
|
||
d = json.loads(self.router.dispatch(
|
||
req("POST", "/api/settings", body=b'{"concurrency": 50}')).body)
|
||
self.assertTrue(d["ok"])
|
||
self.assertEqual(d["config"]["concurrency"], 10)
|
||
self.assertTrue(any("10" in w for w in d["warnings"]))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|