# -*- coding: utf-8 -*-
"""main.py 广告/任务链路进程级 10418 state 的行为锁定。

真机一次冷启动 = 一个 Kwsg10418State（时间驱动 session_seed + counter
随请求单调递增）。这里锁定 main.py 的 ``new_api_sig3_state`` /
``_resolve_api_sig3_state`` 复刻该行为，防止未来回退成"每次 fetch 新建
state、counter 恒定 + 旧 HAR seed 重放"的可检测指纹。
"""

import os
import unittest
from unittest.mock import patch

from core.sig3 import Kwsg10418State, kwsg_10418_digest24_unmix
from main import API_SIG3_SEED, KsNebulaClient, new_api_sig3_state


class NewApiSig3StateTests(unittest.TestCase):
    def test_signing_increments_counter_with_same_seed(self):
        # 连签两次：counter 0x60→0x61，session_seed 不变（同一次冷启动）
        with patch.dict(os.environ, {"KS_API_SIG3_FRESH_STATE": "1"}, clear=False):
            state = new_api_sig3_state()
        self.assertIsNotNone(state)
        d1 = kwsg_10418_digest24_unmix(state.sig3_hex(b"req1"))
        d2 = kwsg_10418_digest24_unmix(state.sig3_hex(b"req2"))
        self.assertEqual(d1["session_seed"], d2["session_seed"])
        self.assertEqual([d1["counter"], d2["counter"]], [0x60, 0x61])

    def test_counter_starts_at_sms_login_value(self):
        with patch.dict(os.environ, {"KS_API_SIG3_FRESH_STATE": "1"}, clear=False):
            state = new_api_sig3_state()
        self.assertEqual(state.counter, 0x5F)  # 同 tools/sms_login_cli.py:_new_sig3_state

    def test_disabled_returns_none(self):
        # KS_API_SIG3_FRESH_STATE=0 → None，由调用方回退旧 from_digest(HAR_SEED)
        with patch.dict(os.environ, {"KS_API_SIG3_FRESH_STATE": "0"}, clear=False):
            self.assertIsNone(new_api_sig3_state())


class ResolveApiSig3StateTests(unittest.TestCase):
    def _bare_client(self):
        # 裸实例避免走重量级 __init__；_resolve 只读写 self.api_sig3_state
        client = object.__new__(KsNebulaClient)
        client.api_sig3_state = None
        return client

    def test_shares_process_state_across_calls(self):
        client = self._bare_client()
        s1 = KsNebulaClient._resolve_api_sig3_state(client, None, "normal_fetch")
        s2 = KsNebulaClient._resolve_api_sig3_state(client, None, "normal_fetch")
        self.assertIs(s1, s2)  # 同一实例 → counter 持续递增而非恒定重放
        d1 = kwsg_10418_digest24_unmix(s1.sig3_hex(b"req1"))
        d2 = kwsg_10418_digest24_unmix(s1.sig3_hex(b"req2"))
        self.assertEqual(d1["session_seed"], d2["session_seed"])
        self.assertEqual(d2["counter"], d1["counter"] + 1)

    def test_explicit_state_wins_and_does_not_pollute_shared(self):
        client = self._bare_client()
        explicit = Kwsg10418State(session_seed=0x12345678, counter=0x10)
        resolved = KsNebulaClient._resolve_api_sig3_state(client, explicit, "normal_fetch")
        self.assertIs(resolved, explicit)
        self.assertIsNone(client.api_sig3_state)  # 显式优先时不写回共享 state

    def test_falls_back_to_har_seed_when_disabled(self):
        client = self._bare_client()
        with patch.dict(os.environ, {"KS_API_SIG3_FRESH_STATE": "0"}, clear=False):
            resolved = KsNebulaClient._resolve_api_sig3_state(client, None, "normal_fetch")
        har = Kwsg10418State.from_digest(API_SIG3_SEED["normal_fetch"])
        self.assertEqual(resolved.session_seed, har.session_seed)
        self.assertEqual(resolved.counter, har.counter)


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