tests/mock_mctpd.py

"""
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 dbus
import dbus.service
import dbus.mainloop.glib
from gi.repository import GLib
import sys
import time
import signal


class MockEndpoint(dbus.service.Object):
    def __init__(self, bus, path, em_config_path):
        dbus.service.Object.__init__(self, bus, path)
        self.associations = dbus.Array([
            dbus.Struct(('configured_by', 'endpoint', em_config_path), signature='(sss)')
        ], signature='(sss)')

    @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 dbus.LowLevel.Variant(self.associations, 'a(sss)')
        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': dbus.LowLevel.Variant(self.associations, 'a(sss)')}
        return {}


class MockTarget(dbus.service.Object):
    def __init__(self, bus):
        dbus.service.Object.__init__(self, bus, '/xyz/openbmc_project/inventory/system/board/MockBoard/mctp_device')

    @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.Configuration.BusInfo':
            return {
                'BusType': 'USB',
                'Port': '1.2.5',
                'Configuration': dbus.UInt32(1)
            }
        return {}


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('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': dbus.Array([
                    dbus.Struct(('configured_by', 'endpoint', em_config_path), signature='(sss)')
                ], signature='(sss)')
            }
        })
        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')

    @dbus.service.signal('com.example.Control', signature='')
    def QueryStarted(self):
        pass

    @dbus.service.signal('com.example.Control', signature='')
    def QueryCompleted(self):
        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()
        return {
            dbus.ObjectPath('/xyz/openbmc_project/inventory/system/board/MockBoard/mctp_device'): {
                'xyz.openbmc_project.Configuration.BusInfo': {
                    'BusType': 'USB',
                    'Port': '1.2.5',
                    'Configuration': dbus.UInt32(1)
                }
            }
        }


def main():
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    import os
    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)

    _target = MockTarget(bus)  # noqa: F841
    _mctpd = MockMctpd(bus)  # noqa: F841
    _em = MockEM(bus)  # noqa: F841

    _name_em = dbus.service.BusName("xyz.openbmc_project.EntityManager", bus)  # noqa: F841
    _name_mctp = dbus.service.BusName("xyz.openbmc_project.Mctp", bus)  # noqa: F841
    print("Mock mctpd and EM running...")
    sys.stdout.flush()
    loop.run()


if __name__ == '__main__':
    main()