Skip to content
Snippets Groups Projects
Select Git revision
  • 69b28076db7c16e668e5545f2d5233e3698dafa7
  • main default protected
  • make_oidc_aud_configurable
  • 187_fix_agnpy
  • helm-chart
  • esap-general#159
  • nico-reviewing-async
  • feature/update-uws-auth
  • SDC-596/fix-internal-server-no-datasets
  • fix-query-errors
  • feature/ida-dekstop-records
  • merge-master
  • feature/better_logs_and_mocks
  • adex-settings-to-configuration
  • astron-vo-quick-fix
  • 69_add_diracIAM
  • sdc380-aladin-cone-search
  • esap-gateway-query
  • esap-general#11
  • esap-general#51
  • esap-api-dirac
  • esap-gateway-release-0_2_0
  • esap-gateway-release_0_1_1
  • esap-gateway-release_0_1_0
24 results

query_controller.py

Blame
  • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    conftest.py 8.33 KiB
    """This module defines elements of the pytest test harness shared by all tests."""
    import logging
    import socket
    from queue import Empty, Queue
    
    import pytest
    import tango
    from tango import EventType
    from tango.test_context import get_host_ip, DeviceTestContext, MultiDeviceTestContext
    
    
    @pytest.fixture(scope="class")
    def device_properties():
        """
        Fixture that returns device_properties to be provided to the device under test.
    
        This is a default implementiong that provides no properties.
        """
        return {}
    
    
    @pytest.fixture()
    def tango_context(device_test_config):
        """Return a Tango test context object, in which the device under test is running."""
        component_manager_patch = device_test_config.pop("component_manager_patch", None)
        if component_manager_patch is not None:
            device_test_config["device"].create_component_manager = component_manager_patch
    
        tango_context = DeviceTestContext(**device_test_config)
        tango_context.start()
        yield tango_context
        tango_context.stop()
    
    
    @pytest.fixture()
    def device_under_test(tango_context):
        """
        Return a device proxy to the device under test.
    
        :param tango_context: a Tango test context with the specified device
            running
        :type tango_context: :py:class:`tango.DeviceTestContext`
    
        :return: a proxy to the device under test
        :rtype: :py:class:`tango.DeviceProxy`
        """
        return tango_context.device
    
    
    def pytest_itemcollected(item):
        """Make Tango-related tests run in forked mode."""
        if "device_under_test" in item.fixturenames:
            item.add_marker("forked")
    
    
    @pytest.fixture(scope="function")
    def tango_change_event_helper(device_under_test):
        """
        Return a helper for testing tango change events.
    
        To use it, call the subscribe method with the name of the attribute
        for which you want change events. The returned value will be a
        callback handler that you can interrogate with
        ``assert_not_called``, ``assert_call``, ``assert_calls``, and
        ``value`` methods.
    
        .. code-block:: py
    
            state_callback = tango_change_event_helper.subscribe("state")
            state_callback.assert_call(DevState.OFF)
    
            # Check that we can't turn off a device that isn't on
            with pytest.raises(DevFailed):
                device_under_test.Off()
            state_callback.assert_not_called()
    
            # Now turn it on and check that we can turn it off
            device_under_test.On()
            state_callback.assert_call(DevState.ON)
    
            # Or we can test a sequence of events
            device_under_test.Off()
            device_under_test.On()
            state_callback.assert_calls([DevState.OFF, DevState.ON])
    
        :param device_under_test: a :py:class:`tango.DeviceProxy` to the
            device under test, running in a
            :py:class:`tango.test_context.DeviceTestContext`.
        :type device_under_test: :py:class:`tango.DeviceProxy`
        """
    
        class _Callback:
            """
            Private callback handler class.
    
            An instance is returned by the tango_change_event_helper each
            time it is used to subscribe to a change event.
            """
    
            @staticmethod
            def subscribe(attribute_name):
                """
                Return an instance that is subscribed to change events on a named attribute.
    
                :param attribute_name: name of the attribute for which
                    change events will be subscribed
                :type attribute_name: str
                :return: an event subscriber helper object
                :rtype: object
                """
                return _Callback(attribute_name)
    
            def __init__(self, attribute_name):
                """
                Initialise a new instance.
    
                The instance will be subscribed to change events on the
                named attribute.
    
                :param attribute_name: name of the attribute for which
                    change events will be subscribed
                :type attribute_name: str
                """
                self._value = None
                self._values_queue = Queue()
                self._errors = []
    
                # Subscription will result in an immediate
                # synchronous callback with the current value,
                # so keep this as the last step in __init__.
                self._id = device_under_test.subscribe_event(
                    attribute_name, EventType.CHANGE_EVENT, self
                )
    
            def __del__(self):
                """Unsubscribe from events before object is destroyed."""
                if hasattr(self, "_id"):
                    device_under_test.unsubscribe_event(self._id)
    
            def __call__(self, event_data):
                """
                Event subscription callback.
    
                :param event_data: data about the change events
                :type event_data: :py:class:`tango.EventData`
                """
                if event_data.err:
                    error = event_data.errors[0]
                    self._errors.append(
                        "Event callback error: [%s] %s" % (error.reason, error.desc)
                    )
                else:
                    self._values_queue.put(event_data.attr_value.value)
    
            def _next(self):
                """
                Get the attribute value from the next event.
    
                A value is returned if there is already one,  or if it
                arrives in time.
    
                :return: the attribute value reported in next change event,
                    or None if there is no event
                :rtype: same as attribute type
                """
                assert not self._errors, f"Some errors: {self._errors}"
                try:
                    return self._values_queue.get(timeout=1.5)
                except Empty:
                    return None
    
            def assert_not_called(self):
                """
                Assert that there are no new callbacks calls.
    
                (That is, there are no callback calls that have not already
                been consumed by an ``assert_call`` or ``assert_calls``.)
                """
                assert self._values_queue.empty()
    
            def assert_call(self, value):
                """
                Assert a call that has been made on this callback.
    
                Specifically, asserts that this callback has been called
                with a change event that updates the attribute value to a
                given value.
    
                Note that this method consumes a single event, but may leave
                other events unconsumed.
    
                :param value: the value that the attribute is asserted to
                    have been changed to
                :type value: same as the attribute type
                """
                assert self._next() == value
    
            def assert_calls(self, values):
                """
                Assert a sequence of calls that have been made on this callback.
    
                Specifically, assert that this callback has been called with
                a sequence of change events that update the attribute values
                to the given sequence of values.
    
                Note that this method consumes the events associated with
                the given values, but may leave subsequent events
                unconsumed.
    
                :param values: sequence of values that the attribute
                    is asserted to have been changed to
                :type values: list
                """
                for value in values:
                    self.assert_call(value)
    
        yield _Callback
    
    
    @pytest.fixture()
    def logger():
        """Fixture that returns a default logger for tests."""
        return logging.Logger("Test logger")
    
    
    @pytest.fixture(scope="module")
    def devices_to_test(request):
        """Fixture for devices to test."""
        yield getattr(request.module, "devices_to_test")
    
    
    @pytest.fixture(scope="function")
    def multi_device_tango_context(
        mocker, devices_to_test  # pylint: disable=redefined-outer-name
    ):
        """
        Create and return a TANGO MultiDeviceTestContext object.
    
        tango.DeviceProxy patched to work around a name-resolving issue.
        """
    
        def _get_open_port():
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.bind(("", 0))
            s.listen(1)
            port = s.getsockname()[1]
            s.close()
            return port
    
        HOST = get_host_ip()
        PORT = _get_open_port()
        _DeviceProxy = tango.DeviceProxy
        mocker.patch(
            "tango.DeviceProxy",
            wraps=lambda fqdn, *args, **kwargs: _DeviceProxy(
                "tango://{0}:{1}/{2}#dbase=no".format(HOST, PORT, fqdn), *args, **kwargs
            ),
        )
        with MultiDeviceTestContext(
            devices_to_test, host=HOST, port=PORT, process=True
        ) as context:
            yield context