Add reproducible Radxa AirPlay and Bluetooth mixer setup

2026-10-12 17:16:51 +09:00
commit ceef64dfed
15 changed files with 1151 additions and 0 deletions
@@ -0,0 +1,46 @@
+import importlib.util
+import json
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import patch
+
+spec=importlib.util.spec_from_file_location('backup',Path(__file__).parents[1]/'scripts/backup.py')
+b=importlib.util.module_from_spec(spec); spec.loader.exec_module(b)
+
+
+class BackupTests(unittest.TestCase):
+    def test_restores_original_file_and_removes_new_file(self):
+        with tempfile.TemporaryDirectory() as d:
+            base=Path(d); original=base/'old.conf'; new=base/'new.conf'
+            original.write_text('original')
+            def ctl(*args): return subprocess.CompletedProcess(args, 1, stdout='')
+            with patch.object(b,'PATHS',[str(original),str(new)]), patch.object(b,'ctl',side_effect=ctl):
+                b.snapshot(base/'backup')
+                original.write_text('changed'); new.write_text('new')
+                b.restore(base/'backup')
+            self.assertEqual(original.read_text(),'original')
+            self.assertFalse(new.exists())
+
+    def test_preserves_symlinks(self):
+        with tempfile.TemporaryDirectory() as d:
+            base=Path(d); link=base/'service'; original=base/'target'
+            original.write_text('original'); link.symlink_to(original)
+            def ctl(*args): return subprocess.CompletedProcess(args, 1, stdout='')
+            with patch.object(b,'PATHS',[str(link)]), patch.object(b,'ctl',side_effect=ctl):
+                b.snapshot(base/'backup')
+                link.unlink(); link.write_text('replaced')
+                b.restore(base/'backup')
+            self.assertTrue(link.is_symlink())
+            self.assertEqual(link.read_text(),'original')
+
+    def test_rejects_snapshot_with_unexpected_file(self):
+        with tempfile.TemporaryDirectory() as d:
+            base=Path(d)
+            (base/'manifest.json').write_text(json.dumps({'files':{'/unexpected':True},'units':{}}))
+            with patch.object(b,'ctl') as ctl, self.assertRaises(ValueError): b.restore(base)
+            ctl.assert_not_called()
+
+
+if __name__=='__main__': unittest.main()
@@ -0,0 +1,79 @@
+import importlib.util
+from pathlib import Path
+import subprocess
+import unittest
+from unittest.mock import patch
+
+spec=importlib.util.spec_from_file_location('manager',Path(__file__).parents[1]/'scripts/radxa-audio.py')
+m=importlib.util.module_from_spec(spec); spec.loader.exec_module(m)
+
+
+class ManagerTests(unittest.TestCase):
+    def config(self, **extra):
+        return dict(dict(name='Radxa Desk', dac_pcm='hw:CARD=Audio,DEV=0',
+                         output='dac', bluetooth_device=''), **extra) if not extra else {
+            **self.config(), **extra}
+
+    def test_rejects_unsafe_hardware_and_addresses(self):
+        for key, value in [('dac_pcm', 'hw:CARD=Audio"; include "evil'),
+                           ('bluetooth_device', 'AA:BB:CC:DD:EE:FF;reboot'),
+                           ('name', 'line\nbreak')]:
+            with self.subTest(key=key), self.assertRaises(ValueError):
+                m.templates(self.config(**{key:value}))
+
+    def test_bluetooth_requires_address_before_any_action(self):
+        with patch.object(m, 'root'), patch.object(m, 'run') as run:
+            with self.assertRaises(ValueError): m.apply(self.config(output='bluetooth'))
+            run.assert_not_called()
+
+    def test_unavailable_headphone_does_not_stop_working_inputs(self):
+        def fake(*args, **kwargs):
+            return subprocess.CompletedProcess(args, 0, stdout='no device')
+        with patch.object(m, 'root'), patch.object(m, 'run', side_effect=fake) as run, patch.object(m, 'snapshot') as snapshot:
+            with self.assertRaises(ValueError):
+                m.apply(self.config(output='bluetooth',bluetooth_device='AA:BB:CC:DD:EE:FF'))
+            snapshot.assert_not_called()
+            self.assertFalse(any('stop' in c.args for c in run.call_args_list))
+
+    def test_two_stereo_inputs_share_same_mixer(self):
+        f=m.templates(self.config())
+        self.assertIn('"radxa_mix"',f['/etc/shairport-sync.conf'])
+        self.assertIn('--pcm=radxa_mix',f['/etc/systemd/system/bluealsa-aplay.service.d/10-radxa-mixer.conf'])
+        self.assertNotIn('--single-audio', ''.join(f.values()))
+        self.assertIn('ttable { 0.0 0.5 1.1 0.5 }',f['/etc/asound.conf'])
+        self.assertIn('rate 44100',f['/etc/asound.conf'])
+
+    def test_output_device_is_not_hardcoded(self):
+        address='AA:BB:CC:DD:EE:FF'
+        f=m.templates(self.config(output='bluetooth',bluetooth_device=address))
+        unit=f['/etc/systemd/system/radxa-bt-output.service']
+        self.assertIn('DEV='+address,unit)
+        self.assertNotIn('E4:58',unit)
+        self.assertIn('Loopback', f['/etc/asound.conf'])
+        self.assertIn('rate 48000',f['/etc/asound.conf'])
+
+    def test_name_quotes_are_escaped(self):
+        f=m.templates(self.config(name='Desk "Two"'))
+        self.assertIn('name = "Desk \\"Two\\"";',f['/etc/shairport-sync.conf'])
+
+    def test_pair_skips_repairing_registered_device(self):
+        def fake(*args, **kwargs):
+            return subprocess.CompletedProcess(args, 0, stdout='Paired: yes')
+        with patch.object(m, 'root'), patch.object(m, 'run', side_effect=fake) as run:
+            m.pair('AA:BB:CC:DD:EE:FF')
+            self.assertFalse(any('pair' in call.args for call in run.call_args_list))
+            self.assertTrue(any('trust' in call.args for call in run.call_args_list))
+
+    def test_failed_service_restores_configuration(self):
+        c=self.config()
+        with patch.object(m, 'root'), patch.object(m, 'run') as run, \
+             patch.object(m, 'snapshot', return_value=Path('/mock-backup')), \
+             patch.object(m, 'restore_files') as restore, \
+             patch.object(m, 'save'), patch.object(m, 'check_dac'), patch.object(m, 'active', return_value=False), \
+             patch.object(m, 'enabled', return_value=False):
+            with self.assertRaises(ValueError): m.apply(c)
+            restore.assert_called_once_with(Path('/mock-backup'))
+            self.assertTrue(any(c.args[:2]==('systemctl','daemon-reload') for c in run.call_args_list))
+
+
+if __name__=='__main__': unittest.main()