Source code for blissoda.tests.conftest

import os
from collections import namedtuple
from contextlib import ExitStack

import mock
import pytest

from ..persistent import mock_redis
from ..persistent import ordereddict as ordereddict_mod
from ..persistent import parameters as parameters_mod
from ..version_utils import has_minimal_version




def has_redis_server() -> bool:
    # TODO: instead of pytest-redis which needs a real Redis server, use fakeredis?
    with os.popen("redis-server --version") as output:
        return bool(output.read())



MockedBlissInfo = namedtuple("MockedBlissInfo", "beacon_connection")

if has_redis_server():

    @pytest.fixture(scope="session")
    def mock_bliss_session(redis_proc):
        # Ensure we do not connect to a real Beacon host
        os.environ["BEACON_HOST"] = "nonexisting:25000"

        # Skip test when Beacon dependencies are not available
        # TODO: mock to the extend we no longer need bliss as a dependency
        if has_minimal_version("blissdata", "1"):
            try:
                from bliss.redis.manager import RedisAddress
            except ImportError:
                pytest.skip("requires bliss")
        else:
            from blissdata.redis import RedisAddress

        try:
            from bliss.config.conductor.client import get_default_connection
            from bliss.config.conductor.connection import Connection
        except ImportError:
            pytest.skip("requires bliss")

        redis_url = f"{redis_proc.host}:{redis_proc.port}"

        def _get_redis_url(self):
            """Return the pytest-redis server URL"""
            return RedisAddress.factory(redis_url)

        with ExitStack() as stack:
            # The Beacon connection should connect to the pytest-redis server
            ctx = mock.patch.object(
                Connection, "get_redis_connection_address", _get_redis_url
            )
            stack.enter_context(ctx)
            ctx = mock.patch.object(
                Connection, "get_redis_data_server_connection_address", _get_redis_url
            )
            stack.enter_context(ctx)

            yield MockedBlissInfo(get_default_connection())

else:



    @pytest.fixture(scope="session")
    def mock_bliss_session():
        # redis_proc raises an exception when redis-server is not available
        pytest.skip("requires redis-server")





@pytest.fixture
def mock_bliss(mock_bliss_session):
    yield
    proxy = mock_bliss_session.beacon_connection.get_redis_proxy()
    proxy.flushall()





@pytest.fixture
def mock_persistent():
    with mock.patch.object(
        parameters_mod, "HashObjSetting", mock_redis.LocalHashObjSetting
    ):
        with mock.patch.object(
            ordereddict_mod,
            "OrderedHashObjSetting",
            mock_redis.LocalOrderedHashObjSetting,
        ):
            try:
                mock_redis.LocalHashObjSetting._LOCAL_REASON = "unit test"
                mock_redis.LocalOrderedHashObjSetting._LOCAL_REASON = "unit test"
                mock_redis.LocalHashObjSetting._LOCAL_WARNING_EMITTED = False
                mock_redis.LocalOrderedHashObjSetting._LOCAL_WARNING_EMITTED = False

                yield mock_redis._REDIS_STORE
            finally:
                mock_redis._REDIS_STORE.clear()

                mock_redis.LocalHashObjSetting._LOCAL_REASON = None
                mock_redis.LocalOrderedHashObjSetting._LOCAL_REASON = None
                mock_redis.LocalHashObjSetting._LOCAL_WARNING_EMITTED = False
                mock_redis.LocalOrderedHashObjSetting._LOCAL_WARNING_EMITTED = False





@pytest.fixture
def mock_processing_triggers():
    enabled = {}

    def enable(processing_id, trigger, timing):
        enabled[processing_id] = (trigger, timing)

    def disable(processing_id):
        enabled.pop(processing_id, None)

    with mock.patch(
        "blissoda.utils.trigger.enable_processing_trigger",
        side_effect=enable,
        autospec=True,
    ), mock.patch(
        "blissoda.utils.trigger.disable_processing_trigger",
        side_effect=disable,
        autospec=True,
    ):
        yield enabled