ksjsb/tests/test_dfp_client.py
2026-07-30 20:25:56 +08:00

97 lines
3.2 KiB
Python

import unittest
from core.device_profile import DeviceProfileGenerator
from core.dfp_client import (
BootstrapIdentity,
apply_bootstrap_identity,
extract_bootstrap_identity,
post_request,
)
from core.dfp_forms import DfpRequestSpec
class DfpClientTests(unittest.TestCase):
def test_extract_bootstrap_identity_from_fetch_and_report(self):
identity = extract_bootstrap_identity(
{
"result": 1,
"cloud_did": "ANDROID_f05497e9cef09a7f",
"did_tag": 2,
"egid": "",
},
{
"result": 1,
"egid": "DFPD5DAB4376320DEA4F0DDFEA76E0B216F894BF2247BD8F6DA8B77AD981AF68",
},
)
self.assertEqual(identity.did, "ANDROID_f05497e9cef09a7f")
self.assertEqual(identity.cdid_tag, 2)
self.assertEqual(
identity.egid,
"DFPD5DAB4376320DEA4F0DDFEA76E0B216F894BF2247BD8F6DA8B77AD981AF68",
)
def test_apply_bootstrap_identity_updates_profile(self):
profile = DeviceProfileGenerator(seed=1234).new_profile()
identity = extract_bootstrap_identity(
{"result": 1, "cloud_did": "ANDROID_f05497e9cef09a7f", "did_tag": "2"},
{"result": 1, "egid": "DFPD5DAB4376320DEA4F0DDFEA76E0B216F894BF2247BD8F6DA8B77AD981AF68"},
)
apply_bootstrap_identity(profile, identity)
self.assertEqual(profile.did, "ANDROID_f05497e9cef09a7f")
self.assertEqual(profile.cdid_tag, 2)
self.assertEqual(
profile.egid,
"DFPD5DAB4376320DEA4F0DDFEA76E0B216F894BF2247BD8F6DA8B77AD981AF68",
)
def test_invalid_egid_is_not_applied(self):
identity = extract_bootstrap_identity(
{"result": 1, "cloud_did": "ANDROID_f05497e9cef09a7f", "did_tag": 2},
{"result": 1, "egid": ""},
)
self.assertEqual(identity.egid, "")
def test_apply_bootstrap_identity_can_refresh_egid_without_new_did(self):
profile = DeviceProfileGenerator(seed=4321).new_profile()
old_did = profile.did
identity = BootstrapIdentity(
egid="DFPD5DAB4376320DEA4F0DDFEA76E0B216F894BF2247BD8F6DA8B77AD981AF68"
)
apply_bootstrap_identity(profile, identity)
self.assertEqual(profile.did, old_did)
self.assertEqual(
profile.egid,
"DFPD5DAB4376320DEA4F0DDFEA76E0B216F894BF2247BD8F6DA8B77AD981AF68",
)
def test_post_request_converts_transport_error_to_response(self):
spec = DfpRequestSpec(
method="POST",
url="https://gdfpsec.ksapisrv.com/rest/infra/gdfp/report/kuaishou/android",
headers={"Content-Type": "application/x-www-form-urlencoded"},
form_order=[],
form={},
body="",
)
def failing_post(url, data, headers, timeout):
raise TimeoutError("network timeout")
response = post_request(spec, timeout=3, post_func=failing_post)
self.assertEqual(response.status_code, 0)
self.assertFalse(response.ok)
self.assertEqual(response.data["error_type"], "TimeoutError")
self.assertIn("network timeout", response.text)
if __name__ == "__main__":
unittest.main()