tests: move lossy endpoint implementation to a separate class

Rather than adding test-specific behaviour to the base Endpoint object,
move the control-protocol-drop implementation to a new LossyEndpoint
object, internal to the retry_timeout test.

We may want to move this to mctpenv later, but it's only used by the one
test currently, and we don't have enough of an API definition to make
that common.

Signed-off-by: Jeremy Kerr <jk@codeconstruct.com.au>

Author: jk-ozlabs

tests/mctpenv/__init__.py

@@ -318,16 +318,14 @@ def to_buf(self):
         return bytes([flags, self.cmd]) + self.data
 
 class Endpoint:
-    def __init__(self, iface, lladdr, ep_uuid = None, eid = 0, types = None, timeout_opcodes = set(), timeout_count = 0):
+    def __init__(self, iface, lladdr, ep_uuid = None, eid = 0, types = None):
         self.iface = iface
         self.lladdr = lladdr
         self.uuid = ep_uuid or uuid.uuid1()
         self.eid = eid
         self.types = types or [0]
         self.bridged_eps = []
         self.allocated_pool = None # or (start, size)
-        self.timeout_opcodes = timeout_opcodes
-        self.timeout_count = timeout_count
 
         # keyed by (type, type-specific-instance)
         self.commands = {}
@@ -370,11 +368,6 @@ async def handle_mctp_control(self, sock, addr, data):
             # Use IID from request, zero Rq and D bits
             hdr = [iid, opcode]
 
-            if opcode in self.timeout_opcodes:
-                if self.timeout_count > 0:
-                    self.timeout_count -= 1
-                    return
-
             if opcode == 1:
                 # Set Endpoint ID
                 (op, eid) = data[2:]

tests/test_mctpd.py

@@ -1378,6 +1378,22 @@ async def test_register_vdm_type_support_errors(dbus, mctpd):
     assert str(ex.value) == "VDM type already registered"
 
 async def test_query_peer_properties_retry_timeout(nursery, dbus, sysnet):
+    class LossyEndpoint(Endpoint):
+        """An endpoint object that may drop a specific number (timeout_count)
+        of MCTP Control Protocol requests.
+        """
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            self.timeout_count = 0
+            self.timeout_opcode = 0x05 # Get Message Type Support
+
+        async def handle_mctp_control(self, sock, addr, data):
+            rq = data[0] & 0x80
+            opcode = data[1]
+            if rq and opcode == self.timeout_opcode and self.timeout_count:
+                self.timeout_count -= 1
+                return
+            return await super().handle_mctp_control(sock, addr, data)
 
     # activate mctpd
     mctpd = MctpdWrapper(dbus, sysnet)
@@ -1389,7 +1405,7 @@ async def test_query_peer_properties_retry_timeout(nursery, dbus, sysnet):
     # define expected message types
     # add a normal endpoint to network
     expected_types = [0, 1, 2]
-    ep = Endpoint(iface, bytes([0x1a]), eid=15, types=expected_types)
+    ep = LossyEndpoint(iface, bytes([0x1a]), eid=15, types=expected_types)
     mctpd.network.add_endpoint(ep)
 
     # call setup_endpoint on ep, which will allocate a object path for it
@@ -1401,7 +1417,6 @@ async def test_query_peer_properties_retry_timeout(nursery, dbus, sysnet):
     assert objtypes == expected_types
 
     ep.lladdr = bytes([0x1b])     # change lladdr to force retry
-    ep.timeout_opcodes.add(0x05)  # set Message Type Support would timeout
     ep.timeout_count = 2          # timeout twice before responding
 
     # call setup_endpoint again, which will trigger query of peer properties