123 lines
8.6 KiB
Python
123 lines
8.6 KiB
Python
|
"""ONVIF contract tests against synthetic devices; no real camera is modified."""
|
||
|
|
import copy
|
||
|
|
import unittest
|
||
|
|
from unittest.mock import patch
|
||
|
|
from xml.etree import ElementTree as ET
|
||
|
|
import camera_settings as cs
|
||
|
|
import onvif
|
||
|
|
|
||
|
|
|
||
|
|
def xml(inner):
|
||
|
|
return ET.fromstring('<root xmlns:t="'+onvif.SCHEMA+'" xmlns:m="'+onvif.MEDIA+'" xmlns:i="'+cs.IMAGING+'">'+inner+'</root>')
|
||
|
|
|
||
|
|
|
||
|
|
class SettingsTests(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
self.controller=cs.Settings()
|
||
|
|
self.camera=dict(host='192.168.0.2',username='test',password='private',sourceKind='direct',enabled=True,
|
||
|
|
mainPath='/stream?profile=lens2',subPath='/sub?profile=lens2')
|
||
|
|
self.encoder=xml('<t:VideoEncoderConfiguration token="encoder2" encoding="H265"><t:Encoding>H264</t:Encoding>'
|
||
|
|
'<t:Resolution><t:Width>1920</t:Width><t:Height>1080</t:Height></t:Resolution><t:Quality>3</t:Quality>'
|
||
|
|
'<t:RateControl><t:FrameRateLimit>15</t:FrameRateLimit><t:BitrateLimit>2048</t:BitrateLimit></t:RateControl>'
|
||
|
|
'<t:H264><t:GovLength>60</t:GovLength><t:H264Profile>Main</t:H264Profile></t:H264>'
|
||
|
|
'<t:Multicast><t:TTL>1</t:TTL></t:Multicast><t:SessionTimeout>PT5S</t:SessionTimeout></t:VideoEncoderConfiguration>')[0]
|
||
|
|
self.info=dict(device=(80,'/onvif/device_service'),services={'Media':(80,'/media'),'Imaging':(80,'/imaging')},
|
||
|
|
source='source2',profile='lens2',encoder=self.encoder)
|
||
|
|
self.imaging=xml('<i:ImagingSettings><t:Brightness>50</t:Brightness><t:IrCutFilter>AUTO</t:IrCutFilter>'
|
||
|
|
'<t:Focus><t:AutoFocusMode>MANUAL</t:AutoFocusMode></t:Focus></i:ImagingSettings>')
|
||
|
|
self.image_options=xml('<i:ImagingOptions><t:Brightness><t:Min>0</t:Min><t:Max>100</t:Max></t:Brightness>'
|
||
|
|
'<t:IrCutFilterModes>AUTO</t:IrCutFilterModes><t:IrCutFilterModes>ON</t:IrCutFilterModes></i:ImagingOptions>')
|
||
|
|
self.video_options=xml('<m:Options><t:QualityRange><t:Min>0</t:Min><t:Max>5</t:Max></t:QualityRange>'
|
||
|
|
'<t:H264><t:FrameRateRange><t:Min>1</t:Min><t:Max>15</t:Max></t:FrameRateRange>'
|
||
|
|
'<t:ResolutionsAvailable><t:Width>1920</t:Width><t:Height>1080</t:Height></t:ResolutionsAvailable></t:H264>'
|
||
|
|
'<t:Extension><t:H264><t:BitrateRange><t:Min>256</t:Min><t:Max>2048</t:Max></t:BitrateRange></t:H264></t:Extension></m:Options>')
|
||
|
|
self.calls=[]
|
||
|
|
|
||
|
|
def call(self,camera,endpoint,ns,operation,body=''):
|
||
|
|
self.calls.append((operation,body))
|
||
|
|
if operation=='GetDeviceInformation':return xml('')
|
||
|
|
if operation=='GetImagingSettings':return copy.deepcopy(self.imaging)
|
||
|
|
if operation=='GetOptions':return self.image_options
|
||
|
|
if operation=='GetVideoEncoderConfigurationOptions':return self.video_options
|
||
|
|
if operation=='SetImagingSettings':
|
||
|
|
root=xml(body.replace('xmlns:', 'xmlns:'))
|
||
|
|
self.imaging=xml(ET.tostring(root.find(cs.I+'ImagingSettings'),encoding='unicode'))
|
||
|
|
return xml('')
|
||
|
|
return xml('')
|
||
|
|
|
||
|
|
def test_device_ranges_and_vendor_codec_are_exposed_without_credentials(self):
|
||
|
|
with patch.object(self.controller,'call',self.call):data=self.controller.details(self.camera,self.info)
|
||
|
|
self.assertEqual(data['imaging']['brightness']['max'],100)
|
||
|
|
self.assertEqual(data['imaging']['dayNight']['choices'],['AUTO','ON'])
|
||
|
|
self.assertEqual(data['video']['codec'],'H265')
|
||
|
|
self.assertEqual(data['video']['options']['bitrate']['max'],2048)
|
||
|
|
self.assertNotIn('private',str(data))
|
||
|
|
self.assertNotIn('sharpness',data['imaging'])
|
||
|
|
|
||
|
|
def test_rejects_bad_values_and_unsupported_controls(self):
|
||
|
|
rule={'brightness':dict(min=0,max=100)}
|
||
|
|
for value in [True,101,-1,float('nan'),float('inf'),'50']:
|
||
|
|
with self.assertRaises(ValueError):self.controller.validate({'brightness':value},rule)
|
||
|
|
for values in [{},{'unknown':1},None,[]]:
|
||
|
|
with self.assertRaises(ValueError):self.controller.validate(values,rule)
|
||
|
|
with self.assertRaises(ValueError):self.controller.validate({'fps':1.5},{'fps':dict(min=1,max=15)},('fps',))
|
||
|
|
with self.assertRaises(ValueError):self.controller.validate({'mode':'OFF'},{'mode':dict(choices=['AUTO'])})
|
||
|
|
|
||
|
|
def test_imaging_write_preserves_other_fields_and_verifies_readback(self):
|
||
|
|
with patch.object(self.controller,'discover',return_value=self.info),patch.object(self.controller,'call',self.call):
|
||
|
|
data,_=self.controller.read(self.camera)
|
||
|
|
result=self.controller.save(self.camera,dict(revision=data['revision'],section='imaging',values={'brightness':60},quality='main'))
|
||
|
|
self.assertTrue(result['verified'])
|
||
|
|
sent=next(body for op,body in self.calls if op=='SetImagingSettings')
|
||
|
|
self.assertIn('MANUAL',sent);self.assertIn('source2',sent);self.assertIn('ForcePersistence',sent)
|
||
|
|
|
||
|
|
def test_stale_revision_never_issues_device_write(self):
|
||
|
|
with patch.object(self.controller,'discover',return_value=self.info),patch.object(self.controller,'call',self.call):
|
||
|
|
with self.assertRaises(ValueError):self.controller.save(self.camera,dict(revision='old',section='imaging',values={'brightness':60}))
|
||
|
|
self.assertFalse(any(op.startswith('Set') for op,_ in self.calls))
|
||
|
|
|
||
|
|
def test_video_write_retains_codec_extension_multicast_and_other_settings(self):
|
||
|
|
with patch.object(self.controller,'discover',return_value=self.info),patch.object(self.controller,'call',self.call):
|
||
|
|
data,_=self.controller.read(self.camera)
|
||
|
|
result=self.controller.save(self.camera,dict(revision=data['revision'],section='video',values={'fps':10}))
|
||
|
|
sent=next(body for op,body in self.calls if op=='SetVideoEncoderConfiguration')
|
||
|
|
self.assertIn('encoding="H265"',sent);self.assertIn('Multicast',sent);self.assertIn('SessionTimeout',sent)
|
||
|
|
self.assertIn('>10<',sent);self.assertFalse(result['verified']) # Synthetic device deliberately ignores this write.
|
||
|
|
|
||
|
|
def test_device_timeout_after_save_does_not_falsely_claim_confirmation(self):
|
||
|
|
with patch.object(self.controller,'call',self.call):data=self.controller.details(self.camera,self.info)
|
||
|
|
with patch.object(self.controller,'read',side_effect=[(data,self.info),OSError()]),patch.object(self.controller,'call',return_value=xml('')):
|
||
|
|
result=self.controller.save(self.camera,dict(revision=data['revision'],section='imaging',values={'brightness':60}))
|
||
|
|
self.assertFalse(result['verified']);self.assertIsNone(result['settings'])
|
||
|
|
|
||
|
|
def test_onvif_requests_do_not_follow_redirects(self):
|
||
|
|
self.assertIsNone(onvif.NoRedirect().redirect_request(None, None, 302, 'Found', {}, 'http://other.example/'))
|
||
|
|
|
||
|
|
def test_network_endpoints_and_recorder_channels_are_rejected(self):
|
||
|
|
for address in ['http://evil.example/api','https://192.168.0.2/api','http://u:p@192.168.0.2/api','file:///etc/passwd']:
|
||
|
|
with self.assertRaises(ValueError):self.controller.endpoint(self.camera,address)
|
||
|
|
for overrides in [dict(sourceKind='recorder'),dict(enabled=False)]:
|
||
|
|
with self.assertRaises(ValueError):self.controller.discover(dict(self.camera,**overrides),'main')
|
||
|
|
with self.assertRaises(ValueError):self.controller.discover(self.camera,'bad')
|
||
|
|
|
||
|
|
def test_exact_profile_path_and_query_selects_second_lens(self):
|
||
|
|
selected=[]
|
||
|
|
def call(c,e,ns,op,body=''):
|
||
|
|
if op=='GetCapabilities':return xml('<t:Media><t:XAddr>http://192.168.0.2/media</t:XAddr></t:Media>')
|
||
|
|
if op=='GetProfiles':return xml(''.join('<m:Profiles token="lens'+str(i)+'"><t:VideoSourceConfiguration><t:SourceToken>source'+str(i)+'</t:SourceToken></t:VideoSourceConfiguration>'+ET.tostring(self.encoder,encoding='unicode')+'</m:Profiles>' for i in [1,2]))
|
||
|
|
if op=='GetStreamUri':
|
||
|
|
token='lens2' if 'lens2' in body else 'lens1';selected.append(token)
|
||
|
|
return xml('<t:Uri>rtsp://192.168.0.2/stream?profile='+token+'</t:Uri>')
|
||
|
|
with patch.object(self.controller,'call',call):info=self.controller.discover(self.camera,'main')
|
||
|
|
self.assertEqual(info['source'],'source2');self.assertEqual(selected,['lens1','lens2'])
|
||
|
|
|
||
|
|
def test_failed_options_disable_writes_but_keep_other_section_readable(self):
|
||
|
|
def call(*args):
|
||
|
|
if args[3]=='GetVideoEncoderConfigurationOptions':raise OSError()
|
||
|
|
return self.call(*args)
|
||
|
|
with patch.object(self.controller,'call',call):data=self.controller.details(self.camera,self.info)
|
||
|
|
self.assertEqual(data['video']['options'],{});self.assertIn('brightness',data['imaging']);self.assertTrue(data['notes'])
|
||
|
|
|
||
|
|
|
||
|
|
if __name__=='__main__':unittest.main()
|