#!/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)