# -*- 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