test: update new test cases for bridge polling

Test that bridge polling discovers downstream endpoints and
creates endpoints d-bus object.

Test that polling stops once downstream endpoint is discovered.
and continues unless endpoint reponds to send polling command
GET_ENDPOINT_ID.

Test that once brige endpoint is removed, downstream endpoints
associated to the brige also gets removed too.

Update bridge pool size to handle recursive bridge logic.
Test that endpoint polling continues with recursive bridge
topology as well

Signed-off-by: Faizan Ali <faizana@nvidia.com>

Author: faizana-nvidia

tests/mctpenv/__init__.py

@@ -349,6 +349,32 @@ def reset(self):
     def add_bridged_ep(self, ep):
         self.bridged_eps.append(ep)
 
+    def pool_size(self):
+        size = 0
+        if self.bridged_eps:
+            size += len(self.bridged_eps)
+            for br_ep in self.bridged_eps:
+                size += br_ep.pool_size()
+        return size
+
+    def assign_eids_recursive(self, pool_start):
+        next_eid = pool_start
+        for br_ep in self.bridged_eps:
+            br_ep.eid = next_eid
+            next_eid += 1
+            # Recursively assign to this endpoint's children
+            next_eid = br_ep.assign_eids_recursive(next_eid)
+        return next_eid
+
+    def lookup_route_recursive(self, eid):
+        for br_ep in self.bridged_eps:
+            if eid == br_ep.eid:
+                return br_ep
+
+            if br_ep.lookup_route_recursive(eid) is not None:
+                return br_ep
+        return None
+
     async def handle_mctp_message(self, sock, addr, data):
         # for us?
         if addr.eid == 0 or addr.eid == self.eid:
@@ -357,9 +383,9 @@ async def handle_mctp_message(self, sock, addr, data):
             else:
                 print(f"unknown MCTP message type {a.type}")
         else:
-            for br_ep in self.bridged_eps:
-                if addr.eid == br_ep.eid:
-                    return await br_ep.handle_mctp_message(sock, addr, data)
+            br_ep = self.lookup_route_recursive(addr.eid)
+            if br_ep:
+                return await br_ep.handle_mctp_message(sock, addr, data)
 
     async def handle_mctp_control(self, sock, addr, data):
         flags, opcode = data[0:2]
@@ -382,7 +408,7 @@ async def handle_mctp_control(self, sock, addr, data):
                 # Set Endpoint ID
                 (op, eid) = data[2:]
                 self.eid = eid
-                pool_size = len(self.bridged_eps)
+                pool_size = self.pool_size()
                 alloc_status = 0x00
                 # request a pool if we have one
                 if pool_size:
@@ -393,7 +419,7 @@ async def handle_mctp_control(self, sock, addr, data):
             elif opcode == 2:
                 # Get Endpoint ID
                 ep_type = 0
-                if len(self.bridged_eps) > 0:
+                if self.pool_size() > 0:
                     ep_type = 0x1 << 4
                 data = bytes(hdr + [0x00, self.eid, ep_type, 0x00])
                 await sock.send(raddr, data)
@@ -417,9 +443,8 @@ async def handle_mctp_control(self, sock, addr, data):
                     alloc_status = 0x01
                 else:
                     self.allocated_pool = (pool_start, pool_size)
-                    # Assign sequential EIDs starting from pool_start
-                    for (n, ep) in enumerate(self.bridged_eps[:pool_size]):
-                        ep.eid = self.allocated_pool[0] + n
+                    # Recursively assign EIDs to all downstream endpoints
+                    self.assign_eids_recursive(pool_start)
 
                 data = bytes(hdr + [0x00, alloc_status,
                                 self.allocated_pool[1], self.allocated_pool[0]])

tests/test_mctpd.py

@@ -1376,3 +1376,266 @@ async def test_register_vdm_type_support_errors(dbus, mctpd):
     with pytest.raises(asyncdbus.errors.DBusError) as ex:
         await mctp.call_register_vdm_type_support(0x00, v_type, 0x0001)
     assert str(ex.value) == "VDM type already registered"
+
+""" Test that we use endpoint poll interval from the config and
+    that we discover bridged endpoints via polling"""
+async def test_bridged_endpoint_poll(dbus, sysnet, nursery, autojump_clock):
+    poll_interval = 2500
+    config = f"""
+    [bus-owner]
+    endpoint_poll_ms = {poll_interval}
+    """
+
+    mctpd = MctpdWrapper(dbus, sysnet, config = config)
+    await mctpd.start_mctpd(nursery)
+
+    iface = mctpd.system.interfaces[0]
+    ep = mctpd.network.endpoints[0]
+    mctp = await mctpd_mctp_iface_obj(dbus, iface)
+
+    bridged_ep = [
+        Endpoint(iface, bytes(), types = [0, 1]),
+        Endpoint(iface, bytes(), types = [0, 1])
+    ]
+    for bep in bridged_ep:
+        mctpd.network.add_endpoint(bep)
+        ep.add_bridged_ep(bep)
+
+    (eid, _, path, new) = await mctp.call_assign_endpoint(ep.lladdr)
+    assert new
+
+    mctp_obj = await dbus.get_proxy_object(MCTPD_C, MCTPD_MCTP_P)
+    mctp_objmgr = await mctp_obj.get_interface(DBUS_OBJECT_MANAGER_I)
+    endpoint_added = trio.Semaphore(initial_value=0)
+
+    # We expect two bridged endpoints to be discovered
+    expected_bridged_eps = len(bridged_ep)
+    bridged_endpoints_found = []
+
+    def ep_added(ep_path, content):
+        if MCTPD_ENDPOINT_I in content:
+            bridged_endpoints_found.append(ep_path)
+            endpoint_added.release()
+
+    await mctp_objmgr.on_interfaces_added(ep_added)
+
+    # Wait for all expected bridged endpoints to be discovered
+    with trio.move_on_after(poll_interval / 1000 * 2) as expected:
+        for i in range(expected_bridged_eps):
+            await endpoint_added.acquire()
+
+    # Verify we found all expected bridged endpoints
+    assert not expected.cancelled_caught, "Timeout waiting for bridged endpoints"
+    assert len(bridged_endpoints_found) == expected_bridged_eps
+
+    res = await mctpd.stop_mctpd()
+    assert res == 0
+
+""" Test that all downstream endpoints are removed when the bridge
+    endpoint is removed"""
+async def test_bridged_endpoint_remove(dbus, sysnet, nursery, autojump_clock):
+    poll_interval = 2500
+    config = f"""
+    [bus-owner]
+    endpoint_poll_ms = {poll_interval}
+    """
+
+    mctpd = MctpdWrapper(dbus, sysnet, config = config)
+    await mctpd.start_mctpd(nursery)
+
+    iface = mctpd.system.interfaces[0]
+    ep = mctpd.network.endpoints[0]
+    mctp = await mctpd_mctp_iface_obj(dbus, iface)
+
+    bridged_ep = [
+        Endpoint(iface, bytes(), types = [0, 1]),
+        Endpoint(iface, bytes(), types = [0, 1])
+    ]
+    for bep in bridged_ep:
+        mctpd.network.add_endpoint(bep)
+        ep.add_bridged_ep(bep)
+
+    (eid, _, path, new) = await mctp.call_assign_endpoint(ep.lladdr)
+    assert new
+
+    # Wait for the bridged endpoints to be discovered
+    await trio.sleep(poll_interval / 1000)
+    removed = trio.Semaphore(initial_value = 0)
+    removed_eps = []
+
+    # Capture the removed endpoints
+    def ep_removed(ep_path, interfaces):
+        if MCTPD_ENDPOINT_I in interfaces:
+            removed.release()
+            removed_eps.append(ep_path)
+
+    mctp_obj = await dbus.get_proxy_object(MCTPD_C, MCTPD_MCTP_P)
+    mctp_objmgr = await mctp_obj.get_interface(DBUS_OBJECT_MANAGER_I)
+    await mctp_objmgr.on_interfaces_removed(ep_removed)
+
+    # Remove the bridge endpoint
+    bridge_obj = await mctpd_mctp_endpoint_control_obj(dbus, path)
+    await bridge_obj.call_remove()
+
+    # Assert that all downstream endpoints were removed
+    assert len(removed_eps) == (len(bridged_ep) + 1)
+    res = await mctpd.stop_mctpd()
+    assert res == 0
+
+""" Test that polling stops once endponit has been discovered """
+async def test_bridged_endpoint_poll_stop(dbus, sysnet, nursery, autojump_clock):
+    poll_interval = 2500
+    config = f"""
+    [bus-owner]
+    endpoint_poll_ms = {poll_interval}
+    """
+
+    mctpd = MctpdWrapper(dbus, sysnet, config = config)
+    await mctpd.start_mctpd(nursery)
+
+    iface = mctpd.system.interfaces[0]
+    ep = mctpd.network.endpoints[0]
+    mctp = await mctpd_mctp_iface_obj(dbus, iface)
+    poll_count = 0
+
+    class BridgedEndpoint(Endpoint):
+        async def handle_mctp_control(self, sock, src_addr, msg):
+            flags, opcode = msg[0:2]
+            if opcode == 0x2: # Get Endpoint ID
+                nonlocal poll_count
+                poll_count += 1
+            return await super().handle_mctp_control(sock, src_addr, msg)
+
+    bridged_ep = BridgedEndpoint(iface, bytes(), types = [0, 1])
+    mctpd.network.add_endpoint(bridged_ep)
+    ep.add_bridged_ep(bridged_ep)
+
+    (eid, _, path, new) = await mctp.call_assign_endpoint(ep.lladdr)
+    assert new
+
+    mctp_obj = await dbus.get_proxy_object(MCTPD_C, MCTPD_MCTP_P)
+    mctp_objmgr = await mctp_obj.get_interface(DBUS_OBJECT_MANAGER_I)
+    endpoint_added = trio.Semaphore(initial_value=0)
+    poll_count_by_discovery = 0
+
+    def ep_added(ep_path, content):
+        if MCTPD_ENDPOINT_I in content:
+            nonlocal poll_count_by_discovery
+            poll_count_by_discovery = poll_count
+            endpoint_added.release()
+
+    await mctp_objmgr.on_interfaces_added(ep_added)
+
+    # Wait longer than the poll interval for the bridged endpoint
+    # to be discovered
+    await trio.sleep(poll_interval / 1000)
+
+    # We should have only poll until the discovery thus count should
+    # be the same even after longer wait.
+    assert poll_count == poll_count_by_discovery
+
+    res = await mctpd.stop_mctpd()
+    assert res == 0
+
+""" Test that polling continues until the endpoint is discovered """
+async def test_bridged_endpoint_poll_continue(dbus, sysnet, nursery, autojump_clock):
+    poll_interval = 2500
+    config = f"""
+    [bus-owner]
+    endpoint_poll_ms = {poll_interval}
+    """
+
+    mctpd = MctpdWrapper(dbus, sysnet, config = config)
+    await mctpd.start_mctpd(nursery)
+
+    iface = mctpd.system.interfaces[0]
+    ep = mctpd.network.endpoints[0]
+    mctp = await mctpd_mctp_iface_obj(dbus, iface)
+    poll_count = 0
+
+    class BridgedEndpoint(Endpoint):
+        async def handle_mctp_control(self, sock, src_addr, msg):
+            flags, opcode = msg[0:2]
+            # dont respond to simiulate device not accessible
+            # but increment poll count for the Get Endpoint ID
+            if opcode == 0x2: # Get Endpoint ID
+                nonlocal poll_count
+                poll_count += 1
+            return None
+
+    bridged_ep = BridgedEndpoint(iface, bytes(), types = [0, 1])
+    mctpd.network.add_endpoint(bridged_ep)
+    ep.add_bridged_ep(bridged_ep)
+
+    (eid, _, path, new) = await mctp.call_assign_endpoint(ep.lladdr)
+    assert new
+
+    # Wait for sometime to continue polling
+    await trio.sleep(poll_interval / 1000)
+
+    poll_count_before = poll_count
+    # Wait more to see if poll count increments
+    await trio.sleep(1)
+    assert poll_count > poll_count_before
+
+    res = await mctpd.stop_mctpd()
+    assert res == 0
+
+""" Test polling continues for recursive bridged endpoints"""
+async def test_bridge_behind_another_bridge_poll(dbus, sysnet, nursery, autojump_clock):
+    poll_interval = 2500
+    config = f"""
+    [bus-owner]
+    endpoint_poll_ms = {poll_interval}
+    """
+
+    mctpd = MctpdWrapper(dbus, sysnet, config = config)
+    await mctpd.start_mctpd(nursery)
+
+    iface = mctpd.system.interfaces[0]
+    bridge1_ep = mctpd.network.endpoints[0]
+    mctp = await mctpd_mctp_iface_obj(dbus, iface)
+
+    bridged1_ep = [
+        Endpoint(iface, bytes(), types = [0, 1]),
+        Endpoint(iface, bytes(), types = [0, 1])
+    ]
+    for bep in bridged1_ep:
+        mctpd.network.add_endpoint(bep)
+        bridge1_ep.add_bridged_ep(bep)
+
+    bridge2_ep = Endpoint(iface, bytes([0x01]), types = [0, 1])
+    bridge2_ep.add_bridged_ep(Endpoint(iface, bytes([0x00]), types = [0, 1]))
+    mctpd.network.add_endpoint(bridge2_ep.bridged_eps[0])
+    mctpd.network.add_endpoint(bridge2_ep)
+    bridge1_ep.add_bridged_ep(bridge2_ep)
+
+    (eid, _, path, new) = await mctp.call_assign_endpoint(bridge1_ep.lladdr)
+    assert new
+
+    mctp_obj = await dbus.get_proxy_object(MCTPD_C, MCTPD_MCTP_P)
+    mctp_objmgr = await mctp_obj.get_interface(DBUS_OBJECT_MANAGER_I)
+    endpoint_added = trio.Semaphore(initial_value=0)
+
+    # We expect two bridged endpoints to be discovered
+    expected_bridged_eps = bridge1_ep.pool_size()
+    bridged_endpoints_found = []
+
+    def ep_added(ep_path, content):
+        if MCTPD_ENDPOINT_I in content:
+            bridged_endpoints_found.append(ep_path)
+            endpoint_added.release()
+
+    await mctp_objmgr.on_interfaces_added(ep_added)
+
+    # Wait for all expected bridged endpoints to be discovered
+    with trio.move_on_after(poll_interval / 1000 * 2) as expected:
+        for i in range(expected_bridged_eps):
+            await endpoint_added.acquire()
+
+    # Verify we found all expected bridged endpoints
+    assert not expected.cancelled_caught, "Timeout waiting for bridged endpoints"
+    assert len(bridged_endpoints_found) == expected_bridged_eps
+
+    res = await mctpd.stop_mctpd()
+    assert res == 0