54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
import threading
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from rpa_engine.credential import validate_im_session
|
|
|
|
|
|
class CredentialResponsivenessTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_uid_lookup_does_not_block_event_loop(self):
|
|
event_loop_thread_id = threading.get_ident()
|
|
lookup_thread_ids = []
|
|
|
|
def get_uid():
|
|
lookup_thread_ids.append(threading.get_ident())
|
|
return 123456
|
|
|
|
session = SimpleNamespace(
|
|
my_uid=0,
|
|
can_direct_im=lambda: True,
|
|
)
|
|
auth = SimpleNamespace(
|
|
get_uid=get_uid,
|
|
is_sign_ready=lambda: True,
|
|
)
|
|
|
|
with (
|
|
patch(
|
|
"rpa_engine.credential.ensure_frontier_ws",
|
|
return_value=None,
|
|
),
|
|
patch(
|
|
"rpa_engine.credential.DouyinAuth.from_im_session",
|
|
return_value=auth,
|
|
),
|
|
):
|
|
result = await validate_im_session(
|
|
session,
|
|
_bypass_global_limit=True,
|
|
)
|
|
|
|
self.assertTrue(result[0])
|
|
self.assertEqual(len(lookup_thread_ids), 1)
|
|
self.assertNotEqual(
|
|
lookup_thread_ids[0],
|
|
event_loop_thread_id,
|
|
"the synchronous UID lookup ran on the event-loop thread",
|
|
)
|
|
self.assertEqual(session.my_uid, 123456)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|