Add tests for AssignBridgeStatic D-Bus method

Add new test for validating AssignBridgeStatic D-Bus method
to verify bridge endpoint configuration and EID allocation.

- Add test_assign_dynamic_bridge_eid() for bridge simulation testing
- Simulate bridge with some downstream endpoints in test framework
- Test EID allocation of bridge and downstream being contiguous

Signed-off-by: Faizan Ali <[email protected]>

Author: faizana-nvidia

tests/mctpenv/__init__.py

@@ -348,7 +348,7 @@ async def handle_mctp_control(self, sock, addr, data):
                 # Set Endpoint ID
                 (op, eid) = data[2:]
                 self.eid = eid
-                data = bytes(hdr + [0x00, 0x00, self.eid, 0x00])
+                data = bytes(hdr + [0x00, 0x01 if len(self.bridged_eps) > 0 else 0x00, self.eid, len(self.bridged_eps)])
                 await sock.send(raddr, data)
 
             elif opcode == 2:
@@ -367,6 +367,18 @@ async def handle_mctp_control(self, sock, addr, data):
                 data = bytes(hdr + [0x00, len(types)] + types)
                 await sock.send(raddr, data)
 
+            elif opcode == 8:
+                # Allocate Endpoint IDs
+                (_, _, _, pool_size, pool_start) = data
+                num_eps = min(pool_size, len(self.bridged_eps))
+                data = bytes(hdr + [0x00, 0x00, num_eps, pool_start])
+
+                # Assign sequential EIDs starting from pool_start
+                for ep in range(num_eps):
+                    self.bridged_eps[ep].eid = pool_start + ep
+
+                await sock.send(raddr, data)
+
             else:
                 await sock.send(raddr, bytes(hdr + [0x05])) # unsupported command
 

tests/test_mctpd.py

@@ -831,3 +831,41 @@ async def test_interface_rename_with_peers(dbus, mctpd):
     # ensure the endpoint persists after rename
     ep_obj = await dbus.get_proxy_object(MCTPD_C, ep_path)
     assert ep_obj is not None
+""" Test bridge endpoint EID assignment and downstream endpoint EID allocation
+
+Tests that:
+- Bridge endpoint can be assigned a static EID
+- Downstream endpoints get sequential EIDs from pool start
+"""
+async def test_assign_dynamic_bridge_eid(dbus, mctpd):
+    iface = mctpd.system.interfaces[0]
+    mctp = await mctpd_mctp_iface_obj(dbus, iface)
+    ep = mctpd.network.endpoints[0]
+
+    pool_size = 2
+
+    # Set up bridged endpoints as undiscovered EID 0
+    for i in range(pool_size):
+        br_ep = Endpoint(iface, bytes(), types=[0, 2])
+        ep.add_bridged_ep(br_ep)
+        mctpd.network.add_endpoint(br_ep)
+
+    mctp = await mctpd_mctp_iface_obj(dbus, iface)
+
+    # dynamic EID assigment for dev1
+    (eid, _, path, new) = await mctp.call_assign_endpoint(
+        ep.lladdr,
+    )
+
+    assert new
+    # Assert for assigned bridge endpoint ID
+    assert path == f'/au/com/codeconstruct/mctp1/networks/1/endpoints/{eid}'
+    assert new
+
+    net = await mctpd_mctp_network_obj(dbus, iface.net)
+    for i in range(pool_size):
+        br_ep = ep.bridged_eps[i]
+        #check if the downstream endpoint eid is contiguous to the bridge endpoint eid
+        assert (eid + i + 1) == br_ep.eid
+        (path, new) = await net.call_learn_endpoint(br_ep.eid)
+        assert path == f'/au/com/codeconstruct/mctp1/networks/1/endpoints/{br_ep.eid}'