213 lines
12 KiB
Python
213 lines
12 KiB
Python
# SPDX-License-Identifier: MPL-2.0
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import socket
|
|
import struct
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
DEVICE=Path(__file__).parents[1]/'devices/beatles'
|
|
sys.path.insert(0,str(DEVICE))
|
|
import radio_control as control
|
|
import sms_pdu as sms
|
|
from radio_status import Parcel
|
|
|
|
HELLO='00040A91214365870900006201812100008005E8329BFD06'
|
|
|
|
class PduTests(unittest.TestCase):
|
|
def test_standard_gsm_vector_and_unicode(self):
|
|
self.assertEqual(sms.pack7(sms.septets('hello')).hex(),'e8329bfd06')
|
|
self.assertEqual(sms.submit('+1234567890','hello'),'01000A912143658709000005E8329BFD06')
|
|
row=sms.deliver(HELLO)
|
|
self.assertEqual(row['sender'],'+1234567890');self.assertEqual(row['body'],'hello')
|
|
self.assertEqual(row['timestamp'],1792317600000)
|
|
self.assertEqual(sms.submit('123','漢').lower(),'0100038121f30008026f22')
|
|
def test_extensions_and_length_limits(self):
|
|
text='é € {} [] ^ ~ \\';values=sms.septets(text)
|
|
self.assertEqual(sms.unpack7(sms.pack7(values),len(values)),text)
|
|
sms.submit('123','A'*160)
|
|
for text in ('A'*161,'€'*81,'漢'*71):
|
|
with self.assertRaisesRegex(ValueError,'SMS_TOO_LONG'):sms.submit('123',text)
|
|
for value in ('112;reboot','*#06#','',None,'+123\n'):
|
|
with self.assertRaises(ValueError):sms.submit(value,'hello')
|
|
def test_truncated_and_unsupported_pdus(self):
|
|
for end in range(0,len(HELLO),2):
|
|
with self.assertRaises((ValueError,UnicodeError)):sms.deliver(HELLO[:end])
|
|
for pdu in ('00ff','xx','00040A9121436587097F006201812100008005E8329BFD06'):
|
|
with self.assertRaises(ValueError):sms.deliver(pdu)
|
|
def test_unicode_and_concatenation(self):
|
|
prefix='00440A912143658709000862018121000080'
|
|
row=sms.deliver(prefix+'080500030702010041')
|
|
self.assertEqual(row['body'],'A');self.assertEqual(row['concat'],(7,2,1))
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path=Path(directory)/'inbox.json';inbox=control.Inbox(path)
|
|
inbox.receive(prefix+'080500030702020042');self.assertEqual(inbox.value['inbox'],[])
|
|
inbox.receive(prefix+'080500030702010041')
|
|
self.assertEqual(inbox.value['inbox'][0]['body'],'AB')
|
|
self.assertEqual(control.Inbox(path).value['inbox'][0]['body'],'AB')
|
|
def test_durable_inbox_dedup_and_failed_save(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path=Path(directory)/'inbox.json';inbox=control.Inbox(path)
|
|
with patch.object(inbox,'save',side_effect=OSError('disk full')):
|
|
with self.assertRaises(OSError):inbox.receive(HELLO)
|
|
self.assertEqual(inbox.value['seen'],{})
|
|
inbox.receive(HELLO);inbox.receive(HELLO)
|
|
self.assertEqual(len(inbox.value['inbox']),1)
|
|
self.assertEqual(path.stat().st_mode&0o777,0o600)
|
|
receipt=inbox.value['inbox'][0]['receipt'];inbox.acknowledge([receipt])
|
|
inbox.receive(HELLO);self.assertEqual(inbox.value['inbox'],[])
|
|
|
|
class ControlTests(unittest.TestCase):
|
|
def test_unanswered_request_is_not_sent_again(self):
|
|
radio=control.ControlRadio('/unused',0,None)
|
|
radio.pending={10:(9,0)}
|
|
broker=control.Broker([radio],None)
|
|
with patch.object(radio,'send') as send:
|
|
with self.assertRaisesRegex(RuntimeError,'RADIO_BUSY'):broker.exchange(9)
|
|
send.assert_not_called()
|
|
|
|
def test_call_list_timeout_is_short_and_request_stays_pending(self):
|
|
radio=control.ControlRadio('/unused',0,None);radio.sock=unittest.mock.Mock()
|
|
broker=control.Broker([radio],None)
|
|
with patch.object(control.time,'monotonic',side_effect=[0,0,1,2,3]),patch.object(broker,'step'):
|
|
with self.assertRaisesRegex(RuntimeError,'RADIO_TIMEOUT'):broker.exchange(9)
|
|
self.assertIsNone(radio.waiting)
|
|
self.assertEqual(len(radio.pending),1)
|
|
|
|
def test_call_layout_and_caller_privacy(self):
|
|
call=control.integer(1,4,2,145,0,1,0,1,0)+control.string('123456789')+control.integer(0)+control.string(None)+control.integer(0,0)
|
|
self.assertEqual(control.current_calls(Parcel(call)),[{'index':2,'state':'incoming','number':'+123456789','direction':'incoming'}])
|
|
hidden=control.integer(1,4,2,145,0,1,0,1,0)+control.string('123456789')+control.integer(1)+control.string(None)+control.integer(0,0)
|
|
self.assertEqual(control.current_calls(Parcel(hidden))[0]['number'],'')
|
|
self.assertEqual(control.current_calls(Parcel(b'')),[])
|
|
def test_late_modem_reply_does_not_complete_another_request(self):
|
|
radio=control.ControlRadio('/unused',0,None);radio.waiting=(9,20)
|
|
radio.response(9,0,Parcel(control.integer(0)),19);self.assertIsNone(radio.result)
|
|
radio.response(9,0,Parcel(control.integer(0)),20);self.assertEqual(radio.result,(True,[]))
|
|
def test_malformed_call_list_does_not_drop_radio_transport(self):
|
|
radio=control.ControlRadio('/unused',0,None);radio.waiting=(9,20)
|
|
radio.sock=unittest.mock.Mock()
|
|
radio.response(9,0,Parcel(control.integer(1)),20)
|
|
self.assertEqual(radio.result,(False,'RADIO_INVALID_CALL_LIST'))
|
|
radio.sock.close.assert_not_called()
|
|
|
|
def test_unknown_operations_do_not_reach_the_modem(self):
|
|
broker=control.Broker([],None)
|
|
with patch.object(broker,'exchange') as exchange:
|
|
for value in ({'version':1,'operation':'shell'},{'operation':'sms'},[],None):
|
|
with self.assertRaises(ValueError):broker.request(value)
|
|
exchange.assert_not_called()
|
|
def test_calls_answer_hangup_and_sms_parcels(self):
|
|
radio=control.ControlRadio('/unused',0,None);radio.state={'voice':{'connected':True}}
|
|
broker=control.Broker([radio],None)
|
|
row={'index':2,'state':'incoming','number':'123','direction':'incoming'}
|
|
with patch.object(broker,'exchange',return_value=[row]) as exchange:
|
|
calls=broker.request({'version':1,'operation':'calls'});path=calls[0]['path']
|
|
broker.request({'version':1,'operation':'hangup','path':path})
|
|
self.assertEqual(exchange.call_args.args,(12,control.integer(1,2)))
|
|
broker.request({'version':1,'operation':'answer','path':path})
|
|
self.assertEqual(exchange.call_args.args,(40,))
|
|
with self.assertRaisesRegex(ValueError,'INVALID_CALL'):
|
|
broker.request({'version':1,'operation':'hangup','path':'/stale/call/2'})
|
|
with patch.object(broker,'exchange',return_value={'state':'sent'}) as exchange:
|
|
self.assertEqual(broker.request({'version':1,'operation':'sms','number':'123','body':'hello'}),{'state':'sent'})
|
|
self.assertEqual(exchange.call_args.args,(25,control.integer(2)+control.string(None)+control.string(sms.submit('123','hello'))))
|
|
def test_second_call_is_rejected_before_dial(self):
|
|
radio=control.ControlRadio('/unused',0,None);radio.state={'voice':{'connected':True}}
|
|
broker=control.Broker([radio],None)
|
|
with patch.object(broker,'calls',return_value=[{'path':'/live/call/1'}]),patch.object(broker,'exchange') as exchange:
|
|
with self.assertRaisesRegex(RuntimeError,'RADIO_BUSY'):broker.request({'version':1,'operation':'dial','number':'123'})
|
|
exchange.assert_not_called()
|
|
|
|
def test_sms_ack_only_after_storage(self):
|
|
inbox=unittest.mock.Mock();radio=control.ControlRadio('/unused',0,inbox)
|
|
with patch.object(radio,'send') as send:
|
|
radio.unsolicited(1003,Parcel(control.string(HELLO)))
|
|
inbox.receive.assert_called_once_with(HELLO);send.assert_called_with(37,control.integer(2,1,0))
|
|
inbox.receive.side_effect=OSError('full')
|
|
radio.unsolicited(1003,Parcel(control.string(HELLO)))
|
|
send.assert_called_with(37,control.integer(2,0,255))
|
|
|
|
|
|
class RadioPowerTests(unittest.TestCase):
|
|
def broker(self):
|
|
radios=[control.ControlRadio('/unused',i,None) for i in range(2)]
|
|
for radio in radios:
|
|
radio.state={'radioState':'enabled'};radio.sock=unittest.mock.Mock()
|
|
return control.Broker(radios,None)
|
|
def test_slot_payload_readback_and_restore(self):
|
|
broker=self.broker()
|
|
def exchange(request,payload,slot=0):
|
|
self.assertEqual(request,23)
|
|
count,enabled=struct.unpack('<ii',payload);self.assertEqual(count,1)
|
|
broker.radios[slot].state['radioState']='enabled' if enabled else 'disabled'
|
|
with patch.object(broker,'exchange',side_effect=exchange) as send,patch.object(broker,'calls',return_value=[]):
|
|
self.assertEqual(broker.request({'version':1,'operation':'radioPower','slot':1,'enabled':False}),
|
|
{'slot':1,'radioState':'disabled'})
|
|
self.assertEqual(broker.radios[0].state['radioState'],'enabled')
|
|
send.assert_called_once_with(23,control.integer(1,0),slot=1)
|
|
broker.restore_power()
|
|
self.assertEqual(broker.radios[1].state['radioState'],'enabled')
|
|
def test_idempotent_power_does_not_send(self):
|
|
broker=self.broker()
|
|
with patch.object(broker,'exchange') as send:
|
|
broker.radio_power(0,True);send.assert_not_called()
|
|
def test_calls_invalid_values_and_unknown_state_block_power(self):
|
|
broker=self.broker()
|
|
with patch.object(broker,'exchange') as send,patch.object(broker,'calls',return_value=[{'state':'active'}]):
|
|
with self.assertRaisesRegex(RuntimeError,'RADIO_BUSY'):broker.radio_power(0,False)
|
|
for slot,enabled in [(False,False),(2,False),(0,0),(0,'false')]:
|
|
with self.assertRaisesRegex(ValueError,'INVALID_REQUEST'):broker.radio_power(slot,enabled)
|
|
broker.radios[0].state['radioState']='unknown'
|
|
with self.assertRaisesRegex(RuntimeError,'RADIO_NOT_READY'):broker.radio_power(0,False)
|
|
send.assert_not_called()
|
|
def test_acknowledgement_without_state_change_fails(self):
|
|
broker=self.broker()
|
|
with patch.object(broker,'exchange'),patch.object(broker,'calls',return_value=[]),patch.object(control.time,'monotonic',side_effect=[0,9]):
|
|
with self.assertRaisesRegex(RuntimeError,'RADIO_READBACK_FAILED'):broker.radio_power(0,False)
|
|
self.assertEqual(broker.original_power,{0:True})
|
|
|
|
|
|
class AudioTests(unittest.TestCase):
|
|
def test_call_mode_routing_and_restore(self):
|
|
from call_audio import CallAudio
|
|
audio=CallAudio();state={'mode':0,'route':0,'muted':0};commands=[]
|
|
def transaction(service,method,*values):
|
|
commands.append((service,method,values))
|
|
if (service,method)==('policy',6):return state['route']
|
|
if (service,method)==('flinger',18):return state['muted']
|
|
if (service,method)==('policy',3):state['mode']=values[0]
|
|
elif (service,method)==('policy',5):state['route']=values[1]
|
|
elif (service,method)==('flinger',17):state['muted']=values[0]
|
|
else:raise AssertionError('Unexpected audio transaction')
|
|
return 0
|
|
with patch.object(audio,'transact',side_effect=transaction):
|
|
audio.update([{'state':'incoming'}]);self.assertEqual(state['mode'],1)
|
|
audio.update([{'state':'active'}]);self.assertEqual(state['mode'],2)
|
|
self.assertTrue(audio.request({'operation':'set','muted':True,'speaker':True})['speaker'])
|
|
audio.restore();self.assertEqual(state,{'mode':0,'route':0,'muted':0})
|
|
def test_audio_failure_does_not_throw_from_call_tracking(self):
|
|
from call_audio import CallAudio
|
|
audio=CallAudio()
|
|
with patch.object(audio,'transact',side_effect=OSError('service missing')):
|
|
audio.update([{'state':'active'}]);self.assertFalse(audio.available)
|
|
with self.assertRaisesRegex(RuntimeError,'CALL_AUDIO_UNAVAILABLE'):audio.request({'operation':'status'})
|
|
|
|
class RilRequestSafetyTests(unittest.TestCase):
|
|
def test_call_list_uses_android_request_nine(self):
|
|
broker=control.Broker([],None)
|
|
with patch.object(broker,'exchange',return_value=[]) as exchange:
|
|
self.assertEqual(broker.calls(),[])
|
|
exchange.assert_called_once_with(9)
|
|
def test_pin_puk_and_identity_requests_never_reach_socket(self):
|
|
radio=control.ControlRadio('/unused',0,None)
|
|
radio.sock=unittest.mock.Mock()
|
|
for request in (2,3,4,5,6,7,8,11,28,38,39,43):
|
|
with self.assertRaisesRegex(ValueError,'RADIO_INVALID_REQUEST'):radio.send(request)
|
|
radio.sock.sendall.assert_not_called()
|
|
self.assertEqual(control.CONTROL_REQUESTS,{9,10,12,23,25,37,40})
|
|
|
|
if __name__=='__main__':unittest.main()
|