Codex
56 lines
3.7 KiB
Python
56 lines
3.7 KiB
Python
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()
|