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_bridge_static() for bridge assignment testing
- Simulate bridge with downstream endpoints in test framework
- Test EID allocation with pool start and size parameters

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

Author: faizana-nvidia

tests/mctpd/__init__.py

@@ -318,7 +318,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:
@@ -337,6 +337,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

@@ -686,3 +686,48 @@ async def test_add_interface(dbus, mctpd):
     assert eid == static_eid
     assert new
     assert mctpd.system.lookup_route(net, static_eid).iface == iface
+
+""" 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_bridge_static(dbus, mctpd):
+    iface = mctpd.system.interfaces[0]
+    ep = mctpd.network.endpoints[0]
+
+    bridge_eid = 12
+    pool_start = 13
+    pool_size = 2
+
+    # Set up bridged endpoints as undiscovered EID 0
+    for i in range(pool_size):
+        br_ep = Endpoint(iface, bytes(), eid=pool_start + i, types=[0, 2])
+        ep.add_bridged_ep(br_ep)
+        mctpd.network.add_endpoint(br_ep)
+
+    mctp = await mctpd_mctp_iface_obj(dbus, iface)
+
+    (eid, net, path, new, msg) = await mctp.call_assign_bridge_static(
+        ep.lladdr,        # physaddr
+        bridge_eid,       # static_eid
+        pool_start,       # pool_start
+        pool_size         # pool_size
+    )
+
+    # Assert for assigned bridge endpoint ID
+    assert eid == bridge_eid
+    assert path == f'/au/com/codeconstruct/mctp1/networks/1/endpoints/{bridge_eid}'
+    assert new
+
+    # Bridge Simulation : Add routes and neighbours and assert for downstream endpoints
+    # static neighbour; no gateway route support at present
+    net = await mctpd_mctp_network_obj(dbus, iface.net)
+    for i in range(pool_size):
+        br_ep = ep.bridged_eps[i]
+        await mctpd.system.add_route(mctpd.system.Route(iface, br_ep.eid, 1))
+        await mctpd.system.add_neighbour(mctpd.system.Neighbour(iface, ep.lladdr, 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}'
+        assert new