cheng
54 lines
2.6 KiB
Python
54 lines
2.6 KiB
Python
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest import TestCase
|
|
from unittest.mock import patch
|
|
import threading
|
|
|
|
from station.remote_camera import settings,bridge_environment
|
|
from station.operator_camera_view import stream_targets
|
|
from station.control_service import StationService
|
|
from arm_application.site_io import load_site_dict
|
|
|
|
|
|
class RemotePreviewTest(TestCase):
|
|
def site(self):
|
|
site=load_site_dict();site['remote_camera_preview']={
|
|
'enabled':True,'domain':0,'host':'192.168.2.10','interface_address':'192.168.2.50',
|
|
'cameras':['d405_left','d435'],'transport':'raw'}
|
|
return site
|
|
|
|
def test_only_explicit_remote_streams_use_raw(self):
|
|
streams=stream_targets(self.site())
|
|
self.assertEqual({s['id'] for s in streams},{'d405_left','d435'})
|
|
self.assertTrue(all(s['preview_transport']=='raw' and s['source']=='remote' for s in streams))
|
|
|
|
def test_remote_environment_is_limited_to_preview_child(self):
|
|
parent={'ROS_DOMAIN_ID':'42','ROS_LOCALHOST_ONLY':'1','REAL_ROS_DOMAIN_ID':'42',
|
|
'ROS_DISCOVERY_SERVER':'some-other-server'}
|
|
before=dict(parent)
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
env=bridge_environment(self.site(),parent,folder)
|
|
self.assertEqual(parent,before)
|
|
self.assertEqual((env['ROS_DOMAIN_ID'],env['ROS_LOCALHOST_ONLY']),('0','0'))
|
|
self.assertEqual(env['REAL_ROS_DOMAIN_ID'],'42')
|
|
self.assertNotIn('ROS_DISCOVERY_SERVER',env)
|
|
xml=Path(env['FASTRTPS_DEFAULT_PROFILES_FILE']).read_text()
|
|
self.assertIn('<address>192.168.2.50</address>',xml)
|
|
self.assertIn('<address>192.168.2.10</address>',xml)
|
|
self.assertEqual(bridge_environment({},parent,'/unused'),parent)
|
|
|
|
def test_bad_address_is_rejected_before_creating_profile(self):
|
|
site=self.site();site['remote_camera_preview']['host']='bad<xml>'
|
|
with self.assertRaises(ValueError):settings(site)
|
|
|
|
def test_refresh_does_not_touch_acquisition_or_robot(self):
|
|
gateway=SimpleNamespace(states=lambda:({},{}),graph=lambda:{'nodes':[],'topics':[],'services':[]},close=lambda:None)
|
|
service=StationService(gateway=gateway);service.site.update(self.site())
|
|
try:
|
|
with patch.object(service,'_camera') as camera,patch.object(service.processes,'start') as start:
|
|
result=service.execute('camera_refresh','89','eye_in_hand',{},threading.Event())
|
|
self.assertTrue(result['ok']);camera.assert_not_called();start.assert_not_called()
|
|
finally:service.close()
|