jsahnandClaude Opus 5.5
1cd00ae21b
inputs() still used GetManagedObjects on BlueZ to look up names and silently dropped every Bluetooth input when an Apple device's advertisement data broke the JSON conversion. Read only Alias per device and fall back to the address. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
119 lines
7.3 KiB
Python
119 lines
7.3 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0,str(Path(__file__).parents[1]/'scripts'))
|
|
import input_controls as a
|
|
import web_server as w
|
|
|
|
|
|
class InputTests(unittest.TestCase):
|
|
def test_bus_property_scalar_and_method_tuple(self):
|
|
with patch.object(a,'command',return_value='{"type":"d","data":-15.0}'):
|
|
self.assertEqual(a.bus('get-property','service','path','interface','Volume'),-15.0)
|
|
with patch.object(a,'command',return_value='{"type":"a{sv}","data":[{"Volume":-15.0}]}'):
|
|
self.assertEqual(a.bus('call','service','path','interface','GetAll'),{'Volume':-15.0})
|
|
|
|
def test_headphone_volume_write_is_rejected(self):
|
|
props={'Device':'/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF','Mode':'sink','Transport':'A2DP-source','Volume':0x7f7f}
|
|
with patch.object(a,'pcms',return_value={'/output':props}), patch.object(a,'command') as command:
|
|
with self.assertRaises(ValueError):a.set_input('AA:BB:CC:DD:EE:FF',volume=20)
|
|
command.assert_not_called()
|
|
|
|
def test_bluetooth_only_writes_input_pcm(self):
|
|
source={'Device':'/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF','Mode':'source','Transport':'A2DP-sink','Volume':0x7f7f}
|
|
sink={**source,'Mode':'sink','Transport':'A2DP-source'}
|
|
with patch.object(a,'pcms',return_value={'/input':source,'/output':sink}),patch.object(a,'command') as command:
|
|
a.set_input('AA:BB:CC:DD:EE:FF',volume=40)
|
|
self.assertEqual(command.call_count,2)
|
|
self.assertTrue(all('/input' in call.args for call in command.call_args_list))
|
|
self.assertTrue(all('/output' not in call.args for call in command.call_args_list))
|
|
|
|
def test_airplay_uses_existing_native_volume(self):
|
|
with patch.object(a,'airplay',return_value={'volume':80,'muted':False}),patch.object(a,'remember'),patch.object(a,'command') as command:
|
|
a.set_input('airplay',volume=50)
|
|
self.assertEqual(command.call_args.args,('busctl','--system','--','set-property',a.SPS,a.SPS_PATH,a.SPS,'Volume','d','-15.0'))
|
|
a.set_input('airplay',mute=True)
|
|
self.assertEqual(command.call_args.args[-1],'-144.0')
|
|
|
|
def test_volume_validation_rejects_outside_range_and_bool(self):
|
|
for level in (-1,101,True,'50'):
|
|
with self.subTest(level=level), self.assertRaises(ValueError): a.set_input('airplay',volume=level)
|
|
|
|
def test_packed_bluetooth_stereo_mute(self):
|
|
value=a.encode_volume(60,True)
|
|
level,muted=a.decode_volume(value)
|
|
self.assertEqual(level,60);self.assertTrue(muted)
|
|
self.assertEqual((value>>8)&255,value&255)
|
|
|
|
|
|
class WebTests(unittest.TestCase):
|
|
def test_discoverable_toggle_is_allowlisted(self):
|
|
self.assertEqual(w.validate_action({'action':'discoverable','on':True,'seconds':300}),
|
|
['discoverable','on','--seconds','300'])
|
|
self.assertEqual(w.validate_action({'action':'discoverable','on':False}),['discoverable','off'])
|
|
for data in [{'action':'discoverable','on':True,'seconds':3600},{'action':'discoverable','on':True,'seconds':'300'},
|
|
{'action':'discoverable','on':'yes','seconds':300},{'action':'discoverable','on':True,'seconds':True}]:
|
|
with self.subTest(data=data), self.assertRaises(ValueError): w.validate_action(data)
|
|
|
|
def test_bluetooth_input_listed_even_if_name_unreadable(self):
|
|
source={'Device':'/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF','Mode':'source','Transport':'A2DP-sink','Volume':0x4040,'Running':True}
|
|
with patch.object(a,'airplay',side_effect=ValueError), patch.object(a,'pcms',return_value={'/input':source}), \
|
|
patch.object(a,'properties',side_effect=ValueError('json')):
|
|
items=a.inputs()
|
|
self.assertEqual([(i['id'],i['name'],i['playing']) for i in items if i['kind']=='bluetooth'],
|
|
[('AA:BB:CC:DD:EE:FF','AA:BB:CC:DD:EE:FF',True)])
|
|
|
|
def test_devices_skip_unreadable_and_tolerate_missing_uuids(self):
|
|
out={'/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF':'{"type":"s","data":"AA:BB:CC:DD:EE:FF"}\n{"type":"s","data":"QCY"}\n'
|
|
'{"type":"b","data":true}\n{"type":"b","data":true}\n{"type":"b","data":false}'}
|
|
def command(*args):
|
|
if 'tree' in args: return '/org/bluez\n/org/bluez/hci0\n/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF\n/org/bluez/hci0/dev_11_22_33_44_55_66\n'
|
|
path=args[5]
|
|
if path not in out or 'UUIDs' in args: raise ValueError('No such property')
|
|
return out[path]
|
|
with patch.object(a,'command',side_effect=command):
|
|
self.assertEqual(a.devices(),[{'address':'AA:BB:CC:DD:EE:FF','name':'QCY','bonded':True,'paired':True,
|
|
'connected':False,'output':False}])
|
|
|
|
def test_pairing_state_reports_remaining_only_when_discoverable(self):
|
|
adapter={'Discoverable':True,'Alias':'Radxa Audio Bluetooth'}
|
|
with patch.object(a,'bluez_paths',return_value=['/org/bluez/hci0']), \
|
|
patch.object(a,'properties',return_value=adapter), patch.object(a,'PAIRING_UNTIL') as until, \
|
|
patch.object(a.time,'time',return_value=1000.0):
|
|
until.read_text.return_value='1120.0'
|
|
self.assertEqual(a.pairing(),{'discoverable':True,'remaining':120,'alias':'Radxa Audio Bluetooth'})
|
|
adapter['Discoverable']=False
|
|
self.assertEqual(a.pairing()['remaining'],0)
|
|
|
|
def test_dac_selection_is_one_atomic_cli_command(self):
|
|
self.assertEqual(w.validate_action({'action':'output','mode':'dac','pcm':'hw:CARD=Audio,DEV=0'}),
|
|
['output','dac','--pcm','hw:CARD=Audio,DEV=0'])
|
|
|
|
def test_no_output_volume_api(self):
|
|
for data in [{'action':'output-volume','volume':20},{'action':'volume','input':'dac','volume':20},
|
|
{'action':'volume','input':'airplay','volume':101},
|
|
{'action':'output','mode':'bluetooth','address':'AA:BB:CC:DD:EE:FF;reboot'}]:
|
|
with self.subTest(data=data),self.assertRaises(ValueError):w.validate_action(data)
|
|
|
|
def test_http_requires_token_and_same_origin(self):
|
|
import threading,json,urllib.request,urllib.error
|
|
server=w.ThreadingHTTPServer(('127.0.0.1',0),w.Handler)
|
|
thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
|
|
url='http://127.0.0.1:'+str(server.server_address[1])+'/api/action'
|
|
try:
|
|
with patch.object(w,'allowed_host',return_value=True),patch.object(w,'launch',return_value='mockjob'):
|
|
body=json.dumps({'action':'scan'}).encode()
|
|
for headers in [{'Content-Type':'application/json'}, {'Content-Type':'application/json','X-Mixer-Token':w.TOKEN,'Origin':'http://other.site'}]:
|
|
req=urllib.request.Request(url,body,headers=headers)
|
|
with self.assertRaises(urllib.error.HTTPError) as error:urllib.request.urlopen(req,timeout=5)
|
|
self.assertEqual(error.exception.code,403);error.exception.close()
|
|
req=urllib.request.Request(url,body,headers={'Content-Type':'application/json','X-Mixer-Token':w.TOKEN})
|
|
with urllib.request.urlopen(req,timeout=5) as response:self.assertEqual(response.status,202)
|
|
finally:server.shutdown();server.server_close();thread.join()
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|