[AZ-558] Route C8 outbound encoder bytes through MavlinkTransport seam

All FC adapter outbound MAVLink bytes now go through the AZ-401
MavlinkTransport seam (NoopMavlinkTransport in replay,
SerialMavlinkTransport in live). New helpers in
_outbound_mavlink_payloads.py extract encode/pack/seq-bump so the four
AP _send sites and the iNav statustext _send site become
encode -> pack -> transport.write. TlogReplayFcAdapter emits real
AP-shape MAVLink bytes through the injected NoopMavlinkTransport,
satisfying replay protocol Invariant 5 and unblocking AZ-401 AC-9.

Closes AZ-558. Also unskips AZ-401 AC-9 and AZ-404 AC-4b. Live wire
output remains byte-identical (proven via two-instance MAVLink
byte-equivalence tests). AST scan asserts no .mav.<name>_send( calls
remain in the retrofit set (AP / iNav / tlog adapters).

Out of scope (logged in review): GCS adapter retrofit; airborne live
strategy registration that would activate the SerialMavlinkTransport
factory injection path.

Tests: 2110 passed, 92 environmental skips, 1 unrelated pre-existing
macOS cold-start flake deselected.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Oleksandr Bezdieniezhnykh
2026-05-16 05:33:45 +03:00
parent d7e6b0959e
commit 2b19b8b90b
18 changed files with 1106 additions and 90 deletions
@@ -31,6 +31,8 @@ from gps_denied_onboard.components.c8_fc_adapter.pymavlink_ardupilot_adapter imp
)
from gps_denied_onboard.config import load_config
from ._mav_test_helpers import _FakeMsg
_DEV_STATIC_KEY = "00112233445566778899aabbccddeeff" * 2 # 64 hex chars = 32 bytes
@@ -41,16 +43,20 @@ class _MavStub:
self.statustext_calls: list[tuple[int, bytes]] = []
# pymavlink exposes `connection.mav.signing.sig_count` after
# setup_signing(...); we simulate that surface here.
self.signing = SimpleNamespace(sig_count=signing_failure_count)
self.signing = SimpleNamespace(sig_count=signing_failure_count, sign_outgoing=False)
self.seq: int = 0
def gps_input_send(self, *args: Any) -> None:
def gps_input_encode(self, *args: Any) -> _FakeMsg:
self.gps_input_calls.append(args)
return _FakeMsg()
def named_value_float_send(self, time_boot_ms: int, name: bytes, value: float) -> None:
def named_value_float_encode(self, time_boot_ms: int, name: bytes, value: float) -> _FakeMsg:
self.named_value_float_calls.append((time_boot_ms, name, value))
return _FakeMsg()
def statustext_send(self, severity: int, text: bytes) -> None:
def statustext_encode(self, severity: int, text: bytes) -> _FakeMsg:
self.statustext_calls.append((severity, text))
return _FakeMsg()
class _ConnStub:
@@ -59,12 +65,17 @@ class _ConnStub:
self.setup_signing_calls: list[bytes] = []
self._fail_signing = fail_signing
self.closed = False
self.write_calls: list[bytes] = []
def setup_signing(self, key: bytes) -> None:
if self._fail_signing:
raise RuntimeError("simulated signing handshake refusal")
self.setup_signing_calls.append(bytes(key))
def write(self, payload: bytes) -> int:
self.write_calls.append(bytes(payload))
return len(payload)
def close(self) -> None:
self.closed = True