Source code for ska_tango_base.software_bus.testing

#
# This file is part of the SKA Tango Base project
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE.txt for more info.
"""Testing helpers for software buses."""

import typing
from contextlib import contextmanager

from .. import type_hints
from . import SharingObserver
from ._bus import _SignalBus

__all__ = ["bus_test_context", "MockSignalObserver"]




@contextmanager
def bus_test_context(
    *observers: type_hints.ObserverProtocol,
) -> typing.Generator[type_hints.BusProtocol, None, None]:
    """
    Context manager that provides a temporary signal bus for testing.

    Automatically registers the provided observers before starting emission
    processing.  Any instances of :class:`.SharingObserver` will have their
    :attr:`~.SharingObserver.shared_bus` property assigned instead of being
    register directly.

    :param observers: The observers to be registered with the bus.
    :return: The signal bus instance.
    """
    bus = _SignalBus()

    for o in observers:
        if isinstance(o, SharingObserver):
            o.shared_bus = bus
        else:
            bus.register_observer(o)

    bus.start_thread()

    try:
        yield bus
    finally:
        bus.shutdown_thread()



try:
    import ska_tango_testing.mock as stt



    class MockSignalObserver(stt.MockCallableGroup):
        """
        Mock observer that allows asserting that specific signals were emitted.

        This is only available if ska_tango_testing can be imported.
        """



        def notify_emission(self, signal: str, value: typing.Any) -> None:
            """
            Notify the observer of an emission asynchronously.

            :param signal: The name of the signal emitted.
            :param value: The value emitted on the signal.
            """
            self[signal](value)


except ImportError:
    pass