| """ |
| mock_mctpd.py - Mock server for mctpd and Entity Manager in MctpUtil tests. |
| |
| This script simulates: |
| 1. mctpd emitting InterfacesAdded signals for MCTP endpoints. |
| 2. Entity Manager responding to GetManagedObjects queries. |
| |
| It also provides a control interface to trigger events from the test: |
| - TriggerAdd(endpoint_path, em_config_path) |
| - TriggerRemove(endpoint_path) |
| |
| To support precise synchronization without modifying product code, this mock |
| server emits custom D-Bus signals on the 'com.example.Control' interface: |
| - QueryStarted: Emitted when GetManagedObjects is called. |
| - QueryCompleted: Emitted when GetManagedObjects returns. |
| |
| Usage in tests: |
| 1. Start this script as a subprocess in SetUp(). |
| 2. Use `busctl call` to trigger events via TriggerAdd/TriggerRemove. |
| 3. Listen for QueryStarted/QueryCompleted to synchronize test steps. |
| """ |
| import os |
| import dbus |
| import dbus.service |
| import dbus.mainloop.glib |
| from gi.repository import GLib |
| import sys |
| import time |
| import signal |
| |
| AVAILABLE_CONFIGS = { |
| '/xyz/openbmc_project/inventory/system/board/MockBoard/mctp_device': { |
| 'xyz.openbmc_project.Configuration.MCTPUSBDevice': { |
| 'BusType': 'USB', |
| 'Port': '1.5.1', |
| 'Configuration': dbus.UInt64(1), |
| 'InterfaceNum': dbus.UInt64(4), |
| 'Name': 'config', |
| 'RootHubPath': '/sys/devices/platform/ahb/f0806000.usb', |
| 'Type': 'MCTPUSBDevice' |
| } |
| }, |
| '/xyz/openbmc_project/inventory/system/board/MockBoard/bridged_endpoint_config': { |
| 'xyz.openbmc_project.Configuration.MCTPBridgeDownstreamDevice': { |
| 'BridgeInventory': '/xyz/openbmc_project/inventory/system/board/MockBoard/mctp_device', |
| 'EIDOffset': dbus.UInt64(1), |
| 'Name': 'Bridged_EP', |
| 'Type': 'MCTPBridgeDownstreamDevice' |
| } |
| }, |
| '/xyz/openbmc_project/inventory/system/board/MockBoard/broken_endpoint_nonexistent': { |
| 'xyz.openbmc_project.Configuration.MCTPBridgeDownstreamDevice': { |
| 'BridgeInventory': '/xyz/openbmc_project/inventory/system/board/MockBoard/non_existent_bridge_path', |
| 'EIDOffset': dbus.UInt64(1), |
| 'Name': 'Broken_EP_Nonexistent', |
| 'Type': 'MCTPBridgeDownstreamDevice' |
| } |
| }, |
| '/xyz/openbmc_project/inventory/system/board/MockBoard/broken_endpoint_invalid': { |
| 'xyz.openbmc_project.Configuration.MCTPBridgeDownstreamDevice': { |
| 'BridgeInventory': '/xyz/openbmc_project/inventory/system/board/Ironheart/Ironheart_i2c_static_NVMe_2', |
| 'EIDOffset': dbus.UInt64(1), |
| 'Name': 'Broken_EP_Invalid', |
| 'Type': 'MCTPBridgeDownstreamDevice' |
| } |
| }, |
| '/xyz/openbmc_project/inventory/system/board/MockBoard/intermediate_bridge': { |
| 'xyz.openbmc_project.Configuration.MCTPBridgeDownstreamDevice': { |
| 'BridgeInventory': '/xyz/openbmc_project/inventory/system/board/MockBoard/mctp_device', |
| 'EIDOffset': dbus.UInt64(1), |
| 'Name': 'Intermediate_Bridge', |
| 'Type': 'MCTPBridgeDownstreamDevice' |
| } |
| }, |
| '/xyz/openbmc_project/inventory/system/board/MockBoard/multilevel_endpoint': { |
| 'xyz.openbmc_project.Configuration.MCTPBridgeDownstreamDevice': { |
| 'BridgeInventory': '/xyz/openbmc_project/inventory/system/board/MockBoard/intermediate_bridge', |
| 'EIDOffset': dbus.UInt64(2), |
| 'Name': 'Multilevel_EP', |
| 'Type': 'MCTPBridgeDownstreamDevice' |
| } |
| }, |
| '/xyz/openbmc_project/inventory/system/board/Ironheart/Ironheart_i2c_static_NVMe_2': { |
| 'xyz.openbmc_project.Configuration.NVME1000': { |
| 'Name': 'Ironheart_i2c_static_NVMe_2', |
| 'Protocol': 'mi_mctp', |
| 'Type': 'NVME1000' |
| }, |
| 'xyz.openbmc_project.Configuration.NVME1000.BusInfo0': { |
| 'BusType': 'USB', |
| 'InterfaceNum': dbus.UInt64(3), |
| 'Port': '1.5.1', |
| 'RootHubPath': '/sys/devices/platform/ahb/f0806000.usb' |
| } |
| } |
| } |
| |
| em = None |
| |
| |
| class MockEndpoint(dbus.service.Object): |
| def __init__(self, bus, path, em_config_path, msg_types=None): |
| dbus.service.Object.__init__(self, bus, path) |
| self._associations = dbus.Array([ |
| dbus.Struct(('configured_by', 'endpoint', em_config_path), signature='(sss)') |
| ], signature='(sss)') |
| if msg_types is not None: |
| self._message_types = dbus.ByteArray(msg_types) |
| else: |
| self._message_types = dbus.ByteArray(b'\x02\x00\x04') |
| |
| @dbus.service.method('org.freedesktop.DBus.Properties', in_signature='ss', out_signature='v') |
| def Get(self, interface_name, property_name): |
| if interface_name == 'xyz.openbmc_project.Association.Definitions' and property_name == 'Associations': |
| return self._associations |
| if interface_name == 'xyz.openbmc_project.MCTP.Endpoint' and property_name == 'SupportedMessageTypes': |
| return self._message_types |
| raise dbus.exceptions.DBusException('org.freedesktop.DBus.Error.InvalidArgs') |
| |
| @dbus.service.method('org.freedesktop.DBus.Properties', in_signature='s', out_signature='a{sv}') |
| def GetAll(self, interface_name): |
| if interface_name == 'xyz.openbmc_project.Association.Definitions': |
| return {'Associations': self._associations} |
| if interface_name == 'xyz.openbmc_project.MCTP.Endpoint': |
| return {'SupportedMessageTypes': self._message_types} |
| return {} |
| |
| |
| class MockConfigObject(dbus.service.Object): |
| def __init__(self, bus, path, interfaces): |
| dbus.service.Object.__init__(self, bus, path) |
| self.interfaces = interfaces |
| |
| @dbus.service.method('org.freedesktop.DBus.Properties', in_signature='s', out_signature='a{sv}') |
| def GetAll(self, interface_name): |
| return self.interfaces.get(interface_name, {}) |
| |
| |
| class MockMctpd(dbus.service.Object): |
| def __init__(self, bus): |
| dbus.service.Object.__init__(self, bus, '/au/com/codeconstruct/mctp1') |
| self.bus = bus |
| self.endpoints = {} |
| |
| @dbus.service.signal('org.freedesktop.DBus.ObjectManager', signature='oa{sa{sv}}') |
| def InterfacesAdded(self, object_path, interfaces_and_properties): |
| print(f"Emitting InterfacesAdded for {object_path}") |
| sys.stdout.flush() |
| |
| @dbus.service.signal('org.freedesktop.DBus.ObjectManager', signature='oas') |
| def InterfacesRemoved(self, object_path, interfaces): |
| pass |
| |
| @dbus.service.method('org.freedesktop.DBus.ObjectManager', out_signature='a{oa{sa{sv}}}') |
| def GetManagedObjects(self): |
| result = {} |
| for path, ep in self.endpoints.items(): |
| result[dbus.ObjectPath(path)] = { |
| 'xyz.openbmc_project.Association.Definitions': { |
| 'Associations': ep._associations |
| }, |
| 'xyz.openbmc_project.MCTP.Endpoint': { |
| 'SupportedMessageTypes': ep._message_types |
| } |
| } |
| return result |
| |
| @dbus.service.method('com.example.Control', in_signature='ss') |
| def TriggerAdd(self, endpoint_path, em_config_path): |
| if endpoint_path not in self.endpoints: |
| self.endpoints[endpoint_path] = MockEndpoint(self.bus, endpoint_path, em_config_path) |
| |
| self.InterfacesAdded(dbus.ObjectPath(endpoint_path), { |
| 'xyz.openbmc_project.Association.Definitions': { |
| 'Associations': self.endpoints[endpoint_path]._associations |
| }, |
| 'xyz.openbmc_project.MCTP.Endpoint': { |
| 'SupportedMessageTypes': self.endpoints[endpoint_path]._message_types |
| } |
| }) |
| |
| # Add to EM side dynamically! |
| if em is not None and em_config_path in AVAILABLE_CONFIGS: |
| config = AVAILABLE_CONFIGS[em_config_path] |
| if dbus.ObjectPath(em_config_path) not in em.managed_objects: |
| em.managed_objects[dbus.ObjectPath(em_config_path)] = config |
| # Create real object |
| MockConfigObject(self.bus, em_config_path, config) |
| # Emit signal from EM Object Manager! |
| em.InterfacesAdded(dbus.ObjectPath(em_config_path), config) |
| print(f"Emitted InterfacesAdded for EM config {em_config_path}") |
| sys.stdout.flush() |
| |
| return "OK" |
| |
| @dbus.service.method('com.example.Control', in_signature='ssay') |
| def TriggerAddCustom(self, endpoint_path, em_config_path, msg_types): |
| if endpoint_path not in self.endpoints: |
| self.endpoints[endpoint_path] = MockEndpoint(self.bus, endpoint_path, em_config_path, msg_types) |
| |
| self.InterfacesAdded(dbus.ObjectPath(endpoint_path), { |
| 'xyz.openbmc_project.Association.Definitions': { |
| 'Associations': self.endpoints[endpoint_path]._associations |
| }, |
| 'xyz.openbmc_project.MCTP.Endpoint': { |
| 'SupportedMessageTypes': self.endpoints[endpoint_path]._message_types |
| } |
| }) |
| |
| # Add to EM side dynamically! |
| if em is not None and em_config_path in AVAILABLE_CONFIGS: |
| config = AVAILABLE_CONFIGS[em_config_path] |
| if dbus.ObjectPath(em_config_path) not in em.managed_objects: |
| em.managed_objects[dbus.ObjectPath(em_config_path)] = config |
| # Create real object |
| MockConfigObject(self.bus, em_config_path, config) |
| # Emit signal from EM Object Manager! |
| em.InterfacesAdded(dbus.ObjectPath(em_config_path), config) |
| print(f"Emitted InterfacesAdded for EM config {em_config_path}") |
| sys.stdout.flush() |
| |
| return "OK" |
| |
| @dbus.service.method('com.example.Control', in_signature='s') |
| def TriggerRemove(self, endpoint_path): |
| print("TriggerRemove called for " + endpoint_path) |
| sys.stdout.flush() |
| if endpoint_path in self.endpoints: |
| self.InterfacesRemoved(dbus.ObjectPath(endpoint_path), ['xyz.openbmc_project.Association.Definitions']) |
| self.endpoints[endpoint_path].remove_from_connection() |
| del self.endpoints[endpoint_path] |
| return "OK" |
| return "Not Found" |
| |
| |
| class MockEM(dbus.service.Object): |
| def __init__(self, bus): |
| dbus.service.Object.__init__(self, bus, '/xyz/openbmc_project/inventory') |
| self.managed_objects = {} |
| self.fail_mode = False |
| |
| @dbus.service.method('com.example.Control', in_signature='b') |
| def SetFailMode(self, fail): |
| self.fail_mode = fail |
| print(f"EM fail mode set to {fail}") |
| sys.stdout.flush() |
| |
| @dbus.service.signal('com.example.Control', signature='') |
| def QueryStarted(self): |
| pass |
| |
| @dbus.service.signal('com.example.Control', signature='') |
| def QueryCompleted(self): |
| pass |
| |
| @dbus.service.signal('org.freedesktop.DBus.ObjectManager', signature='oa{sa{sv}}') |
| def InterfacesAdded(self, object_path, interfaces_and_properties): |
| pass |
| |
| @dbus.service.method('org.freedesktop.DBus.ObjectManager', out_signature='a{oa{sa{sv}}}') |
| def GetManagedObjects(self): |
| self.QueryStarted() |
| print("GetManagedObjects called, delaying reply...") |
| sys.stdout.flush() |
| time.sleep(1) |
| self.QueryCompleted() |
| if self.fail_mode: |
| print("Simulating EM failure!") |
| sys.stdout.flush() |
| raise dbus.exceptions.DBusException('org.freedesktop.DBus.Error.Failed', 'Simulated EM failure') |
| return self.managed_objects |
| |
| |
| def main(): |
| dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) |
| if 'DBUS_SESSION_BUS_ADDRESS' in os.environ: |
| bus = dbus.SessionBus() |
| else: |
| bus = dbus.SystemBus() |
| loop = GLib.MainLoop() |
| |
| def signal_handler(sig, frame): |
| print("Received SIGTERM, shutting down gracefully...") |
| loop.quit() |
| |
| signal.signal(signal.SIGTERM, signal_handler) |
| |
| global em |
| em = MockEM(bus) |
| mctpd = MockMctpd(bus) # noqa: F841 |
| |
| name_em = dbus.service.BusName("xyz.openbmc_project.EntityManager", bus) # noqa: F841 |
| name_mctp = dbus.service.BusName("au.com.codeconstruct.MCTP1", bus) # noqa: F841 |
| |
| print("Mock mctpd and EM running...") |
| sys.stdout.flush() |
| loop.run() |
| |
| |
| if __name__ == '__main__': |
| main() |
