Clean up UI windows and strengthen settings workflows

2026-09-30 15:22:09 +02:00
parent 7c85397af0
commit 0093f3f4aa
18 changed files with 2533 additions and 118 deletions
@@ -1,6 +1,9 @@
 """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
@@ -100,9 +103,18 @@ class WorkerWatchdogTests(unittest.TestCase):

    def assertNeutral(self):
        self.worker._send_control()
-        self.assertEqual(self.worker.connection.mav.manual_control_send.call_args.args,
+        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)
@@ -118,7 +130,7 @@ class WorkerWatchdogTests(unittest.TestCase):
            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.worker.connection.mav.manual_control_send.call_args.args,
+            self.assertEqual(self.control_values(),
                             (1, 50, 0, 525, 0, 0))

    def test_heartbeat_freeze_neutralizes_even_with_live_ui(self):
@@ -131,6 +143,160 @@ class WorkerWatchdogTests(unittest.TestCase):
        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):