feat(scripts): 添加 YouTube Studio 群组名到 entity_id 批量解析脚本
支持 Playwright 自动捕获鉴权、requests 回放查询、Excel/CSV 输出，包含在线捕获和离线回放两种模式
This commit is contained in:
274
tests/test_lookup_groups.py
Normal file
274
tests/test_lookup_groups.py
Normal file
@@ -0,0 +1,274 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""lookup_groups.py 测试。
|
||||
|
||||
覆盖不依赖真实浏览器/登录态/网络的全部逻辑：
|
||||
parse_owner_id / parse_names_payload / load_names URL 与名单解析
|
||||
build_body / pick_match / search / resolve 请求构造、匹配、多所有者归并
|
||||
validate_bundle / merge_bundles 套件校验与合并
|
||||
write_out csv / xlsx 输出
|
||||
selftest() / --selftest / 无参 CLI 内置自测与引导输出
|
||||
|
||||
capture_bundles() 在线捕获需已登录浏览器会话，不在自动化测试范围。
|
||||
"""
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from conftest import LOOKUP_SCRIPT, run_script
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 测试替身
|
||||
# ---------------------------------------------------------------------------
|
||||
class FakeResp:
|
||||
def __init__(self, code, payload=None):
|
||||
self.status_code = code
|
||||
self._payload = payload
|
||||
|
||||
def json(self):
|
||||
if self._payload is None:
|
||||
raise ValueError("no json")
|
||||
return self._payload
|
||||
|
||||
|
||||
class FakeSession:
|
||||
"""按 `模板marker::query` 路由响应，可区分不同所有者的套件。
|
||||
|
||||
routing 键形如 "O1::名字"；未命中键返回空结果（NF）。
|
||||
"""
|
||||
|
||||
def __init__(self, routing):
|
||||
self.routing = routing
|
||||
self.calls = []
|
||||
|
||||
def post(self, url, json=None, headers=None, timeout=None):
|
||||
self.calls.append((url, json, headers))
|
||||
body = json or {}
|
||||
key = f"{body.get('marker', '')}::{body.get('query', '')}"
|
||||
item = self.routing.get(key)
|
||||
if isinstance(item, Exception):
|
||||
raise item
|
||||
if item is not None:
|
||||
return item
|
||||
return FakeResp(200, {"groupDatas": []})
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def bundle():
|
||||
tpl = {"context": {"user": {"delegationContext": {"externalOwnerId": "O1"}}},
|
||||
"marker": "O1", "query": ""}
|
||||
return {
|
||||
"ownerId": "O1", "ownerDisplay": "Owner One",
|
||||
"url": "https://studio.youtube.com/youtubei/v1/yta_web/search_groups?alt=json",
|
||||
"headers": {
|
||||
"Authorization": "SAPISIDHASH x", "Cookie": "SID=1",
|
||||
"X-YouTube-Delegation-Context": "ctx",
|
||||
"Content-Length": "3", "Host": "studio.youtube.com",
|
||||
},
|
||||
"bodyTemplate": tpl,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# URL / 名单解析
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestParse:
|
||||
def test_parse_owner_id_path_first(self, lookup_groups):
|
||||
url = ("https://studio.youtube.com/owner/bqSUnNpU67xJ51TxH4PKpQ/analytics"
|
||||
"?o=WRONG&explore=...")
|
||||
assert lookup_groups.parse_owner_id(url) == "bqSUnNpU67xJ51TxH4PKpQ"
|
||||
|
||||
def test_parse_owner_id_query_fallback(self, lookup_groups):
|
||||
assert lookup_groups.parse_owner_id(
|
||||
"https://studio.youtube.com/analytics?o=AbCdEf123") == "AbCdEf123"
|
||||
assert lookup_groups.parse_owner_id("https://studio.youtube.com/") == ""
|
||||
|
||||
def test_parse_names_payload_variants(self, lookup_groups):
|
||||
assert lookup_groups.parse_names_payload(["A", "B", "A "]) == ["A", "B"]
|
||||
assert lookup_groups.parse_names_payload(
|
||||
[["GROUP_NAME"], ["A"], ["B"]]) == ["A", "B"]
|
||||
assert lookup_groups.parse_names_payload(
|
||||
[["群组名称", "备注"], ["A", "x"], ["B", "y"]]) == ["A", "B"]
|
||||
assert lookup_groups.parse_names_payload({"names": ["A", "B"]}) == ["A", "B"]
|
||||
assert lookup_groups.parse_names_payload({"foo": 1}) == []
|
||||
assert lookup_groups.parse_names_payload("not-a-list") == []
|
||||
|
||||
def test_load_names_txt_csv(self, lookup_groups, tmp_path):
|
||||
p_txt = tmp_path / "names.txt"
|
||||
p_txt.write_text("A\n B \n\nA\n", encoding="utf-8")
|
||||
assert lookup_groups.load_names(str(p_txt)) == ["A", "B"]
|
||||
|
||||
p_csv = tmp_path / "names.csv"
|
||||
p_csv.write_text("group_name,备注\nA,x\nB,\n", encoding="utf-8-sig")
|
||||
assert lookup_groups.load_names(str(p_csv)) == ["A", "B"]
|
||||
|
||||
p_single = tmp_path / "single.csv"
|
||||
p_single.write_text("A\nB\n", encoding="utf-8")
|
||||
assert lookup_groups.load_names(str(p_single)) == ["A", "B"]
|
||||
|
||||
def test_load_names_json(self, lookup_groups, tmp_path):
|
||||
p = tmp_path / "names.json"
|
||||
p.write_text(json.dumps(["A", "B"]), encoding="utf-8")
|
||||
assert lookup_groups.load_names(str(p)) == ["A", "B"]
|
||||
|
||||
def test_load_names_xlsx(self, lookup_groups, tmp_path):
|
||||
openpyxl = pytest.importorskip("openpyxl")
|
||||
wb = openpyxl.Workbook()
|
||||
ws = wb.active
|
||||
ws.append(["群组名称", "备注"])
|
||||
ws.append(["A", "x"])
|
||||
ws.append(["B", None])
|
||||
ws.append([None, "空行跳过"])
|
||||
p = tmp_path / "names.xlsx"
|
||||
wb.save(str(p))
|
||||
assert lookup_groups.load_names(str(p)) == ["A", "B"]
|
||||
|
||||
def test_load_names_unknown_column_raises(self, lookup_groups, tmp_path):
|
||||
p = tmp_path / "bad.csv"
|
||||
p.write_text("列一,列二\nA,B\n", encoding="utf-8")
|
||||
with pytest.raises(ValueError, match="无法识别名字列"):
|
||||
lookup_groups.load_names(str(p))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 请求构造 / 匹配 / 回放
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestSearch:
|
||||
def test_build_body_isolation(self, lookup_groups):
|
||||
tpl = {"context": {"user": {"delegationContext": {"externalOwnerId": "O1"}}},
|
||||
"query": ""}
|
||||
body = lookup_groups.build_body(tpl, "靓舟桃")
|
||||
assert body["query"] == "靓舟桃"
|
||||
assert tpl["query"] == "" # 模板不被污染
|
||||
assert body["context"] is not tpl["context"] # 深拷贝
|
||||
body["context"]["user"]["delegationContext"]["externalOwnerId"] = "CHANGED"
|
||||
assert tpl["context"]["user"]["delegationContext"]["externalOwnerId"] == "O1"
|
||||
|
||||
def test_pick_match(self, lookup_groups):
|
||||
gd = [{"displayName": "X 漫剧-1", "groupId": "G1"},
|
||||
{"displayName": "X 漫剧-2", "groupId": "G2"}]
|
||||
assert lookup_groups.pick_match(gd, "X 漫剧-2") == ("G2", [])
|
||||
gid, cands = lookup_groups.pick_match(gd, "X 漫剧")
|
||||
assert gid is None and cands == ["X 漫剧-1=G1", "X 漫剧-2=G2"]
|
||||
|
||||
def test_search_ok_nf_http_err(self, lookup_groups, bundle):
|
||||
sess = FakeSession({
|
||||
"O1::命中": FakeResp(200, {"groupDatas": [{"displayName": "命中", "groupId": "G9"}]}),
|
||||
"O1::疑似": FakeResp(200, {"groupDatas": [
|
||||
{"displayName": "X 漫剧-1", "groupId": "G1"}]}),
|
||||
"O1::限流": FakeResp(429),
|
||||
"O1::断网": OSError("refused"),
|
||||
"O1::坏响应": FakeResp(200, None),
|
||||
})
|
||||
assert lookup_groups.search(bundle, "命中", sess) == {"status": "OK", "groupId": "G9"}
|
||||
r = lookup_groups.search(bundle, "疑似", sess)
|
||||
assert r["status"] == "NF" and "X 漫剧-1=G1" in r["cands"]
|
||||
r = lookup_groups.search(bundle, "限流", sess)
|
||||
assert r["status"] == "HTTP" and r["detail"] == 429 and "限流" in r["hint"]
|
||||
assert lookup_groups.search(bundle, "断网", sess)["status"] == "ERR"
|
||||
assert lookup_groups.search(bundle, "坏响应", sess)["status"] == "ERR"
|
||||
|
||||
def test_search_strips_hop_by_hop_headers(self, lookup_groups, bundle):
|
||||
sess = FakeSession({})
|
||||
lookup_groups.search(bundle, "任意", sess)
|
||||
url, body, headers = sess.calls[0]
|
||||
assert "Content-Length" not in headers and "Host" not in headers
|
||||
assert headers["Authorization"] == "SAPISIDHASH x"
|
||||
assert headers["X-YouTube-Delegation-Context"] == "ctx"
|
||||
assert body["query"] == "任意"
|
||||
|
||||
def test_resolve_first_hit_and_miss(self, lookup_groups, bundle):
|
||||
b1 = dict(bundle, ownerId="O1", ownerDisplay="一号")
|
||||
tpl2 = dict(bundle["bodyTemplate"], marker="O2")
|
||||
b2 = dict(bundle, ownerId="O2", ownerDisplay="二号", bodyTemplate=tpl2)
|
||||
sess = FakeSession({
|
||||
"O2::跨主": FakeResp(200, {"groupDatas": [
|
||||
{"displayName": "跨主", "groupId": "G2"}]}),
|
||||
})
|
||||
row = lookup_groups.resolve("跨主", [b1, b2], sess)
|
||||
assert row == ["跨主", "O2", "二号", "G2", "精确命中；此前 一号(O1) 无结果"]
|
||||
row = lookup_groups.resolve("查无", [b1], sess)
|
||||
assert row[:4] == ["查无", "", "", ""]
|
||||
assert "未匹配任何 owner" in row[4] and "一号(O1)" in row[4]
|
||||
|
||||
def test_resolve_http_hint_in_note(self, lookup_groups, bundle):
|
||||
sess = FakeSession({"O1::401名字": FakeResp(401)})
|
||||
row = lookup_groups.resolve("401名字", [bundle], sess)
|
||||
assert "HTTP401" in row[4] and "鉴权失败" in row[4]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 套件校验 / 合并 / 输出
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestBundleAndOutput:
|
||||
def test_validate_bundle(self, lookup_groups):
|
||||
ok = {"headers": {"Authorization": "x", "Cookie": "y",
|
||||
"X-YouTube-Delegation-Context": "z"},
|
||||
"bodyTemplate": {"context": {"user": {"delegationContext": {
|
||||
"externalOwnerId": "O1"}}}}}
|
||||
assert lookup_groups.validate_bundle(ok) == []
|
||||
|
||||
bad = {"headers": {"Authorization": "x"},
|
||||
"bodyTemplate": {"context": {}}}
|
||||
problems = lookup_groups.validate_bundle(bad)
|
||||
assert any("Cookie" in p for p in problems)
|
||||
assert any("X-YouTube-Delegation-Context" in p for p in problems)
|
||||
assert any("delegationContext" in p for p in problems)
|
||||
assert lookup_groups.validate_bundle({}) != []
|
||||
|
||||
def test_merge_bundles(self, lookup_groups):
|
||||
merged = lookup_groups.merge_bundles(
|
||||
[{"ownerId": "O1", "v": 1}, {"bad": 1}],
|
||||
[{"ownerId": "O1", "v": 2}, {"ownerId": "O2", "v": 3}])
|
||||
assert [b["ownerId"] for b in merged] == ["O1", "O2"]
|
||||
assert merged[0]["v"] == 2 # captured 覆盖同名
|
||||
|
||||
def test_write_out_csv(self, lookup_groups, tmp_path):
|
||||
out = tmp_path / "result.csv"
|
||||
path = lookup_groups.write_out(
|
||||
str(out), lookup_groups.RESULT_HEADER,
|
||||
[["A", "O1", "一号", "G1", "精确命中"]])
|
||||
assert path == str(out)
|
||||
text = out.read_text(encoding="utf-8-sig")
|
||||
assert "group_name,ownerid,owner_display,groupid,备注" in text
|
||||
assert "精确命中" in text
|
||||
|
||||
def test_write_out_xlsx(self, lookup_groups, tmp_path):
|
||||
openpyxl = pytest.importorskip("openpyxl")
|
||||
out = tmp_path / "result.xlsx"
|
||||
path = lookup_groups.write_out(
|
||||
str(out), lookup_groups.RESULT_HEADER,
|
||||
[["A", "O1", "一号", "G1", "精确命中"]])
|
||||
assert path == str(out)
|
||||
wb = openpyxl.load_workbook(str(out))
|
||||
ws = wb["result"]
|
||||
assert [c.value for c in ws[1]] == lookup_groups.RESULT_HEADER
|
||||
assert ws["A2"].value == "A" and ws["D2"].value == "G1"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# selftest / CLI
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestCli:
|
||||
def test_selftest_function(self, lookup_groups, capsys):
|
||||
lookup_groups.selftest()
|
||||
out = capsys.readouterr().out
|
||||
assert "selftest OK" in out
|
||||
|
||||
def test_cli_selftest(self):
|
||||
r = run_script(LOOKUP_SCRIPT, ["--selftest"])
|
||||
assert r.returncode == 0, r.stderr
|
||||
assert "selftest OK" in r.stdout
|
||||
|
||||
def test_cli_noargs_runs_selftest_and_hint(self):
|
||||
r = run_script(LOOKUP_SCRIPT, [])
|
||||
assert r.returncode == 0, r.stderr
|
||||
assert "selftest OK" in r.stdout
|
||||
assert "方式 A" in r.stdout and "离线回放" in r.stdout
|
||||
|
||||
def test_cli_bad_bundles_file_exits(self, tmp_path):
|
||||
r = run_script(LOOKUP_SCRIPT, [
|
||||
"--bundles", str(tmp_path / "nope.json"),
|
||||
"--names", str(tmp_path / "names.json")])
|
||||
assert r.returncode != 0
|
||||
assert "读取 bundles 失败" in r.stdout + r.stderr
|
||||
Reference in New Issue
Block a user