The Test Utilities API

tango.test_context.get_server_port_via_pid(pid, host, retries=400, delay=0.03)[source]

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.

Parameters:

  • pid (int) – operating system process identifier

  • host (str) – 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.

  • retries (int) – number of times to retry attempts, optional

  • delay (float) – time to wait (seconds) between retries, optional

Returns:

TCP port number

Return type:

int

Raises:

RuntimeError – If the GIOP port couldn’t be identified

Added in version 9.4.0.

Added in version 9.5.0: retries parameter. delay parameter.

tango.test_utils.wait_for_proxy(dev_name: str, proxy_class: type[~tango._tango.DeviceProxy] = <class 'tango._tango.DeviceProxy'>, retries: int = 600, delay: float = 0.02) → DeviceProxy

Create a new DeviceProxy, retrying until it is responding.

The DeviceProxy will try to ping the device and read the state at least once.

Parameters:

  • dev_name – device name for DeviceProxy connection string

  • proxy_class – Type of DeviceProxy class to instantiate (could be synchronous, asyncio, or gevent versions of the class). If using asyncio, then provide: partial(tango.asyncio.DeviceProxy, wait=True).

  • retries – number of times to retry attempts, optional

  • delay – time to wait (seconds) between retries, optional

Returns:

DeviceProxy object created to access the specified device

Raises:

RuntimeError – If the device is not responding before the timeout

Added in version 10.1.0.

tango.test_utils.wait_for_nodb_proxy_via_pid(pid: int, host: str, dev_name: str, proxy_class: type[DeviceProxy], retries: int = 600, delay: float = 0.02) → DeviceProxy

Create new DeviceProxy, with retrying until it is ready and responding.

The PID is used to get the process information and probe it to find the correct TCP port number to connect to. Creation of the DeviceProxy is retried until the device is responding, or times out. The DeviceProxy will ping the device and read the state.

Parameters:

  • pid – operating system process identifier

  • host – hostname/IP that device server is listening on.

  • dev_name – device name for DeviceProxy connection string

  • proxy_class – Type of DeviceProxy class to instantiate (could be synchronous, asyncio, or gevent versions of the class). If using asyncio, then provide: partial(tango.asyncio.DeviceProxy, wait=True).

  • retries – number of times to retry attempts, optional

  • delay – time to wait (seconds) between retries, optional

Returns:

DeviceProxy object created to access the specified device

Raises:

RuntimeError – If the GIOP port couldn’t be identified, or proxy timed out

See also

tango.test_context.get_server_port_via_pid()

Added in version 10.1.0.

Device Test Context API

See Device Test Context Classes API for details of the classes that can be used to launch a device server for testing.