# Source code for tango.test_context

    # SPDX-FileCopyrightText: All Contributors to the PyTango project
    # SPDX-License-Identifier: LGPL-3.0-or-later

    """Provide a context to run a device without a database."""

    # Imports
    import os
    import time
    import struct
    import socket
    import tempfile
    import traceback
    import collections
    import psutil
    from functools import partial

    # Concurrency imports
    import threading
    import multiprocessing
    import queue

    # CLI imports
    from ast import literal_eval
    from importlib import import_module
    from argparse import ArgumentParser, ArgumentTypeError

    # Local imports
    from tango.server import run
    from tango.utils import (
        _clear_test_context_tango_host_fqtrl,
        is_non_str_seq,
        _set_test_context_tango_host_fqtrl,
    )
    from tango.green import switch_existing_global_executors_to_thread
    from tango import DeviceProxy, Database, DevFailed, Util

    __all__ = (
        "MultiDeviceTestContext",
        "DeviceTestContext",
        "run_device_test_context",
        "get_server_port_via_pid",
    )

    # Helpers

    _DEFAULT_THREAD_TIMEOUT = 5.0
    _DEFAULT_PROCESS_TIMEOUT = 7.0

    IOR = collections.namedtuple(
        "IOR",
        "first dtype_length dtype nb_profile tag "
        "length major minor wtf host_length host port body",
    )

    NO_DB_FRAGMENT = "dbase=no"

    def ascii_to_bytes(s):
        convert = lambda x: bytes((int(x, 16),))
        return b"".join(convert(s[i : i + 2]) for i in range(0, len(s), 2))

    def parse_ior(encoded_ior):
        assert encoded_ior[:4] == "IOR:"
        ior = ascii_to_bytes(encoded_ior[4:])
        dtype_length = struct.unpack_from("II", ior)[-1]
        form = f"II{dtype_length:d}sIIIBBHI"
        host_length = struct.unpack_from(form, ior)[-1]
        form = f"II{dtype_length:d}sIIIBBHI{host_length:d}sH0I"
        values = struct.unpack_from(form, ior)
        values += (ior[struct.calcsize(form) :],)
        strip = lambda x: x[:-1] if isinstance(x, bytes) else x
        return IOR(*map(strip, values))

    def get_server_host_port():
        util = Util.instance()
        ds = util.get_dserver_device()
        encoded_ior = util.get_dserver_ior(ds)
        ior = parse_ior(encoded_ior)
        return ior.host.decode(), ior.port

    def get_server_port_via_pid(pid, host, retries=400, delay=0.03):
        """Return the TCP port that a device server process is listening on (GIOP).

        This checks TCP sockets open on the process with the given PID, and attempts
        to find the one that accepts GIOP traffic.  A connection will be made to each
        listening socket, and data may be sent to them.

        General Inter-ORB Protocol (GIOP) is the message protocol which object
        request brokers (ORBs) communicate in CORBA.  This port is the one that is
        used when connecting a DeviceProxy.  These are not the port(s) used for ZMQ
        event traffic.

        :param pid: operating system process identifier
        :type pid: int
        :param host: hostname/IP that device server is listening on.  E.g., 127.0.0.1,
        IP address of a non-loopback network interface, etc.  Note that starting a device
        server on "localhost" may fail if OmniORB creates an IPv6-only socket.
        :type pid: str
        :param retries: number of times to retry attempts, optional
        :type retries: int
        :param delay: time to wait (seconds) between retries, optional
        :type delay: float

        :returns: TCP port number
        :rtype: int

        :raises RuntimeError: If the GIOP port couldn't be identified

        .. versionadded:: 9.4.0
        .. versionadded:: 9.5.0
            *retries* parameter.
            *delay* parameter.
        """

        count = 0
        port = None
        last_err = None
        while port is None and count < retries:
            ports = _get_listening_tcp_ports(pid)
            try:
                port = _get_giop_port(host, ports)
            except Exception as err:
                last_err = err
                time.sleep(delay)
            count += 1

        if port is None:
            raise RuntimeError(
                f"Failed to get GIOP TCP port within {count * delay:.1f} sec"
            ) from last_err

        return port

    def _get_listening_tcp_ports(pid):
        p = psutil.Process(pid)
        if hasattr(p, "net_connections"):
            conns = p.net_connections(kind="tcp")
        else:
            conns = p.connections(kind="tcp")  # deprecated in psutil v6.0.0
        return list(set([c.laddr.port for c in conns if c.status == "LISTEN"]))

    def _get_giop_port(host, ports):
        protocols = _try_get_protocols_on_ports(host, ports)
        for port, protocol in protocols.items():
            if protocol == "GIOP":
                return port
        raise RuntimeError(
            f"None of ports {ports} appear to have GIOP protocol. "
            f"Guessed protocols: {protocols}."
        )

    def _try_get_protocols_on_ports(host, ports):
        """Return a dict with port to protocol mapping.

        This attempts to establish a TCP socket connection to the host
        for each port, and then determine the protocol it supports.

        ZMQ client sockets receive an unsolicited version check message on connection.
        CORBA GIOP client sockets don't receive an unsolicited message, so we send
        a requested to disconnect, and expect an empty message back.
        """
        zmq_response = (
            b"\xff\x00\x00\x00\x00\x00\x00\x00\x01\x7f"  # signature + version check
        )
        giop_send = b"GIOP\x01\x02\x01\x05\x00\x00\x00\x00"  # request disconnect
        giop_response = b""  # graceful disconnect
        max_bytes_expected = len(zmq_response)

        protocols = dict.fromkeys(ports, "Unknown")
        for port in ports:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            try:
                sock.settimeout(0.001)
                server_address = (host, port)
                sock.connect(server_address)

                try:
                    data = sock.recv(max_bytes_expected)
                    if data == zmq_response:
                        protocols[port] = "ZMQ"
                        continue
                except OSError:
                    pass

                try:
                    sock.sendall(giop_send)
                    data = sock.recv(max_bytes_expected)
                    if data == giop_response:
                        protocols[port] = "GIOP"
                        continue
                except OSError:
                    pass
            except OSError:
                pass
            finally:
                sock.close()
        return protocols

    def literal_dict(arg):
        return dict(literal_eval(arg))

    def device(path):
        """Get the device class from a given module."""
        module_name, device_name = path.rsplit(".", 1)
        try:
            module = import_module(module_name)
        except Exception:
            raise ArgumentTypeError(
                f"Error importing {module_name}.{device_name}:\n"
                f"{traceback.format_exc()}"
            )
        return getattr(module, device_name)

    def get_host_ip():
        """Get the primary external host IP.

        This is used to be useful because an explicit IP was required to get
        tango events to work properly.  Now remains for backwards compatibility.
        """
        return "127.0.0.1"

    def _device_class_from_field(field):
        """
        Helper function that extracts and return a device class from a
        'class' field of a devices_info dictionary

        :param field: the field from which to extract the device class

        :return: the device class extracted from the field
        """
        device_cls_class = None
        if is_non_str_seq(field):
            (device_cls_class, device_class) = (field[0], field[1])
            if isinstance(device_cls_class, str):
                device_cls_class = device(device_cls_class)
        else:
            device_class = field
        if isinstance(device_class, str):
            device_class = device(device_class)
        return (device_cls_class, device_class)

    class MultiDeviceTestContext:
        """Context to run device(s) without a database.

        The difference with respect to
        :class:`~tango.test_context.DeviceTestContext` is that it allows
        to export multiple devices (even of different Tango classes).

        Example usage::

            from tango import DeviceProxy
            from tango.server import Device, attribute
            from tango.test_context import MultiDeviceTestContext

            class Device1(Device):
                @attribute(dtype=int)
                def attr1(self):
                    return 1

            class Device2(Device):
                @attribute(dtype=int)
                def attr2(self):
                    dev1 = DeviceProxy("test/device/1")
                    return dev1.attr1 * 2

            devices_info = (
                {
                    "class": Device1,
                    "devices": [
                        {"name": "test/device/1"},
                    ],
                },
                {
                    "class": Device2,
                    "devices": [
                        {
                            "name": "test/device/2",
                        },
                    ],
                },
            )

            def test_devices():
                with MultiDeviceTestContext(devices_info, process=True) as context:
                    proxy1 = context.get_device("test/device/1")
                    proxy2 = context.get_device("test/device/2")
                    assert proxy1.attr1 == 1
                    assert proxy2.attr2 == 2

        :param devices_info:
          a sequence of dicts with information about
          devices to be exported. Each dict consists of the following keys:

            * "class" which value is either of:

              * a :class:`~tango.server.Device` or the name of some such class
              * a sequence of two elements, the first element being a
                :class:`~tango.DeviceClass` or the name of some such class,
                the second element being a :class:`~tango.DeviceImpl` or the
                name of some such class

            * "devices" which value is a sequence of dicts with the following keys:

              * "name" (str)
              * "properties" (dict)
              * "memorized" (dict)
              * "root_atts" (dict)"

        :type devices_info:
          sequence<dict>
        :param server_name:
          Name to use for the device server.
          Optional.  Default is the first device's class name.
        :type server_name:
          :py:obj:`str`
        :param instance_name:
          Name to use for the device server instance.
          Optional.  Default is lower-case version of the server name.
        :type instance_name:
          :py:obj:`str`
        :param db:
          Path to a pre-populated text file to use for the
          database.
          Optional.  Default is to create a new temporary file and populate it
          based on the devices and properties supplied in `devices_info`.
        :type db:
          :py:obj:`str`
        :param host:
          Hostname to use for device server's ORB endpoint.
          Optional.  Default is the loopback IP address, 127.0.0.1.
        :type host:
          :py:obj:`str`
        :param port:
          Port number to use for the device server's ORB endpoint.
          Optional.  Default is chosen by omniORB.
        :type port:
          :py:obj:`int`
        :param debug:
          Debug level for the device server logging.
          0=OFF, 1=FATAL, 2=ERROR, 3=WARN, 4=INFO, 5=DEBUG.
          Optional. Default is warn.
        :type debug:
          :py:obj:`int`
        :param process:
          True if the device server should be launched in a new process, otherwise
          use a new thread.  Note:  if the context will be used mutiple times, it
          may seg fault if the thread mode is chosen.
          See the :ref:`issues <testing-approaches-issues>` and
          :ref:`process kwarg <testing-approaches-process-kwarg>` discussion in the docs.
          Optional.  Default is thread.
        :type process:
          :py:obj:`bool`
        :param daemon:
          True if the new thread/process must be created in daemon mode.
          Optional.  Default is not daemon.
        :type daemon:
          :py:obj:`bool`
        :param timeout:
          How long to wait (seconds) for the device server to start up, and also
          how long to wait on joining the thread/process when stopping.
          Optional.  Default differs for thread and process modes.
        :type timeout:
          :py:obj:`float`
        :param green_mode:
          Green mode to use for the device server.
          Optional.  Default uses the Device specification (via green_mode class attribute),
          or if that isn't specified the global green mode.
        :type green_mode:
          :obj:`~tango.GreenMode`

        .. versionadded:: 9.3.2

        .. versionadded:: 9.3.3
            Added support for `memorized` key to "devices" field in `devices_info`.
            Added support for literal names for "class" field in `devices_info`.

        .. versionadded:: 9.3.3
            added *green_mode* parameter.

        .. versionchanged:: 9.5.0
            By default, devices launched by a test context can be accessed using
            short names with :class:`~tango.AttributeProxy`, :class:`~tango.DeviceProxy`,
            and :class:`~tango.Group`.
            This can be disabled by setting the `enable_test_context_tango_host_override`
            class/instance attribute to `False` before starting the test context.
            Added support for `root_atts` key to "devices" field in `devices_info`.
        """

        command = "{0} {1} -ORBendPoint giop:tcp:{2}:{3} -file={4}"
        enable_test_context_tango_host_override = True

        thread_timeout = _DEFAULT_THREAD_TIMEOUT
        process_timeout = _DEFAULT_PROCESS_TIMEOUT

        def __init__(
            self,
            devices_info,
            server_name=None,
            instance_name=None,
            db=None,
            host=None,
            port=0,
            debug=3,
            process=False,
            daemon=False,
            timeout=None,
            green_mode=None,
        ):
            if not server_name:
                _, first_device = _device_class_from_field(devices_info[0]["class"])
                server_name = first_device.__name__
            if not instance_name:
                instance_name = server_name.lower()
            if db is None:
                handle, db = tempfile.mkstemp()
                self.handle = handle
            else:
                self.handle = None
            if host is None:
                host = "127.0.0.1"  # note: localhost does not currently work
            if timeout is None:
                timeout = self.process_timeout if process else self.thread_timeout
            # Attributes
            self.db = db
            self.host = host
            self.port = port
            self.timeout = timeout
            self.server_name = "/".join(("dserver", server_name, instance_name))
            if process:
                self._startup_exception_queue = multiprocessing.Queue()
                self._discovered_port_queue = multiprocessing.JoinableQueue()
            else:
                self._startup_exception_queue = queue.Queue()
                self._discovered_port_queue = queue.Queue()
            self._startup_exception = None
            self._devices = {}
            self._saved_environ = {}
            self._process = process

            # Command args
            string = self.command.format(server_name, instance_name, host, port, db)
            string += f" -v{debug}" if debug else ""
            cmd_args = string.split()

            class_list = []
            device_list = []
            tangoclass_list = []
            for device_info in devices_info:
                device_cls, device = _device_class_from_field(device_info["class"])
                tangoclass = device.__name__
                if tangoclass in tangoclass_list:
                    self.delete_db()
                    raise ValueError(
                        "multiple entries in devices_info pointing "
                        "to the same Tango class"
                    )
                tangoclass_list.append(tangoclass)
                # File
                self.append_db_file(
                    server_name, instance_name, tangoclass, device_info["devices"]
                )
                if device_cls:
                    class_list.append((device_cls, device, tangoclass))
                else:
                    device_list.append(device)

            # Target and arguments
            if class_list and device_list:
                self.delete_db()
                raise ValueError(
                    "mixing HLAPI and classical API in devices_info is not supported"
                )
            if class_list:
                runserver = partial(run, class_list, cmd_args, green_mode=green_mode)
            elif len(device_list) == 1 and hasattr(device_list[0], "run_server"):
                runserver = partial(device.run_server, cmd_args, green_mode=green_mode)
            elif device_list:
                runserver = partial(run, device_list, cmd_args, green_mode=green_mode)
            else:
                raise ValueError("Wrong format of devices_info")

            cls = multiprocessing.Process if process else threading.Thread
            self.thread = cls(target=self.target, args=(runserver, process))
            self.thread.daemon = daemon

        def target(self, runserver, process=False):
            threading.current_thread().name += " TestContext DS launcher"
            try:
                runserver(
                    pre_init_callback=self.pre_init,
                    post_init_callback=self.post_init,
                    raises=True,
                )
            except Exception as exc:
                self._startup_exception_queue.put(exc)
                self._discovered_port_queue.put(-1)  # don't block queue reader
            finally:
                # Put something in the queue just in case
                exc = RuntimeError("The server failed to report anything")
                self._startup_exception_queue.put(exc)
                # Make sure the process has enough time to send the items
                # because it might segfault while cleaning up the tango resources
                if process:
                    time.sleep(0.1)

        def pre_init(self):
            try:
                if not self.port:
                    util = Util.instance()
                    pid = util.get_pid()
                    self.port = get_server_port_via_pid(pid, self.host)
                if self._process:
                    # set TANGO_HOST override for device server (DS) launcher child process
                    self._override_test_context_tango_host()
                # report it to the test context process (DS launcher's parent)
                self._discovered_port_queue.put(self.port)
                # and wait till it has been processed
                self._discovered_port_queue.join()
            except Exception as exc:
                self._startup_exception_queue.put(exc)
                self._discovered_port_queue.put(-1)  # don't block queue reader

        def _wait_until_port_is_known(self):
            try:
                self.port = self._discovered_port_queue.get(timeout=self.timeout)
            except queue.Empty:
                raise RuntimeError(
                    "GIOP TCP port of TextContext device server not available during startup"
                )

            # set the TANGO_HOST override for the process that created the test context
            self._override_test_context_tango_host()
            # notify the server process, so it can continue
            self._discovered_port_queue.task_done()

        def _override_test_context_tango_host(self):
            if self.enable_test_context_tango_host_override:
                _set_test_context_tango_host_fqtrl(
                    f"tango://{self.host}:{self.port}#{NO_DB_FRAGMENT}"
                )

        def post_init(self):
            # now we can connect to the device - success!
            self._startup_exception_queue.put(None)

        def append_db_file(self, server, instance, tangoclass, device_prop_info):
            """Generate a database file corresponding to the given arguments."""
            device_names = [info["name"] for info in device_prop_info]
            # Open the file
            with open(self.db, "a") as f:
                f.write("/".join((server, instance, "DEVICE", tangoclass)))
                f.write(": ")
                f.write(", ".join(device_names))
                f.write("\n")
            # Create database
            db = Database(self.db)
            # Write properties
            for info in device_prop_info:
                device_name = info["name"]
                properties = dict(info.get("properties", {}))
                # Patch the property dict to avoid a PyTango bug
                for key, value in properties.items():
                    if is_non_str_seq(value):
                        properties[key] = [v if v != "" else " " for v in value]
                    else:
                        properties[key] = value if value != "" else " "
                db.put_device_property(device_name, properties)

                root_atts = info.get("root_atts", {})
                properties_to_save = {
                    attribute_name: {"__root_att": root_att}
                    for (attribute_name, root_att) in root_atts.items()
                }

                memorized = info.get("memorized", {})
                for attribute_name, memorized_value in memorized.items():
                    if attribute_name in properties_to_save:
                        properties_to_save[attribute_name]["__value"] = memorized_value
                    else:
                        properties_to_save[attribute_name] = {"__value": memorized_value}

                db.put_device_attribute_property(device_name, properties_to_save)
            try:
                validated_db = Database(self.db)
            except DevFailed:
                with open(self.db, "r") as f:
                    content = f.read()
                raise RuntimeError(
                    f"Invalid FileDatabase file was created.\n"
                    f"Check device properties (empty list or str?): "
                    f"{device_prop_info}.\n"
                    f"Problematic file has content:\n{content}"
                )
            return validated_db

        def delete_db(self):
            """delete temporary database file only if it was created by this class"""
            if self.handle is not None:
                os.close(self.handle)
                os.unlink(self.db)

        def get_server_access(self) -> str:
            """Return the full server name."""
            return f"tango://{self.host}:{self.port}/{self.server_name}#{NO_DB_FRAGMENT}"

        def get_device_access(self, device_name) -> str:
            """Return the full device name."""
            return f"tango://{self.host}:{self.port}/{device_name}#{NO_DB_FRAGMENT}"

        def get_device(self, device_name) -> DeviceProxy:
            """Return the device proxy corresponding to the given device name.

            Maintains previously accessed device proxies in a cache to not recreate
            then on every access.
            """
            if device_name not in self._devices:
                device = DeviceProxy(self.get_device_access(device_name))
                self._devices[device_name] = device
            return self._devices[device_name]

        def start(self):
            """Run the server."""
            self._set_up_environment_variables()
            self.thread.start()
            self.connect()
            return self

        def connect(self):
            try:
                self._wait_until_port_is_known()
                self._wait_until_startup_status_is_known()
            except RuntimeError:
                if self.thread.is_alive():
                    raise RuntimeError(
                        "The server appears to be stuck at initialization. "
                        "Check stdout/stderr for more information."
                    )
                elif hasattr(self.thread, "exitcode"):
                    raise RuntimeError(
                        f"The server process stopped with exitcode {self.thread.exitcode}. "
                        f"Check stdout/stderr for more information."
                    )
                else:
                    raise RuntimeError(
                        "The server stopped without reporting. "
                        "Check stdout/stderr for more information."
                    )

            if self._startup_exception:
                raise self._startup_exception

            # Get server proxy
            self.server = DeviceProxy(self.get_server_access())
            self.server.ping()

            # If server has green_mode, we should switch global executor to the main thread
            if not self._process:
                switch_existing_global_executors_to_thread()

        def _wait_until_startup_status_is_known(self):
            try:
                self._startup_exception = self._startup_exception_queue.get(
                    timeout=self.timeout
                )
            except queue.Empty:
                raise RuntimeError(
                    f"Timed-out waiting for device server startup ({self.timeout:1.3f} sec)"
                )

        def stop(self):
            """Kill the server."""
            try:
                if self.server:
                    self.server.command_inout("Kill")
                self.join(self.timeout)
            finally:
                _clear_test_context_tango_host_fqtrl()
                self.delete_db()
                self._restore_environment_variables()

        def join(self, timeout=None):
            self.thread.join(timeout)

        def _set_up_environment_variables(self):
            self._saved_environ = dict(os.environ)
            # Patch bug #819
            if self._process:
                os.environ["ORBscanGranularity"] = "0"

        def _restore_environment_variables(self):
            modified_environ = dict(os.environ)
            saved_environ = self._saved_environ
            for key, value in modified_environ.items():
                if key in saved_environ:
                    saved_value = saved_environ[key]
                    if value != saved_value:
                        os.environ[key] = saved_value
                else:
                    os.environ.pop(key)  # unset environment var

        def __enter__(self) -> "MultiDeviceTestContext":
            """Enter method for context support.

            :return:
              Instance of this test context.  Use `get_device` to get proxy
              access to any of the devices started by this context.
            :rtype:
              :class:`~tango.test_context.MultiDeviceTestContext`
            """
            if not self.thread.is_alive():
                self.start()
            return self

        def __exit__(self, exc_type, exception, trace):
            """Exit method for context support."""
            self.stop()
            return False

    # Single device test context

    class DeviceTestContext(MultiDeviceTestContext):
        """Context to run a single device without a database.

        The difference with respect to
        :class:`~tango.test_context.MultiDeviceTestContext` is that it only
        allows to export a single device.

        Example usage::

            from time import sleep

            from tango.server import Device, attribute, command
            from tango.test_context import DeviceTestContext

            class PowerSupply(Device):

                @attribute(dtype=float)
                def voltage(self):
                    return 1.23

                @command
                def calibrate(self):
                    sleep(0.1)

            def test_calibrate():
                '''Test device calibration and voltage reading.'''
                with DeviceTestContext(PowerSupply, process=True) as proxy:
                    proxy.calibrate()
                    assert proxy.voltage == 1.23

        :param device:
          Device class to be run.
        :type device:
          :class:`~tango.server.Device` or :class:`~tango.DeviceImpl`
        :param device_cls:
          The device class can be provided if using the low-level API.
          Optional.  Not required for high-level API devices, of type
          :class:`~tango.server.Device`.
        :type device_cls:
          :class:`~tango.DeviceClass`

         The rest of the parameters are described in
         :class:`~tango.test_context.MultiDeviceTestContext`.

        .. versionadded:: 9.2.1

        .. versionadded:: 9.3.3
            added *memorized* parameter.

        .. versionadded:: 9.3.6
            added *green_mode* parameter.
        """

        def __init__(
            self,
            device,
            device_cls=None,
            server_name=None,
            instance_name=None,
            device_name=None,
            properties=None,
            db=None,
            host=None,
            port=0,
            debug=3,
            process=False,
            daemon=False,
            timeout=None,
            memorized=None,
            root_atts=None,
            green_mode=None,
        ):
            # Argument
            if not server_name:
                server_name = device.__name__
            if not instance_name:
                instance_name = server_name.lower()
            if not device_name:
                device_name = "test/nodb/" + server_name.lower()
            if properties is None:
                properties = {}
            if memorized is None:
                memorized = {}
            if root_atts is None:
                root_atts = {}
            if device_cls:
                cls = (device_cls, device)
            else:
                cls = device
            devices_info = (
                {
                    "class": cls,
                    "devices": (
                        {
                            "name": device_name,
                            "properties": properties,
                            "memorized": memorized,
                            "root_atts": root_atts,
                        },
                    ),
                },
            )
            super().__init__(
                devices_info,
                server_name=server_name,
                instance_name=instance_name,
                db=db,
                host=host,
                port=port,
                debug=debug,
                process=process,
                daemon=daemon,
                timeout=timeout,
                green_mode=green_mode,
            )

            self.device_name = device_name
            self.device = self.server = None

        def get_device_access(self, device_name=None) -> str:
            """Return the full device name."""
            if device_name is None:
                device_name = self.device_name
            return super().get_device_access(device_name)

        def connect(self):
            super().connect()
            # Get device proxy
            self.device = self.get_device(self.device_name)
            self.device.ping()

        def __enter__(self) -> DeviceProxy:
            """Enter method for context support.

            :return:
              A device proxy to the device started by this context.
            :rtype:
              :class:`~tango.DeviceProxy`
            """
            if not self.thread.is_alive():
                self.start()
            return self.device

    # Command line interface

    def parse_command_line_args(args=None):
        """Parse arguments given in command line."""
        desc = "Run a given device on a given port."
        parser = ArgumentParser(description=desc)
        # Add arguments
        msg = "The device to run as a python path."
        parser.add_argument("device", metavar="DEVICE", type=device, help=msg)
        msg = "The hostname to use."
        parser.add_argument(
            "--host", metavar="HOST", type=str, help=msg, default=get_host_ip()
        )
        msg = "The port to use."
        parser.add_argument("--port", metavar="PORT", type=int, help=msg, default=8888)
        msg = "The debug level."
        parser.add_argument("--debug", metavar="DEBUG", type=int, help=msg, default=3)
        msg = "The properties to set as python dict."
        parser.add_argument(
            "--prop", metavar="PROP", type=literal_dict, help=msg, default="{}"
        )
        msg = "Whether to disable short-name lookup for devices launched by test context"
        parser.add_argument(
            "--disable-short-name-lookup",
            dest="disable_short_name_lookup",
            action="store_true",
            help=msg,
        )
        # Parse arguments
        namespace = parser.parse_args(args)
        return (
            namespace.device,
            namespace.host,
            namespace.port,
            namespace.prop,
            namespace.debug,
            namespace.disable_short_name_lookup,
        )

    def run_device_test_context(args=None):
        (
            device_,
            host,
            port,
            properties,
            debug,
            disable_short_name_lookup,
        ) = parse_command_line_args(args)
        context = DeviceTestContext(
            device_, properties=properties, host=host, port=port, debug=debug
        )
        context.enable_test_context_tango_host_override = not disable_short_name_lookup
        context.start()
        msg = f"{device_.__name__} started on port {context.port} with properties {properties}"
        print(msg)
        print(f"Device access: {context.get_device_access()}")
        print(f"Server access: {context.get_server_access()}")
        context.join()
        print("Done")

    # Main execution

    if __name__ == "__main__":
        run_device_test_context()