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 1.0 1.1 1.0 }',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'])
        self.assertIn('--sbc-quality=medium',f['/etc/systemd/system/bluealsa.service.d/10-radxa-sink.conf'])
        self.assertIn('-t 200000',unit)

    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_bond_request_must_not_report_temporary_pair_as_success(self):
        def fake(*args, **kwargs):
            return subprocess.CompletedProcess(args,0,stdout='Paired: yes\nBonded: no')
        with patch.object(m,'root'),patch.object(m,'run',side_effect=fake) as run, \
                patch.object(m,'agent_pair') as agent:
            with self.assertRaises(ValueError):m.pair('AA:BB:CC:DD:EE:FF',output=True,bond=True)
            agent.assert_called_once_with('AA:BB:CC:DD:EE:FF')
            self.assertTrue(any('remove' in call.args for call in run.call_args_list))
            self.assertFalse(any('trust' in call.args for call in run.call_args_list))

    def test_bond_request_keeps_existing_bond(self):
        def fake(*args, **kwargs):
            return subprocess.CompletedProcess(args,0,stdout='Paired: yes\nBonded: yes')
        with patch.object(m,'root'),patch.object(m,'run',side_effect=fake) as run:
            m.pair('AA:BB:CC:DD:EE:FF',bond=True)
            self.assertFalse(any('pair' in call.args for call in run.call_args_list))

    def test_reconfigure_does_not_require_headphone_to_be_on(self):
        c=self.config(output='bluetooth',bluetooth_device='AA:BB:CC:DD:EE:FF')
        with patch.object(m,'root'), patch.object(m,'run') as run, patch.object(m,'load',return_value=c), \
             patch.object(m,'snapshot',return_value=Path('/mock')), patch.object(m,'save'), \
             patch.object(m,'active',return_value=True),patch.object(m,'enabled',return_value=True):
            m.apply(c,reconfigure=True)
            self.assertFalse(any('connect' 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()