lda
Verified
038e152bbc
Checks / tests (ubuntu-latest, 3.11) (push) Canceled after 0s
Checks / tests (ubuntu-latest, 3.14) (push) Canceled after 0s
Checks / tests (windows-latest, 3.11) (push) Canceled after 0s
Checks / tests (windows-latest, 3.14) (push) Canceled after 0s
Checks / package (push) Canceled after 0s
64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
"""Exercise the connection state machine with a scripted serial peer."""
|
|
|
|
import queue
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from controller.link import Link
|
|
|
|
|
|
class HandshakeTests(unittest.TestCase):
|
|
def run_peer(self, malformed=False):
|
|
events = queue.Queue()
|
|
link = Link("COM7", events)
|
|
packets = []
|
|
# Replay actual device ACK + CONNECT_INFO, then a link-complete ACK.
|
|
# Peer data sequence 0x1d is independent of our outbound sequence 1.
|
|
info = (
|
|
bytes.fromhex("00 bb 01 1d 00 01 02 20")
|
|
if malformed
|
|
else bytes.fromhex("00 bb 01 1d 00 0b 02 10 01 34 30 04 01 02 04 05 01 b0")
|
|
)
|
|
reads = iter(
|
|
(bytes.fromhex("00 bb 00 01 00 00 01"), info, bytes.fromhex("00 bb 00 02 00 00 02"))
|
|
)
|
|
|
|
class Peer:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
pass
|
|
|
|
def write(self, data):
|
|
packets.append(data)
|
|
return len(data)
|
|
|
|
def read(self, size):
|
|
try:
|
|
return next(reads)
|
|
except StopIteration:
|
|
link.stop()
|
|
return b""
|
|
|
|
with patch.dict(
|
|
sys.modules, {"serial": types.SimpleNamespace(Serial=lambda *a, **k: Peer())}
|
|
):
|
|
link.run()
|
|
return packets, list(events.queue)
|
|
|
|
def test_waits_for_peer_before_ready(self):
|
|
packets, events = self.run_peer()
|
|
self.assertEqual([p[6:-1] for p in packets], [b"\x01", b"\x04"])
|
|
self.assertEqual(packets[1], bytes.fromhex("00 bb 01 02 00 01 04 08"))
|
|
self.assertEqual(sum(kind == "ready" for kind, _ in events), 1)
|
|
self.assertFalse(any(kind == "error" for kind, _ in events))
|
|
|
|
def test_malformed_info_never_enables_controls(self):
|
|
packets, events = self.run_peer(malformed=True)
|
|
self.assertEqual(len(packets), 1)
|
|
self.assertFalse(any(kind == "ready" for kind, _ in events))
|
|
self.assertTrue(any(kind == "error" for kind, _ in events))
|