334 lines
15 KiB
Python
334 lines
15 KiB
Python
"""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()
|