"""No serial port is opened by these tests.""" import queue import threading import time import unittest from types import SimpleNamespace from unittest.mock import Mock, patch from app import Axes, MavlinkWorker, NEUTRAL_Z from vehicle_state import VehicleState from web_ui import ControlApi class VehicleStateTests(unittest.TestCase): def setUp(self): self.now = 10.0 self.state = VehicleState(lambda: self.now) self.state.heartbeat(False) def test_ack_is_not_armed_confirmation(self): self.state.request("arm") self.now += .2 self.state.ack("MAV_RESULT_ACCEPTED") self.assertFalse(self.state.snapshot()["armed"]) self.assertEqual(self.state.snapshot()["pending_command"], "arm") self.state.heartbeat(True) self.assertTrue(self.state.snapshot()["controls_allowed"]) def test_heartbeat_confirms_even_when_ack_packet_is_lost(self): self.state.request("arm") self.now += .5 self.state.heartbeat(True) self.assertIsNone(self.state.pending) self.assertTrue(self.state.armed) def test_old_events_cannot_confirm_new_request(self): self.state.request("arm") self.state.ack("MAV_RESULT_DENIED", received_at=9) self.state.heartbeat(True, received_at=9) self.assertIsNotNone(self.state.pending) self.assertFalse(self.state.armed) def test_rejected_engage_stays_disarmed(self): self.state.request("arm") self.now += .1 self.state.ack("MAV_RESULT_DENIED") self.assertIsNone(self.state.pending) self.assertIn("rejected", self.state.error) self.assertFalse(self.state.armed) def test_engage_timeout_then_late_armed_heartbeat_is_truth(self): self.state.request("arm") self.now += 6 self.state.tick() self.assertIn("timed out", self.state.error) self.assertFalse(self.state.snapshot()["armed_known"]) self.state.heartbeat(True) self.assertTrue(self.state.snapshot()["armed"]) def test_disarm_timeout_never_claims_disarmed(self): self.state.heartbeat(True) self.state.request("disarm") self.now += 6 self.state.heartbeat(True) snap = self.state.snapshot() self.assertTrue(snap["armed"]) self.assertIn("timed out", snap["command_error"]) def test_disarm_ack_waits_for_disarmed_heartbeat(self): self.state.heartbeat(True) self.state.request("disarm") self.now += .2 self.state.ack("MAV_RESULT_ACCEPTED") self.assertTrue(self.state.snapshot()["armed"]) self.assertFalse(self.state.snapshot()["controls_allowed"]) self.state.heartbeat(False) self.assertIsNone(self.state.pending) self.assertFalse(self.state.armed) def test_disconnect_preserves_last_known_armed_but_invalidates_it(self): self.state.heartbeat(True) self.state.disconnect() snap = self.state.snapshot() self.assertTrue(snap["armed"]) self.assertFalse(snap["armed_known"]) self.assertFalse(snap["controls_allowed"]) def test_frozen_telemetry_blocks_new_engage(self): self.now += 3.1 self.assertFalse(self.state.snapshot()["armed_known"]) with self.assertRaises(RuntimeError): self.state.request("arm") class WorkerWatchdogTests(unittest.TestCase): def setUp(self): self.worker = MavlinkWorker(queue.Queue()) self.worker.require_input_lease = True self.worker.connection = Mock() self.worker.target_system = 1 self.worker.armed = True self.worker.last_heartbeat = 10 def assertNeutral(self): self.worker._send_control() self.assertEqual(self.control_values(), (1, 0, 0, NEUTRAL_Z, 0, 0)) def control_values(self): return self.worker.connection.mav.manual_control_send.call_args.args def test_manual_control_uses_native_v2_without_forced_long_payload(self): self.worker._send_mav(self.worker.connection, 'manual_control_send', 1, 50, 0, 525, -20, 0) self.assertEqual(self.control_values(), (1, 50, 0, 525, -20, 0)) self.assertEqual(self.worker.connection.mav.manual_control_send.call_args.kwargs, {}) def test_ui_freeze_neutralizes_and_does_not_replay(self): with patch("app.time.monotonic", return_value=10): self.worker.set_axes(Axes(forward=1), 1) self.worker._send_control() self.assertGreater(self.worker.output_axes.forward, 0) with patch("app.time.monotonic", return_value=11.1): self.assertNeutral() self.assertEqual(self.worker.axes, Axes()) def test_intermediate_demands_reach_mavlink_with_output_limit(self): with patch("app.time.monotonic", return_value=10): self.worker.last_control_at = 9.9 self.worker.set_axes(Axes(forward=.25, vertical=.25), 1) self.worker._send_control() # 25% input at the default 20% limit: 5% command, not 100%. self.assertEqual(self.control_values(), (1, 50, 0, 525, 0, 0)) def test_heartbeat_freeze_neutralizes_even_with_live_ui(self): with patch("app.time.monotonic", return_value=13.1): self.worker.set_axes(Axes(forward=1), 1) self.assertNeutral() def test_stale_queued_input_cannot_override_neutral(self): self.worker.set_axes(Axes(), 20) self.worker.set_axes(Axes(forward=1), 19) self.assertEqual(self.worker.axes, Axes()) def test_mavlink_transmissions_are_serialized(self): first_entered = threading.Event() release_first = threading.Event() second_entered = threading.Event() class Mav: def first_send(self, **_): first_entered.set() release_first.wait(1) def second_send(self, **_): second_entered.set() connection = SimpleNamespace(mav=Mav()) first = threading.Thread(target=self.worker._send_mav, args=(connection, "first_send")) second = threading.Thread(target=self.worker._send_mav, args=(connection, "second_send")) first.start() self.assertTrue(first_entered.wait(.5)) second.start() time.sleep(.05) self.assertFalse(second_entered.is_set()) release_first.set() first.join(1) second.join(1) self.assertTrue(second_entered.is_set()) def test_live_console_is_read_only_bounded_and_keeps_newest_lines(self): for index in range(310): self.worker.log("RX_TEST", f"vehicle {index}") self.worker.log("TX_TEST", f"computer {index}") sides = self.worker.read_console(2) self.assertEqual(set(sides), {"rx", "tx"}) self.assertTrue(sides["rx"][0].endswith("[RX_TEST] vehicle 308")) self.assertTrue(sides["rx"][1].endswith("[RX_TEST] vehicle 309")) self.assertTrue(sides["tx"][1].endswith("[TX_TEST] computer 309")) self.assertEqual(len(self.worker.console_rx_lines), 300) self.assertEqual(len(self.worker.console_tx_lines), 300) self.assertEqual(len(self.worker.console_lines), 500) def test_arm_request_pauses_parameter_traffic(self): self.worker.armed = False self.worker.param_scan.append("SERVO1_FUNCTION") with patch("app.time.monotonic", return_value=10): self.worker.command_arm(True) self.worker.config_paused = False # A disarmed heartbeat arrived. self.worker._tick_param_scan(10.1) self.assertEqual(self.worker.command_quiet_until, 16) self.worker.connection.mav.param_request_read_send.assert_not_called() self.worker.connection.mav.command_long_send.assert_called_once() def test_arm_request_retries_until_state_confirmation(self): self.worker.armed = False with patch("app.time.monotonic", return_value=10): self.worker.command_arm(True) self.worker._tick_arm_retry(10.24) self.assertEqual(self.worker.connection.mav.command_long_send.call_count, 1) self.worker._tick_arm_retry(10.25) self.assertEqual(self.worker.connection.mav.command_long_send.call_count, 2) self.worker.pending_arm_state = None self.worker._tick_arm_retry(10.50) self.assertEqual(self.worker.connection.mav.command_long_send.call_count, 2) def test_outbound_command_has_valid_heartbeat_lead_in(self): self.worker._send_mav(self.worker.connection, "timesync_send", 0, 123) names = [call[0] for call in self.worker.connection.mav.method_calls] self.assertEqual(names[-2:], ["heartbeat_send", "timesync_send"]) def test_disarm_sends_neutral_and_pauses_parameter_traffic(self): self.worker.param_scan.append("SERVO1_FUNCTION") with patch("app.time.monotonic", return_value=10): self.worker.command_arm(False) self.worker.config_paused = False self.worker._tick_param_scan(10.1) self.assertEqual(self.worker.disarm_requested_until, 16) self.assertEqual(self.control_values(), (1, 0, 0, NEUTRAL_Z, 0, 0)) self.worker.connection.mav.param_request_read_send.assert_not_called() def test_telemetry_config_is_paced_and_includes_low_rate_esc(self): self.worker.armed = False with patch("app.time.monotonic", return_value=10): self.worker._configure_low_bandwidth_streams() self.assertEqual(len(self.worker.stream_config), 10) self.assertEqual(self.worker.stream_config[-1][0], "ESC_TELEMETRY_1_TO_4") self.assertEqual(self.worker.stream_config[-1][2], 1_000_000) self.worker.connection.mav.command_long_send.assert_not_called() self.worker._tick_stream_config(10) self.assertEqual(self.worker.connection.mav.command_long_send.call_count, 1) self.worker._tick_stream_config(10.1) self.assertEqual(self.worker.connection.mav.command_long_send.call_count, 1) self.worker.command_quiet_until = 16 self.worker._tick_stream_config(11) self.assertEqual(self.worker.connection.mav.command_long_send.call_count, 1) self.worker._tick_stream_config(16) self.assertEqual(self.worker.connection.mav.command_long_send.call_count, 2) def test_link_probe_uses_configured_rate_and_measures_exact_echo(self): self.worker.armed = False self.worker.ping_rate_hz = 5 with patch("app.time.monotonic", return_value=10), \ patch("app.time.monotonic_ns", return_value=10_000_000_000): self.worker._tick_link_probe(10) self.worker.connection.mav.timesync_send.assert_called_once_with( 0, 10_000_000_000) self.assertEqual(self.worker.pending_pings, {10_000_000_000: 10}) self.assertEqual(self.worker.next_ping_at, 10.2) message = SimpleNamespace( tc1=1, ts1=10_000_000_000, get_type=lambda: "TIMESYNC", to_dict=lambda: {"tc1": 1, "ts1": 10_000_000_000}) with patch("app.time.monotonic", return_value=10.04): self.worker._handle_message(message) quality = self.worker.events.get_nowait() self.assertEqual(quality[0], "link_quality") self.assertEqual(quality[1]["latency_ms"], 40.0) self.assertEqual(quality[1]["loss_percent"], 0.0) def test_link_rates_rebuild_requested_telemetry_intervals(self): self.worker.armed = False with patch("app.time.monotonic", return_value=10): self.worker.set_link_rates(10, { "SYS_STATUS": 5, "SERVO_OUTPUT_RAW": 10, "RC_CHANNELS": 0, "ESC_TELEMETRY_1_TO_4": 2, }) self.assertEqual(self.worker.ping_rate_hz, 10) configured = {name: interval for name, _, interval in self.worker.stream_config} self.assertEqual(configured["SYS_STATUS"], 200_000) self.assertEqual(configured["SERVO_OUTPUT_RAW"], 100_000) self.assertEqual(configured["RC_CHANNELS"], -1) self.assertEqual(configured["ESC_TELEMETRY_1_TO_4"], 500_000) def test_link_probe_counts_timeout_and_pauses_for_commands(self): self.worker.pending_pings[4] = 5 self.worker.rx_delivery_results.extend(((10, True), (10, False))) self.worker.next_ping_at = 0 self.worker.command_quiet_until = 20 with patch("app.time.monotonic", return_value=10): self.worker._tick_link_probe(10) self.worker.connection.mav.timesync_send.assert_not_called() self.assertEqual(list(self.worker.ping_results), [(10, None)]) quality = self.worker.events.get_nowait()[1] self.assertEqual(quality["loss_percent"], 100.0) self.assertEqual(quality["grade"], "testing") def test_clean_downlink_cannot_hide_failed_round_trips(self): self.worker.ping_results.extend([(10, None)] * 20) self.worker.rx_delivery_results.extend([(10, True)] * 200) with patch("app.time.monotonic", return_value=10): self.worker._publish_link_quality() quality = self.worker.events.get_nowait()[1] self.assertEqual(quality["loss_percent"], 100.0) self.assertEqual(quality["grade"], "poor") class BridgeTests(unittest.TestCase): def test_delayed_arm_after_focus_reset_or_expiry_is_not_sent(self): api = ControlApi() api.worker.connection = Mock() api.worker.target_system = 1 api._consume(("heartbeat", 1, "ARDUPILOT", "MANUAL", False)) api.worker.set_axes(Axes(), 20) with self.assertRaisesRegex(RuntimeError, "newer input reset"): api.command("arm", 19) with self.assertRaisesRegex(RuntimeError, "expired"): api.command("arm", 21, expires_at=0) api.worker.connection.mav.command_long_send.assert_not_called() def test_real_bridge_ack_pending_disconnect_and_stale_battery(self): api = ControlApi() api.worker.connection = Mock() api.worker.target_system = 1 api._consume(("heartbeat", 1, "ARDUPILOT", "MANUAL", False)) pending = api.command("arm") self.assertEqual(pending["pending_command"], "arm") self.assertFalse(pending["armed"]) api._consume(("ack", 400, "MAV_RESULT_ACCEPTED")) self.assertFalse(api._snapshot()["armed"]) api._consume(("heartbeat", 1, "ARDUPILOT", "MANUAL", True)) self.assertTrue(api._snapshot()["armed"]) api._consume(("disconnected", "Disconnected")) self.assertTrue(api._snapshot()["armed"]) self.assertFalse(api._snapshot()["armed_known"]) self.assertTrue(api._snapshot()["battery_stale"]) if __name__ == "__main__": unittest.main()