"""No serial port is opened by these tests.""" import queue import unittest 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.worker.connection.mav.manual_control_send.call_args.args, (1, 0, 0, NEUTRAL_Z, 0, 0)) 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.worker.connection.mav.manual_control_send.call_args.args, (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()) 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()