live-hub-py/tests/test_huya_dfp_register.py

"""虎牙新设备注册链 (core/huya/dfp_register) 测试。

覆盖:
* 随机 dfpReport body/cw 结构 (4226B / 4146B / 固定10B尾);
* dfpReport 响应解析 (t1/t2/t5) 及缺字段失败;
* register_device 全流程: 三步 POST 顺序与 content-type、fingerprint 注入、
  注册结果三元组返回;
* 失败语义: 模板缺失/HTTP 异常 -> DfpRegistrationError (不静默、不发登录帧)。
"""
import base64
import json
import unittest
from pathlib import Path
from unittest.mock import patch

import requests

from core.huya.dfp_register import (
    CHAIN_FILE,
    CW_LEN,
    DfpRegistrationError,
    _build_random_dfp_body,
    _parse_response,
    register_device,
)

TAIL = bytes.fromhex("3600400c0b8c980ca80c")


def _fake_resp() -> bytes:
    """合成 dfpReport 响应: t1(32hex) + t2(180B base64 action) + t5(40hex)。"""
    return (b"\x16\x20" + b"a" * 32
            + b"\x26\xb4" + b"A" * 180
            + b"\x56\x28" + b"b" * 40)


_GOLDEN_FP = b"02df398797432eadefcc12767119ad5e80999389"


def _dummy_chain() -> dict:
    # selectOperator 请求体包含金样本 fingerprint, 供替换注入测试
    return {
        "getDfpConfig": (b"cfg-req", b"cfg-resp"),
        "selectOperator": (b"sel-hdr" + _GOLDEN_FP + b"sel-tail", b"sel-resp"),
        "dfpReport": (b"dfp-req", b"dfp-resp"),
    }


class TestRandomDfpBody(unittest.TestCase):

    def test_body_shape(self):
        body = _build_random_dfp_body()
        self.assertEqual(len(body), 4226)
        cw = body[-CW_LEN:]
        self.assertEqual(len(cw), CW_LEN)  # 4146
        self.assertTrue(cw.endswith(TAIL), "cw 尾部应为固定 10B")

    def test_body_is_randomized(self):
        b1 = _build_random_dfp_body()
        b2 = _build_random_dfp_body()
        self.assertNotEqual(b1, b2, "每次注册应生成不同的随机 cw")


class TestParseResponse(unittest.TestCase):

    def test_parse_ok(self):
        t1, t2, t5 = _parse_response(_fake_resp())
        self.assertEqual(t1, "a" * 32)
        self.assertEqual(t2, "A" * 180)
        self.assertEqual(t5, "b" * 40)

    def test_parse_golden_evidence(self):
        """真实抓包模板响应必须可解析 (证据: evidence/dfp_chain_golden.json)。"""
        if not CHAIN_FILE.exists():
            self.skipTest("缺失 golden 注册链模板")
        data = json.loads(CHAIN_FILE.read_text(encoding="utf-8"))
        resp = base64.b64decode(data["dfpReport"]["resp_b64"])
        t1, t2, t5 = _parse_response(resp)
        self.assertEqual(len(t1), 32)
        self.assertEqual(len(t2), 180)
        self.assertEqual(len(t5), 40)

    def test_parse_missing_fields_raises(self):
        with self.assertRaises(DfpRegistrationError):
            _parse_response(b"\x16\x20" + b"a" * 32)  # 缺 t2/t5
        with self.assertRaises(DfpRegistrationError):
            _parse_response(b"\x26\xb4" + b"A" * 180)  # 缺 t1/t5
        with self.assertRaises(DfpRegistrationError):
            _parse_response(b"\x56\x28" + b"b" * 40)  # 缺 t1/t2


class TestRegisterDevice(unittest.TestCase):

    def test_flow_three_steps_and_returns_triple(self):
        calls = []

        def spy_post(body, content_type="application/octet-stream",
                     timeout=20, proxies=None):
            calls.append(content_type)
            return _fake_resp()

        with patch("core.huya.dfp_register._load_chain", return_value=_dummy_chain()), \
             patch("core.huya.dfp_register._post", side_effect=spy_post):
            t1, t2, t5 = register_device(fingerprint=None)
        self.assertEqual((t1, t2, t5), ("a" * 32, "A" * 180, "b" * 40))
        # getDfpConfig -> selectOperator -> dfpReport 的 content-type 序列
        self.assertEqual(
            calls,
            ["application/octet-stream", "application/x-wup", "application/octet-stream"],
        )

    def test_select_operator_injects_fingerprint(self):
        captured = []

        def spy_post(body, content_type="application/octet-stream",
                     timeout=20, proxies=None):
            captured.append((body, content_type))
            return _fake_resp()

        new_fp = b"0" * 20 + b"f" * 20
        with patch("core.huya.dfp_register._load_chain", return_value=_dummy_chain()), \
             patch("core.huya.dfp_register._post", side_effect=spy_post):
            register_device(fingerprint=new_fp.decode("ascii"))
        sel_body, sel_ct = captured[1]
        self.assertEqual(sel_ct, "application/x-wup")
        self.assertIn(new_fp, sel_body, "selectOperator 应注入当前账号画像 fingerprint")
        self.assertNotIn(_GOLDEN_FP, sel_body)
        # dfpReport 请求体为随机 cw 结构 (零设备生成)
        dfp_body, dfp_ct = captured[2]
        self.assertEqual(len(dfp_body), 4226)
        self.assertEqual(dfp_ct, "application/octet-stream")

    def test_chain_template_missing_raises(self):
        """模板缺失 -> 注册链直接报错, 不应发起任何网络请求。"""
        with patch("core.huya.dfp_register.CHAIN_FILE",
                   Path("/tmp/definitely_missing_chain.json")), \
             patch("core.huya.dfp_register._post") as m_post:
            with self.assertRaises(DfpRegistrationError):
                register_device()
        m_post.assert_not_called()

    def test_http_failure_raises(self):
        """HTTP 层异常由 _post 转换为 DfpRegistrationError 后向上传播 (不静默)。"""
        with patch("core.huya.dfp_register._load_chain", return_value=_dummy_chain()), \
             patch("core.huya.dfp_register.requests.post",
                   side_effect=requests.RequestException("connection reset")):
            with self.assertRaises(DfpRegistrationError):
                register_device()


if __name__ == "__main__":
    unittest.main()