43 lines
2.9 KiB
Python
43 lines
2.9 KiB
Python
import json
|
|
import unittest
|
|
from unittest.mock import Mock
|
|
from browser_gateway.platform.notice_signal import NoticeSignal, notice_ids, detail_script
|
|
from browser_gateway.platform.douyin import DouyinError
|
|
|
|
class NoticeSignalTest(unittest.TestCase):
|
|
def test_ids_remain_exact_and_unknown_groups_are_ignored(self):
|
|
event={'service':20313,'payload':json.dumps({'notices':[{'effect_groups':[700],'notice_id_str':'9007199254740993123'},{'effect_groups':[3],'notice_id_str':'7'}]})}
|
|
self.assertEqual(notice_ids(event),['9007199254740993123'])
|
|
self.assertEqual(notice_ids({'service':20003,'payload':'{"notice_type":45,"notice_id_str":"8"}'}),['8'])
|
|
self.assertEqual(notice_ids({'service':20003,'payload':'{"notice_type":0}'}),[])
|
|
def test_bad_payload_or_id_fails(self):
|
|
for payload in ('[]','{"notices":"bad"}','{"notices":[{"effect_groups":[700],"notice_id_str":2}]}'):
|
|
with self.assertRaises(DouyinError):notice_ids({'service':20313,'payload':payload})
|
|
def bridge(self, responses):
|
|
browser=Mock();connection=browser._connect.return_value
|
|
connection.evaluate.side_effect=[json.dumps(v) for v in responses]
|
|
return NoticeSignal(browser,'safe','123'),connection
|
|
def test_empty_heartbeat_does_not_fetch_notification_api(self):
|
|
bridge,conn=self.bridge([{'connected':True},[],None]);self.assertEqual(bridge.receive(),[])
|
|
bridge.close();self.assertEqual(conn.evaluate.call_count,3);conn.close.assert_called_once()
|
|
def test_details_are_decoded_in_python_with_no_numeric_rounding(self):
|
|
nid=9007199254740993123
|
|
push={'kind':'push','service':20313,'payload':json.dumps({'notices':[{'effect_groups':[700],'notice_id_str':str(nid)}]})}
|
|
bridge,conn=self.bridge([{'connected':True},[push],{'status':200,'body':json.dumps({'status_code':0,'notice_list_v2':[{'nid':nid}]})},None])
|
|
self.assertEqual(bridge.receive()[0]['nid'],nid);bridge.close()
|
|
self.assertIn('is_mark_read:0',detail_script([str(nid)]));self.assertIn('eject(observer)',detail_script([str(nid)]))
|
|
def test_closed_signal_and_incomplete_details_fail_explicitly(self):
|
|
bridge,conn=self.bridge([{'connected':True},[{'kind':'close'}]])
|
|
with self.assertRaisesRegex(DouyinError,'disconnected'):bridge.receive()
|
|
conn.close()
|
|
push={'kind':'push','service':20313,'payload':'{"notices":[{"effect_groups":[700],"notice_id_str":"9"}]}'}
|
|
bridge,conn=self.bridge([{'connected':True},[push],{'status':200,'body':'{"status_code":0,"notice_list_v2":[]}'}])
|
|
with self.assertRaisesRegex(DouyinError,'incomplete'):bridge.receive()
|
|
conn.close()
|
|
def test_install_failure_closes_connection(self):
|
|
browser=Mock();browser._connect.return_value.evaluate.return_value=42
|
|
with self.assertRaisesRegex(DouyinError,'raw JSON'):NoticeSignal(browser,'safe','123')
|
|
browser._connect.return_value.close.assert_called_once()
|
|
|
|
if __name__=='__main__':unittest.main()
|