105 lines
7.4 KiB
Python
105 lines
7.4 KiB
Python
|
import threading
|
||
|
|
import unittest
|
||
|
|
from unittest.mock import patch, Mock
|
||
|
|
from xml.etree.ElementTree import fromstring
|
||
|
|
import ptz
|
||
|
|
|
||
|
|
|
||
|
|
def xml(body):
|
||
|
|
return fromstring('<root xmlns:t="' + ptz.onvif.SCHEMA + '" xmlns:p="' + ptz.PTZ + '">' + body + '</root>')
|
||
|
|
|
||
|
|
|
||
|
|
class PresetTests(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
self.ctrl = ptz.Controller()
|
||
|
|
self.camera = dict(host='192.168.0.2', username='admin', password='example', mainPath='/stream1')
|
||
|
|
self.info = dict(endpoint=(80, '/onvif/PTZ'), profile='p1', node='shared')
|
||
|
|
self.snapshot = dict(revision='v1', maximum=12, items=[dict(token='1', name='Door')])
|
||
|
|
|
||
|
|
def test_read_presets_validates_device_and_list(self):
|
||
|
|
node = xml('<t:MaximumNumberOfPresets>12</t:MaximumNumberOfPresets>')
|
||
|
|
items = xml('<p:Preset token="1"><t:Name>Door</t:Name></p:Preset>')
|
||
|
|
with patch.object(self.ctrl, 'discover', return_value=self.info), patch.object(self.ctrl, 'call', side_effect=[node, items]):
|
||
|
|
result, _ = self.ctrl.read_presets(self.camera)
|
||
|
|
self.assertEqual(result['items'], self.snapshot['items'])
|
||
|
|
self.assertEqual(len(result['revision']), 64)
|
||
|
|
self.assertNotIn('endpoint', result)
|
||
|
|
for response in [xml('<t:MaximumNumberOfPresets>0</t:MaximumNumberOfPresets>'), xml('<t:MaximumNumberOfPresets>999999</t:MaximumNumberOfPresets>')]:
|
||
|
|
with patch.object(self.ctrl, 'discover', return_value=self.info), patch.object(self.ctrl, 'call', return_value=response):
|
||
|
|
with self.assertRaises(ValueError): self.ctrl.read_presets(self.camera)
|
||
|
|
with patch.object(self.ctrl, 'discover', return_value=self.info), patch.object(self.ctrl, 'call', side_effect=[node, xml('<p:Preset token="1"/><p:Preset token="1"/>')]):
|
||
|
|
with self.assertRaises(ValueError): self.ctrl.read_presets(self.camera)
|
||
|
|
|
||
|
|
def command(self, data, latest=None, response=None):
|
||
|
|
versions = [(self.snapshot, self.info)]
|
||
|
|
if latest is not None: versions.append((latest, self.info))
|
||
|
|
with patch.object(self.ctrl, 'discover', return_value=self.info), patch.object(self.ctrl, 'read_presets', side_effect=versions), patch.object(self.ctrl, 'call', return_value=response) as call:
|
||
|
|
result = self.ctrl.preset_command(self.camera, dict(revision='v1', **data))
|
||
|
|
return result, call
|
||
|
|
|
||
|
|
def test_creation_escapes_name_and_confirms_returned_token(self):
|
||
|
|
latest = dict(self.snapshot, items=[*self.snapshot['items'], dict(token='2', name='Yard & Gate')])
|
||
|
|
result, call = self.command(dict(action='create', name='Yard & Gate'), latest, xml('<p:PresetToken>2</p:PresetToken>'))
|
||
|
|
self.assertTrue(result['verified'])
|
||
|
|
self.assertIn('Yard & Gate', call.call_args.args[4])
|
||
|
|
self.assertNotIn('<m:PresetToken>', call.call_args.args[4])
|
||
|
|
|
||
|
|
def test_stale_snapshot_and_unknown_token_cannot_write(self):
|
||
|
|
for data in [dict(action='create', name='Yard', revision='old'), dict(action='goto', token='unknown', revision='v1')]:
|
||
|
|
with patch.object(self.ctrl, 'discover', return_value=self.info), patch.object(self.ctrl, 'read_presets', return_value=(self.snapshot,self.info)), patch.object(self.ctrl, 'call') as call:
|
||
|
|
with self.assertRaises(ValueError): self.ctrl.preset_command(self.camera, data)
|
||
|
|
call.assert_not_called()
|
||
|
|
self.assertFalse(self.ctrl.locks[self.ctrl.node_key(self.camera,self.info)].locked())
|
||
|
|
|
||
|
|
def test_invalid_names_duplicate_names_and_full_device_rejected(self):
|
||
|
|
for name in ['', 'x'*41, 'bad\nname', None, ' door ']:
|
||
|
|
with self.assertRaises(ValueError): self.command(dict(action='create', name=name))
|
||
|
|
self.snapshot['maximum']=1
|
||
|
|
with self.assertRaises(ValueError): self.command(dict(action='create',name='Yard'))
|
||
|
|
|
||
|
|
def test_remove_readback_and_unsupported_replace_rejected(self):
|
||
|
|
result,call=self.command(dict(action='remove',token='1'),dict(self.snapshot,items=[]))
|
||
|
|
self.assertTrue(result['verified']);self.assertEqual(call.call_args.args[3],'RemovePreset')
|
||
|
|
with patch.object(self.ctrl,'call') as call:
|
||
|
|
with self.assertRaises(ValueError):self.ctrl.preset_command(self.camera,dict(action='replace',token='1',name='Door'))
|
||
|
|
call.assert_not_called()
|
||
|
|
|
||
|
|
def test_ignored_write_is_not_reported_as_verified(self):
|
||
|
|
result,_=self.command(dict(action='create',name='Yard'),self.snapshot,xml('<p:PresetToken>2</p:PresetToken>'))
|
||
|
|
self.assertFalse(result['verified'])
|
||
|
|
|
||
|
|
def test_readback_failure_releases_lock_and_is_not_retried(self):
|
||
|
|
with patch.object(self.ctrl,'discover',return_value=self.info),patch.object(self.ctrl,'read_presets',side_effect=[(self.snapshot,self.info),OSError()]),patch.object(self.ctrl,'call') as call:
|
||
|
|
with self.assertRaises(OSError):self.ctrl.preset_command(self.camera,dict(action='remove',token='1',revision='v1'))
|
||
|
|
self.assertEqual(call.call_count,1)
|
||
|
|
self.assertFalse(self.ctrl.locks[self.ctrl.node_key(self.camera,self.info)].locked())
|
||
|
|
|
||
|
|
def test_goto_polls_device_and_always_stops(self):
|
||
|
|
status=xml('<t:MoveStatus><t:PanTilt>IDLE</t:PanTilt></t:MoveStatus>')
|
||
|
|
with patch.object(self.ctrl,'discover',return_value=self.info),patch.object(self.ctrl,'read_presets',return_value=(self.snapshot,self.info)),patch.object(self.ctrl,'call',side_effect=[None,status,None]) as call,patch('ptz.threading.Event',return_value=Mock(wait=Mock(return_value=False))):
|
||
|
|
result=self.ctrl.preset_command(self.camera,dict(action='goto',token='1',revision='v1'))
|
||
|
|
self.assertTrue(result['completed'])
|
||
|
|
self.assertEqual([c.args[3] for c in call.call_args_list],['GotoPreset','GetStatus','Stop'])
|
||
|
|
self.assertFalse(self.ctrl.cancellations)
|
||
|
|
|
||
|
|
def test_goto_network_failure_still_attempts_stop(self):
|
||
|
|
with patch.object(self.ctrl,'discover',return_value=self.info),patch.object(self.ctrl,'read_presets',return_value=(self.snapshot,self.info)),patch.object(self.ctrl,'call',side_effect=[OSError(),None]) as call:
|
||
|
|
with self.assertRaises(OSError):self.ctrl.preset_command(self.camera,dict(action='goto',token='1',revision='v1'))
|
||
|
|
self.assertEqual(call.call_args.args[3],'Stop');self.assertFalse(self.ctrl.cancellations)
|
||
|
|
|
||
|
|
def test_timeout_reports_incomplete_and_stops(self):
|
||
|
|
with patch.object(self.ctrl,'discover',return_value=self.info),patch.object(self.ctrl,'read_presets',return_value=(self.snapshot,self.info)),patch.object(self.ctrl,'call',return_value=xml('<t:MoveStatus><t:PanTilt>MOVING</t:PanTilt></t:MoveStatus>')) as call,patch('ptz.time.monotonic',side_effect=[0,0,11]),patch('ptz.threading.Event',return_value=Mock(wait=Mock(return_value=False))):
|
||
|
|
result=self.ctrl.preset_command(self.camera,dict(action='goto',token='1',revision='v1'))
|
||
|
|
self.assertFalse(result['completed']);self.assertEqual(call.call_args.args[3],'Stop')
|
||
|
|
|
||
|
|
def test_shared_node_blocks_preset_writes_but_stop_cancels(self):
|
||
|
|
key=self.ctrl.node_key(self.camera,self.info);lock=threading.Lock();lock.acquire();self.ctrl.locks[key]=lock
|
||
|
|
event=threading.Event();self.ctrl.cancellations[key]=event
|
||
|
|
with patch.object(self.ctrl,'discover',return_value=self.info),patch.object(self.ctrl,'call'):
|
||
|
|
with self.assertRaises(ValueError):self.ctrl.preset_command(self.camera,dict(action='create',name='Yard'))
|
||
|
|
self.ctrl.command(self.camera,'stop');self.assertTrue(event.is_set())
|
||
|
|
lock.release()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__=='__main__':unittest.main()
|