"""V2 protocol and persistence checks. Never opens a vehicle connection.""" import math import queue import tempfile import time import unittest from pathlib import Path from types import SimpleNamespace from unittest.mock import Mock, patch from app import Axes, MavlinkWorker, normalize_endpoint from rov_v2 import Telemetry, clean_station, clean_profile, decode_telemetry, save_json from web_ui import ControlApi class V2Tests(unittest.TestCase): def worker(self, frame=0, version=(4, 5, 3)): worker = MavlinkWorker(queue.Queue()) worker.connection = Mock() worker.target_system = 1 worker.frame_type = frame worker.firmware_version = version worker.gain = .5 worker.last_control_at = time.monotonic() - 1 return worker def test_six_axis_mapping_and_envelope(self): worker = self.worker() worker.set_axes(Axes(forward=1, lateral=1, vertical=1, yaw=1, pitch=1, roll=1)) worker._send_control() args = worker.connection.mav.manual_control_send.call_args.args self.assertEqual(args[6:8], (0, 3)) self.assertEqual(args[8:], (33, 33)) # S=pitch, T=roll; ArduSub 4.5 ±400. self.assertEqual(args[1:5], (83, 83, 542, 83)) def test_modern_normalized_attitude(self): worker = self.worker(version=(4, 8, 0)) worker.set_axes(Axes(pitch=.5, roll=-.5)) worker._send_control() self.assertEqual(worker.connection.mav.manual_control_send.call_args.args[8:], (250, -250)) def test_unsupported_frame_and_old_firmware(self): worker = self.worker(frame=1) worker.set_axes(Axes(pitch=1, roll=.2)) worker._send_control() self.assertEqual(worker.connection.mav.manual_control_send.call_args.args[8], 0) worker.firmware_version = (4, 1, 0) worker._send_control() self.assertEqual(len(worker.connection.mav.manual_control_send.call_args.args), 6) def test_watchdog_neutralizes_extended_axes(self): worker = self.worker() worker.require_input_lease = True worker.armed = True worker.last_heartbeat = time.monotonic() worker.set_axes(Axes(pitch=1, roll=-1), 1) worker.last_input_at -= 2 worker._send_control() self.assertEqual(worker.connection.mav.manual_control_send.call_args.args, (1, 0, 0, 500, 0, 0, 0, 3, 0, 0)) self.assertEqual(worker.axes, Axes()) def test_endpoints_preserve_configured_network_host(self): for endpoint in ('udpin:0.0.0.0:14550', 'udpout:192.168.2.2:14550', 'tcp:192.168.2.2:5760'): self.assertEqual(normalize_endpoint(endpoint), endpoint) def test_sensor_units_and_staleness(self): msg = SimpleNamespace(get_type=lambda: 'ATTITUDE', roll=math.pi/2, pitch=0, yaw=-math.pi/2, rollspeed=1, pitchspeed=0, yawspeed=0) group, values = decode_telemetry(msg) self.assertEqual(values['roll'], 90) self.assertEqual(values['heading'], 270) clock = [10] store = Telemetry(lambda: clock[0]) store.update(group, values) clock[0] = 14 self.assertTrue(store.snapshot()[group]['stale']) msg = SimpleNamespace(get_type=lambda: 'GLOBAL_POSITION_INT', relative_alt=-1200, vz=25) self.assertEqual(decode_telemetry(msg)[1], {'depth_m': 1.2, 'vertical_ms': .25}) def test_config_validation_and_atomic_roundtrip(self): with self.assertRaises(ValueError): clean_station({'video_url': 'javascript:alert(1)'}) with self.assertRaises(ValueError): clean_station({'lights_relay': 1, 'camera_relay': 1}) with self.assertRaises(ValueError): clean_profile({'version': 2, 'expo': float('nan')}) with tempfile.TemporaryDirectory() as folder: target = Path(folder) / 'station.json' save_json(target, clean_station({'frame': 2})) import json self.assertEqual(json.loads(target.read_text())['frame'], 2) self.assertFalse(target.with_suffix('.json.tmp').exists()) def test_stale_current_is_not_refreshed_by_voltage(self): api = ControlApi() api.state['connected'] = True with patch('web_ui.time.monotonic', return_value=10): api._consume_battery({'current_a': 4, 'voltage': 16}) with patch('web_ui.time.monotonic', return_value=16): api._consume_battery({'voltage': 15.9}) self.assertIsNone(api._snapshot()['battery']['current_a']) with patch('web_ui.time.monotonic', return_value=17): api._consume_battery({'current_a': None}) self.assertIsNone(api.state['battery']['current_a']) def test_attitude_stream_is_never_disabled_when_requested(self): worker = self.worker() worker._configure_low_bandwidth_streams() rates = [(name, interval) for name, _, interval in worker.stream_config] self.assertIn(('ATTITUDE', 500000), rates) self.assertNotIn(('ATTITUDE', -1), rates) def test_network_control_omits_serial_priming_heartbeat(self): worker = self.worker() worker.network_link = True worker._send_control() worker.connection.mav.heartbeat_send.assert_not_called() def test_profile_saves_atomically_and_cannot_change_while_armed(self): api = ControlApi() profile = {'version': 2, 'bindings': {}, 'actions': {}} with tempfile.TemporaryDirectory() as folder: api.profiles_path = Path(folder) / 'profiles.json' api.save_controller_profile('$xbox', profile) self.assertTrue(api.profiles_path.is_file()) before = api.profiles_path.read_bytes() api.state.update(connected=True, armed=True) with self.assertRaises(RuntimeError): api.save_controller_profile('$xbox', profile) self.assertEqual(before, api.profiles_path.read_bytes()) def test_native_recording_chunk_roundtrip_and_expired_token(self): import base64 api = ControlApi() with tempfile.TemporaryDirectory() as folder, patch('web_ui.__file__', str(Path(folder) / 'web_ui.py')): token = api.recording_begin() api.recording_chunk(token, base64.b64encode(b'video chunk').decode()) path = api.recording_end(token) self.assertEqual(Path(path).read_bytes(), b'video chunk') with self.assertRaises(ValueError): api.recording_chunk(token, 'AA==') def test_relay_action_waits_for_ack(self): api = ControlApi() api.station_settings['lights_relay'] = 0 worker = api.worker worker.connection = Mock() worker.target_system = worker.target_component = 1 worker.last_heartbeat = time.monotonic() api.set_accessory('lights', True) self.assertIsNone(api.state['accessories']['lights']['accepted']) api._consume(('accessory_ack', 'lights', True, 2)) self.assertFalse(api.state['accessories']['lights']['accepted']) self.assertEqual(api.state['accessories']['lights']['error'], 'Rejected') def test_motor_names_survive_app_restart(self): with tempfile.TemporaryDirectory() as folder, patch('web_ui.__file__', str(Path(folder) / 'web_ui.py')): first = ControlApi() settings = first.motor_settings settings['motors']['1']['name'] = 'Port' first.save_motor_settings(settings) self.assertEqual(ControlApi().motor_settings['motors']['1']['name'], 'Port') if __name__ == '__main__': unittest.main()