Source code for ska_ser_skallop.subscribing.configuration

import inspect
from contextlib import contextmanager
from typing import NamedTuple

import mock

from ska_ser_skallop.utils.singleton import Singleton

from .base import MessageBoardBase
from .message_board import MessageBoard


def determine_messageboard() -> MessageBoardBase:
    # default Messageboard TODO add option for determining messageboard from env file
    return MessageBoard()


def get_messageboard(init_before_provided=True):
    container = MessageboardContainer()  # type: ignore
    board = container.get_message_board()
    if init_before_provided:
        board.__init__()
    return container.get_message_board()


def set_messageboard(board: MessageBoardBase, provided_by: str):
    container = MessageboardContainer()  # type: ignore
    container.inject(board, provided_by)


def reset_configurations():
    container = MessageboardContainer()  # type: ignore
    container.reset()


@contextmanager
def patch_messageboard(board: MessageBoardBase = None, provided_by: str = ""):
    if not provided_by:
        parent = inspect.getouterframes(inspect.currentframe())[2][3]
        provided_by = f"configured from {parent}"
    if board is None:
        board = mock.MagicMock(MessageBoardBase)
        # makes board iterable
        board.get_items.return_value = []
    container = MessageboardContainer()  # type: ignore
    original_board_provided = container.get_current_provided_messageboard()
    container.reset()
    container.inject(board, provided_by)
    yield board
    container.reset()
    if original_board_provided:
        container.inject(
            original_board_provided.provider,
            original_board_provided.provided_by,
        )


class Provider(NamedTuple):
    provider: MessageBoardBase
    provided_by: str


class MessageboardContainer(metaclass=Singleton):
    def __init__(self):
        self._message_board = None

    def get_current_provided_messageboard(self):
        return self._message_board

    def get_message_board(self) -> MessageBoardBase:
        """retrieves the message board for use in the app, if this is the first time
        call has been made, the object shall first determine a default messageboard
        """
        if self._message_board:
            return self._message_board.provider
        board = determine_messageboard()
        self.inject(board, "environment during dependency call")
        return board

    def reset(self):
        self.__init__()

    def inject(self, board: MessageBoardBase, provided_by: str):
        assert not self._message_board, (
            f"A message_board ({type(self._message_board.provider)}) has already been "
            f"defined, provided by: {self._message_board.provided_by}"
        )
        self._message_board = Provider(board, provided_by)