import threading
import unittest
from unittest.mock import patch
from xml.etree.ElementTree import fromstring
import ptz


class PtzTests(unittest.TestCase):
    def setUp(self):
        self.controller = ptz.Controller()
        self.camera = dict(host='192.168.0.2', username='admin', password='example', mainPath='/stream1')
        self.info = dict(endpoint=(80, '/onvif/PTZ'), profile='profile1', node='shared')

    def test_nudge_has_device_deadline_and_explicit_stop(self):
        with patch.object(self.controller, 'discover', return_value=self.info), patch.object(self.controller, 'call') as call, patch('ptz.time.sleep'):
            self.assertTrue(self.controller.command(self.camera, 'left')['stopped'])
            self.assertEqual([c.args[3] for c in call.call_args_list], ['ContinuousMove', 'Stop'])
            self.assertIn('<m:Timeout>PT1S</m:Timeout>', call.call_args_list[0].args[4])
            self.assertIn('x="-0.25"', call.call_args_list[0].args[4])

    def test_failed_move_still_stops(self):
        with patch.object(self.controller, 'discover', return_value=self.info), patch.object(self.controller, 'call', side_effect=[OSError(), None]) as call:
            with self.assertRaises(OSError):
                self.controller.command(self.camera, 'up')
            self.assertEqual(call.call_args_list[-1].args[3], 'Stop')
            self.assertFalse(self.controller.locks[('192.168.0.2', 80, 'shared')].locked())

    def test_shared_node_rejects_overlapping_moves_but_allows_stop(self):
        lock=threading.Lock();lock.acquire();self.controller.locks[('192.168.0.2',80,'shared')]=lock
        with patch.object(self.controller, 'discover', return_value=self.info), patch.object(self.controller, 'call') as call:
            with self.assertRaises(ValueError):self.controller.command(self.camera,'right')
            self.controller.command(self.camera,'stop')
            self.assertEqual(call.call_args.args[3],'Stop')
        lock.release()

    def test_arbitrary_commands_and_foreign_services_rejected(self):
        for action in ['zoom','home','SetPreset',None]:
            with self.assertRaises(ValueError):self.controller.command(self.camera,action)
        for endpoint in ['http://example.org/onvif/PTZ','http://admin:secret@192.168.0.2/onvif/PTZ','https://192.168.0.2/onvif/PTZ']:
            with self.assertRaises(ValueError):self.controller.endpoint(self.camera,endpoint)

    def test_discovery_and_short_timeout_requirement(self):
        def xml(text):return fromstring('<root xmlns:tt="'+ptz.onvif.SCHEMA+'" xmlns:m="'+ptz.onvif.MEDIA+'">'+text+'</root>')
        caps=xml('<tt:PTZ><tt:XAddr>http://192.168.0.2/onvif/PTZ</tt:XAddr></tt:PTZ><tt:Media><tt:XAddr>http://192.168.0.2/onvif/Media</tt:XAddr></tt:Media>')
        profiles=xml('<m:Profiles token="profile1"><tt:PTZConfiguration token="config"><tt:NodeToken>shared</tt:NodeToken></tt:PTZConfiguration></m:Profiles>')
        template='<tt:ContinuousPanTiltVelocitySpace><tt:URI>'+ptz.SPACE+'</tt:URI><tt:XRange><tt:Min>-1</tt:Min><tt:Max>1</tt:Max></tt:XRange><tt:YRange><tt:Min>-1</tt:Min><tt:Max>1</tt:Max></tt:YRange></tt:ContinuousPanTiltVelocitySpace><tt:PTZTimeout><tt:Min>{minimum}</tt:Min><tt:Max>PT600S</tt:Max></tt:PTZTimeout>'
        self.camera['mainPath']='/stream1?profile=profile1'
        with patch.object(self.controller,'call',side_effect=[caps,profiles,xml(template.format(minimum='PT1S'))]):
            self.assertTrue(self.controller.capabilities(self.camera)['supported'])
        self.controller.cache.clear()
        with patch.object(self.controller,'call',side_effect=[caps,profiles,xml(template.format(minimum='PT5S'))]):
            with self.assertRaises(ValueError):self.controller.discover(self.camera)


if __name__=='__main__':unittest.main()