autotest: add tests for Plane MAVLink command and message handling

Coverage analysis of the Plane and QuadPlane autotest suites showed
large parts of ArduPlane/GCS_MAVLink_Plane.cpp were never executed.
These tests exercise that code:

 - AVAILABLE_MODES, including FLTMODE_GCSBLOCK and single-index requests
 - rejection paths of DO_REPOSITION, DO_CHANGE_ALTITUDE, DO_CHANGE_SPEED,
   GUIDED_CHANGE_SPEED/ALTITUDE/HEADING, MISSION_START, DO_LAND_START
   and the VTOL commands on a non-VTOL Plane
 - flying headings and course-over-ground with GUIDED_CHANGE_HEADING
 - SET_POSITION_TARGET_LOCAL_NED altitude offsets
 - SET_ATTITUDE_TARGET and SET_POSITION_TARGET_GLOBAL_INT being ignored
   outside GUIDED, and an invalid frame being reported
 - DO_RETURN_PATH_START as a command
 - SET_HAGL triggering the landing flare
 - DO_PARACHUTE enable, disable, unknown action and repeated release
 - DO_SET_MISSION_CURRENT as a command
 - DO_REPOSITION loiter radius and direction
 - DO_SET_HOME while in RTL
 - HEARTBEAT system_status through standby, active, failsafe and crash
 - EXTENDED_SYS_STATE landed_state through a fixed-wing flight
 - PID_TUNING axes selected by GCS_PID_MASK, including the landing PID

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Peter Barker
2026-09-26 08:05:40 +10:00
committed by Andrew Tridgell
co-authored by Claude Opus 5
parent c1f6786ede
commit 484ca5d8ad
2 changed files with 883 additions and 10 deletions
File diff suppressed because it is too large Load Diff
+89 -10
View File
@@ -12635,9 +12635,71 @@ Also, ignores heartbeats not from our target system'''
continue continue
return m return m
def get_messages_frame(self, msg_names): def request_available_modes(self, index=0, timeout=10):
'''request AVAILABLE_MODES using MAV_CMD_REQUEST_MESSAGE. index is
the 1-based mode_index wanted, 0 meaning every mode. Returns the
messages received keyed by mode_index, which is empty if nothing
arrived within timeout'''
self.context_push()
try:
self.context_collect('AVAILABLE_MODES')
self.run_cmd(
mavutil.mavlink.MAV_CMD_REQUEST_MESSAGE,
p1=mavutil.mavlink.MAVLINK_MSG_ID_AVAILABLE_MODES,
p2=index,
)
tstart = self.get_sim_time()
ret = {}
while self.get_sim_time_cached() - tstart < timeout:
self.mav.recv_match(blocking=True, timeout=0.1)
ret = {}
for m in self.context_collection('AVAILABLE_MODES'):
if m.mode_index in ret:
raise NotAchievedException(f"Received mode_index {m.mode_index} twice")
ret[m.mode_index] = m
if len(ret) == 0:
continue
if index != 0:
break
number_modes = next(iter(ret.values())).number_modes
if len(ret) == number_modes:
break
finally:
self.context_pop()
for m in ret.values():
self.progress(str(m))
return ret
def assert_available_modes(self, expected_modes, not_user_selectable):
'''request every mode with AVAILABLE_MODES and check the modes
reported match expected_modes, a dict of custom_mode: mode_name.
not_user_selectable is the collection of custom_mode numbers
expected to carry MAV_MODE_PROPERTY_NOT_USER_SELECTABLE. Returns
the messages keyed by mode_index'''
modes = self.request_available_modes()
want_count = len(expected_modes)
if sorted(modes.keys()) != list(range(1, want_count+1)):
raise NotAchievedException(f"Want mode_index 1..{want_count} got {sorted(modes.keys())}")
got_modes = {}
for m in modes.values():
if m.number_modes != want_count:
raise NotAchievedException(f"{m.mode_name}: want number_modes={want_count} got {m.number_modes}")
if m.custom_mode in got_modes:
raise NotAchievedException(f"custom_mode {m.custom_mode} reported twice")
got_modes[m.custom_mode] = m.mode_name
want_properties = 0
if m.custom_mode in not_user_selectable:
want_properties = mavutil.mavlink.MAV_MODE_PROPERTY_NOT_USER_SELECTABLE
if m.properties != want_properties:
raise NotAchievedException(f"{m.mode_name}: want properties={want_properties} got {m.properties}")
if got_modes != expected_modes:
raise NotAchievedException(f"Unexpected modes: want {expected_modes} got {got_modes}")
return modes
def get_messages_frame(self, msg_names, timeout=None):
'''try to get a "frame" of named messages - a set of messages as close '''try to get a "frame" of named messages - a set of messages as close
in time as possible''' in time as possible. timeout is in seconds of simulation time;
None waits forever'''
msgs = {} msgs = {}
def get_msgs(mav, m): def get_msgs(mav, m):
@@ -12645,15 +12707,20 @@ Also, ignores heartbeats not from our target system'''
if t in msg_names: if t in msg_names:
msgs[t] = m msgs[t] = m
self.do_timesync_roundtrip() self.do_timesync_roundtrip()
tstart = self.get_sim_time()
self.install_message_hook(get_msgs) self.install_message_hook(get_msgs)
for msg_name in msg_names: try:
self.send_poll_message(msg_name) for msg_name in msg_names:
while True: self.send_poll_message(msg_name)
self.mav.recv_match(blocking=True) while True:
if len(msgs.keys()) == len(msg_names): self.mav.recv_match(blocking=True, timeout=0.1)
break if len(msgs.keys()) == len(msg_names):
break
self.remove_message_hook(get_msgs) if timeout is not None and self.get_sim_time_cached() - tstart > timeout:
missing = set(msg_names) - set(msgs.keys())
raise NotAchievedException(f"Did not receive {sorted(missing)} within {timeout}s")
finally:
self.remove_message_hook(get_msgs)
return msgs return msgs
@@ -14346,6 +14413,18 @@ switch value'''
if m is not None: if m is not None:
raise PreconditionFailedException("Receiving %s messages" % message) raise PreconditionFailedException("Receiving %s messages" % message)
def received_pid_tuning_axes(self, duration=3):
'''return the set of PID_TUNING axes received over duration seconds'''
self.context_push()
try:
self.context_collect('PID_TUNING')
self.delay_sim_time(duration, reason="collect PID_TUNING")
axes = set([m.axis for m in self.context_collection('PID_TUNING')])
finally:
self.context_pop()
self.progress(f"PID_TUNING axes: {sorted(axes)}")
return axes
def PIDTuning(self): def PIDTuning(self):
'''Test PID Tuning''' '''Test PID Tuning'''
self.assert_not_receiving_message('PID_TUNING', timeout=5) self.assert_not_receiving_message('PID_TUNING', timeout=5)