168 lines
6.5 KiB
Python
168 lines
6.5 KiB
Python
"""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()
|