# -*- coding: utf-8 -*- """创作者后台客户端的解析与签名。 **字段名尚未亲眼验证过**:Phase 0 抓响应时账号的数据权限还没生效,列表接口返回的是 空壳(`data.result` 里只有 `{success, code, message}`)。所以这些解析写成多别名匹配, 而这份测试就是它的规格 —— 等真实响应到手,先跑这里看哪些假设破了。 """ import pytest from api.creator import signing from api.creator.client import ( CreatorClient, as_float, as_int, as_seconds, find_note_list, normalize_note, trans_cookies, ) from api.creator.models import CreatorAccount from api.creator.service import _account_dict # --- cookie 解析 ----------------------------------------------------------- @pytest.mark.parametrize( "raw, expected", [ ("a1=abc; web_session=xyz", {"a1": "abc", "web_session": "xyz"}), ("a1=abc;web_session=xyz;", {"a1": "abc", "web_session": "xyz"}), ("a1=abc\nweb_session=xyz", {"a1": "abc", "web_session": "xyz"}), (" a1 = abc ; ", {"a1": "abc"}), ("", {}), (None, {}), # 值里可以有等号,不能被截断 ("a1=abc=def", {"a1": "abc=def"}), ], ) def test_trans_cookies(raw, expected): assert trans_cookies(raw) == expected def test_client_without_a1_cannot_sign(): """a1 参与签名,没有它连请求都发不出去 —— 要提前拦而不是发出去再猜。""" assert CreatorClient("web_session=abc").looks_authenticated is False assert CreatorClient("a1=abc").looks_authenticated is True # --- 数值解析 -------------------------------------------------------------- # # 后台返回的可能是数字,也可能是 "1.2万" / "12.3%" / "1分30秒" 这类展示值。 # 解析不出来一律 None —— 不是 0。0 是真实值,None 是"不知道"。 @pytest.mark.parametrize( "raw, expected", [ (123, 123), ("123", 123), ("1,234", 1234), ("1.2万", 12000), ("3万", 30000), ("1.5w", 15000), ("1亿", 100000000), (0, 0), ("0", 0), # 这些必须是 None 而不是 0 —— 把"没给"当成"是零"会让报表说谎 (None, None), ("", None), ("-", None), ("暂无", None), ("abc", None), (True, None), ], ) def test_as_int(raw, expected): assert as_int(raw) == expected @pytest.mark.parametrize( "raw, expected", [ (12.3, 12.3), ("12.3%", 12.3), ("12.3", 12.3), (None, None), ("-", None), ("暂无数据", None), ], ) def test_as_float(raw, expected): assert as_float(raw) == expected @pytest.mark.parametrize( "raw, expected", [ (45, 45.0), ("45", 45.0), ("1分30秒", 90.0), ("2分", 120.0), ("30秒", 30.0), ("01:30", 90.0), ("1:00:00", 3600.0), (None, None), ("-", None), ("abc", None), ], ) def test_as_seconds(raw, expected): assert as_seconds(raw) == expected # --- 字段归一化 ------------------------------------------------------------ def test_normalize_note_maps_aliases(): """不同来源的记录用不同字段名,别名表要能都接住。""" note = normalize_note( { "note_id": "abc123", "title": "标题", "publish_time": 1700000000000, "view_count": "1.2万", "like_count": 34, "collected_count": 5, "share_count": 2, "comment_count": 7, "cover_click_rate": "12.5%", "avg_watch_time": "1分30秒", } ) assert note["note_id"] == "abc123" assert note["title"] == "标题" assert note["views"] == 12000 assert note["likes"] == 34 assert note["favorites"] == 5 assert note["shares"] == 2 assert note["comments"] == 7 assert note["cover_ctr"] == 12.5 assert note["avg_watch_seconds"] == 90.0 def test_normalize_note_leaves_missing_fields_as_none(): note = normalize_note({"note_id": "abc123"}) assert note["note_id"] == "abc123" assert note["views"] is None assert note["likes"] is None def test_find_note_list_digs_the_array_out_of_a_nested_payload(): """接口的确切结构没见过,所以按"像是一批笔记记录"来找,不写死路径。""" payload = { "code": 0, "data": { "result": { "success": True, "notes": [ {"note_id": "n1", "views": 10, "likes": 1}, {"note_id": "n2", "views": 20, "likes": 2}, ], } }, } found = find_note_list(payload) assert [item["note_id"] for item in found] == ["n1", "n2"] def test_find_note_list_returns_empty_for_the_permission_gated_envelope(): """权限未生效时接口返回的就是这个 —— 必须安静地给出空列表,不是报错。""" payload = { "code": 0, "success": True, "msg": "成功", "data": {"result": {"success": True, "code": 0, "message": "success"}}, } assert find_note_list(payload) == [] # --- 签名 ------------------------------------------------------------------ def test_signed_api_carries_the_url_prefix(): """待签字符串必须带 `url=`。少了它网关返回 406,而 406 的响应体看不出错在哪。""" assert signing.signed_api("/api/galaxy/user/info") == "url=/api/galaxy/user/info" assert ( signing.signed_api("/api/x", "a=1&b=2") == "url=/api/x?a=1&b=2" ) def test_sign_returns_xs_and_xt(): headers = signing.sign_xyw("url=/api/galaxy/user/info", "some-a1") assert set(headers) == {"x-s", "x-t"} assert headers["x-s"].startswith("XYW_") assert headers["x-t"].isdigit() def test_signature_is_stable_for_a_fixed_timestamp(): """同一输入同一时间戳必须得到同一签名 —— 否则说明有隐藏的随机源。""" first = signing.sign_xyw("url=/api/x", "a1", timestamp_ms=1700000000000) second = signing.sign_xyw("url=/api/x", "a1", timestamp_ms=1700000000000) assert first == second def test_signature_changes_with_the_signed_string(): """签名必须真的绑定待签内容,否则改参数不会被发现 —— 那这个签名就没意义了。""" base = signing.sign_xyw("url=/api/x?a=1", "a1", timestamp_ms=1700000000000) other = signing.sign_xyw("url=/api/x?a=2", "a1", timestamp_ms=1700000000000) assert base["x-s"] != other["x-s"] # --- 凭证不外泄 ------------------------------------------------------------ def test_account_dict_never_carries_the_cookie(): """cookie 等于登录态。对外结构里只该有 `has_cookie`。""" account = CreatorAccount( id=1, nickname="测试", user_id="u1", cookie="a1=SECRET; web_session=SECRET", created_at=0, updated_at=0, ) payload = _account_dict(account) assert payload["has_cookie"] is True assert "cookie" not in payload assert "SECRET" not in str(payload)