v2: SQLite + 多路流 + 鉴权 + 导入/删除 + 资源本地化 + 模块化重构

test_write.py

@@ -0,0 +1,85 @@
+import urllib.request, urllib.error, http.cookiejar, json, sys, os, io, uuid
+
+BASE = "http://127.0.0.1:3000"
+cj = http.cookiejar.CookieJar()
+op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
+
+def call(path, method="GET", data=None):
+    body = json.dumps(data).encode() if data is not None else None
+    req = urllib.request.Request(BASE + path, data=body, method=method)
+    if body: req.add_header("Content-Type", "application/json")
+    try:
+        r = op.open(req, timeout=30)
+        return r.status, r.read()
+    except urllib.error.HTTPError as e:
+        return e.code, e.read()
+
+results = []
+def check(name, cond, extra=""):
+    results.append((name, cond)); print(("PASS" if cond else "FAIL"), name, extra)
+
+# 登录
+call("/api/auth/login", "POST", {"password": "test123"})
+
+# ---------- 导入测试 ----------
+tmp_file = r"F:\music_online\demo\tmp\test_import.mp3"
+boundary = "----test" + uuid.uuid4().hex
+with open(tmp_file, "rb") as f:
+    audio_data = f.read()
+
+title = "测试导入歌"
+artist = "测试作者"
+parts = []
+def add_field(name, value):
+    parts.append(("--" + boundary + "\r\n").encode())
+    parts.append(('Content-Disposition: form-data; name="%s"\r\n\r\n' % name).encode())
+    parts.append((value + "\r\n").encode())
+def add_file(name, filename, content):
+    parts.append(("--" + boundary + "\r\n").encode())
+    parts.append(('Content-Disposition: form-data; name="%s"; filename="%s"\r\n' % (name, filename)).encode())
+    parts.append(("Content-Type: audio/mpeg\r\n\r\n").encode())
+    parts.append(content)
+    parts.append(b"\r\n")
+
+add_field("title", title)
+add_field("artist", artist)
+add_file("audio", "测试导入歌.mp3", audio_data)
+parts.append(("--" + boundary + "--\r\n").encode())
+body = b"".join(parts)
+
+req = urllib.request.Request(BASE + "/api/music/import", data=body, method="POST")
+req.add_header("Content-Type", "multipart/form-data; boundary=" + boundary)
+try:
+    r = op.open(req, timeout=60)
+    d = json.loads(r.read())
+    check("导入音频", r.status == 200 and d.get("ok"), str(d)[:120])
+    imported_file = d.get("file")
+    print("   导入文件名:", imported_file)
+    print("   元数据:", json.dumps(d.get("song", {}), ensure_ascii=False)[:200])
+except urllib.error.HTTPError as e:
+    check("导入音频", False, f"HTTP {e.code}: {e.read()[:200]}")
+    imported_file = None
+
+# 确认资源数增加
+st, body = call("/api/resources")
+d = json.loads(body)
+check("资源数变为6", len(d["list"]) == 6, f"{len(d['list'])}")
+
+# ---------- 删除测试 ----------
+if imported_file:
+    st, body = call("/api/music/delete", "POST", {"file": imported_file})
+    d = json.loads(body)
+    check("删除音频", st == 200 and d.get("ok"), str(d)[:120])
+
+    # 确认文件没了
+    fp = os.path.join(r"F:\music_online\demo\audio", imported_file)
+    check("音频文件已删", not os.path.exists(fp))
+    # 确认资源数恢复
+    st, body = call("/api/resources")
+    d = json.loads(body)
+    check("资源数恢复为5", len(d["list"]) == 5, f"{len(d['list'])}")
+else:
+    check("删除音频", False, "无导入文件")
+
+print("\n=== 汇总:", sum(1 for _,c in results if c), "/", len(results), "通过 ===")
+sys.exit(0 if all(c for _,c in results) else 1)