Mock callable

This module implements infrastructure for mocking callbacks and other callables.

class MockCallable(return_value=None, called_timeout=5.0, not_called_timeout=1.0)

This class implements a mock callable.

It is useful for when you want to assert that a callable is called, but the callback is called asynchronously, so that you might have to wait a short time for the call to occur.

If you use a regular mock for the callback, your tests will end up littered with sleeps:

antenna_apiu_proxy.start_communicating()
communication_state_changed_callback.assert_called_once_with(
    CommunicationStatus.NOT_ESTABLISHED
)
time.sleep(0.1)
communication_state_changed_callback.assert_called_once_with(
    CommunicationStatus.ESTABLISHED
)

These sleeps waste time, slow down the tests, and they are difficult to tune: maybe you only need to sleep 0.1 seconds on your development machine, but what if the CI pipeline deploys the tests to an environment that needs 0.2 seconds for this?

This class solves that by putting each call to the callback onto a queue. Then, each time we assert that a callback was called, we get a call from the queue, waiting if necessary for the call to arrive, but with a timeout:

antenna_apiu_proxy.start_communicating()
communication_state_changed_callback.assert_next_call(
    CommunicationStatus.NOT_ESTABLISHED
)
communication_state_changed_callback.assert_next_call(
    CommunicationStatus.ESTABLISHED
)

__init__(return_value=None, called_timeout=5.0, not_called_timeout=1.0)

Initialise a new instance.

Parameters:

return_value (Optional[Any]) – what to return when called
called_timeout (float) – how long to wait for a call to occur when we are expecting one. It makes sense to wait a long time for the expected call, as it will generally arrive much much sooner anyhow, and failure for the call to arrive in time will cause the assertion to fail. The default is 5 seconds.
not_called_timeout (float) – how long to wait for a callback when we are not expecting one. Since we need to wait the full timeout period in order to determine that a callback has not arrived, asserting that a call has not been made can severely slow down your tests. By keeping this timeout quite short, we can speed up our tests, at the risk of prematurely passing an assertion. The default is 0.5

assert_last_call(*args, **kwargs)

Assert the arguments of the last call to this mock callback.

The “last” call is the last call before an attempt to get the next event times out.

This is useful for situations where we know a device may call a callback several time, and we don’t care too much about the exact order of calls, but we do know what the final call should be.

Parameters:

args (Any) – positional args that the call is asserted to have
kwargs (Any) – keyword args that the call is asserted to have

Raises:

AssertionError – if the callback has not been called.

Return type:

None

assert_next_call(*args, **kwargs)

Assert the arguments of the next call to this mock callback.

If the call has not been made, this method will wait up to the specified timeout for a call to arrive.

Parameters:

args (Any) – positional args that the call is asserted to have
kwargs (Any) – keyword args that the call is asserted to have

Raises:

AssertionError – if the callback has not been called.

Return type:

None

assert_not_called(timeout=None)

Assert that the callback still has not been called after the timeout period.

This is a slow method because it has to wait the full timeout period in order to determine that the call is not coming. An optional timeout parameter is provided for the situation where you are happy for the assertion to pass after a shorter wait time.

Parameters:: timeout (Optional[float]) – optional timeout for the check. If not provided, the default is the class setting
Return type:: None

calls_in_queue(expected_arguments_list)

Docstring.

Parameters:: expected_arguments_list (list[Any]) – A list of arguments this mock is expected to be called with and found in the queue.
Return type:: bool
Returns:: True if all arguments provided were found in the queue else returns False.

get_next_call()

Return the arguments of the next call to this mock callback.

This is useful for situations where you do not know exactly what the arguments of the next call will be, so you cannot use the assert_next_call() method. Instead you want to assert some specific properties on the arguments:

(args, kwargs) = mock_callback.get_next_call()
event_data = args[0].attr_value
assert event_data.name == "healthState"
assert event_data.value == HealthState.UNKNOWN
assert event_data.quality == tango.AttrQuality.ATTR_VALID

If the call has not been made, this method will wait up to the specified timeout for a call to arrive.

Raises:: AssertionError – if the callback has not been called
Return type:: tuple[Sequence[Any], dict[str, Any]]
Returns:: an (args, kwargs) tuple

get_whole_queue()

Return the arguments of all calls to this mock callback currently in the queue.

This is useful for situations where you do not know exactly what order the calls will happen but you do know what the arguments will be. Instead you want to assert that your call is somewhere in the queue.

If the call has not been made, this method will wait up to the specified timeout for a call to arrive.

Return type:: list[tuple[Sequence[Any], dict[str, Any]]]
Returns:: a list of (args, kwargs) tuple

class MockCallableDeque(return_value=None, called_timeout=5.0, not_called_timeout=1.0)

An extension to the MockCallable class to allow the queue to be interrogated.

This class alters MockCallable to use a deque instead of a queue and adds the assert_in_deque method which checks the deque for calls to this mock with specific arguments.

It is a special case of a MockCallable where the callable will be called in a non-deterministic order.

This class allows inspection of the deque to find specific calls.

__init__(return_value=None, called_timeout=5.0, not_called_timeout=1.0)

Initialise a new instance.

Parameters:

return_value (Optional[Any]) – what to return when called
called_timeout (float) – how long to wait for a call to occur when we are expecting one. It makes sense to wait a long time for the expected call, as it will generally arrive much much sooner anyhow, and failure for the call to arrive in time will cause the assertion to fail. The default is 5 seconds.
not_called_timeout (float) – how long to wait for a callback when we are not expecting one. Since we need to wait the full timeout period in order to determine that a callback has not arrived, asserting that a call has not been made can severely slow down your tests. By keeping this timeout quite short, we can speed up our tests, at the risk of prematurely passing an assertion. The default is 0.5

assert_all_in_deque(expected_arguments_list)

Assert multiple calls with arguments have been made to this mock.

Assert that a list of calls to the mocked callback with the expected arguments are present anywhere in the deque.

Parameters:: expected_arguments_list (list[Any]) – A list of arguments this mock is expected to be called with and found in the deque.
Return type:: None

assert_in_deque(expected_argument)

Assert a single call with argument has been made to this mock.

Assert that a single call to the callback with the expected argument is present in the deque.

Parameters:: expected_argument (Any) – An argument this mock is expected to be called with and found in the deque.
Raises:: AssertionError – if the expected argument was not found.
Return type:: None

assert_next_call_with_keys(expected_argument, fqdn=None)

Assert that the call to this mock with a given key also has the given value.

This method searches the deque for the next call to the mock with the specified key while ignoring other keys. If a match to the key is found then the value must also match. If the key is not found or the value does not match the expected value this method will raise an AssertionError.

Parameters:

expected_argument (dict[str, Any]) – A dict containing the key-value argument this mock is expected to be called with.
fqdn (Optional[str]) – fqdn to be searched for in the queue

Raises:

AssertionError – If the key is not found or the value does not match the expected value.

Return type:

None

assert_next_calls_with_keys(expected_arguments_list)

Assert that the calls to this mock with given keys also have the given values.

This method searches the deque for the next calls to the mock with the specified key while ignoring other keys. If a match to the key is found then the value must also match. If the key is not found or the value does not match the expected value this method will raise an AssertionError

Parameters:: expected_arguments_list (list[Union[dict[str, Any], tuple[dict[str, Any], str]]]) – A list of dicts containing the key-value arguments this mock is expected to be called with.
Return type:: None

assert_not_called_with_keys(*state_change_keys, fqdn=None)

Assert that this mock has not been called with the given key and fqdn.

Assert that no call to this mock has been made where its state_change argument has the given key(s) and its fqdn keyword- argument matches the specified fqdn.

Parameters:

state_change_keys (str) – state_change keys to be searched for in the queue
fqdn (Optional[str]) – fqdn to be searched for in the queue

Raises:

AssertionError – If a key is not found or a value does not match an expected value.

Return type:

None

assert_ordered_in_deque(expected_arguments_list)

Assert that the mock has been called with the provided arguments in order.

Parameters:: expected_arguments_list (list[Any]) – A list of ordered arguments this mock is expected to have been called with.
Raises:: AssertionError – if any argument is not found or they are in a different order.
Return type:: None

get_next_call_with_keys(*state_change_keys, fqdn=None)

Get the next state change with specific keys that this mock was called with.

This method searches the deque for the next call to this mock where the keys of its state_change argument match the specified keys, and the value of its fqdn keyword-argument match the specified fqdn. If a match is found, the corresponding call is removed from the deque, and the dictionary values of the state_change argument with matching keys is returned.

Parameters:

state_change_keys (str) – state_change keys to be searched for in the queue
fqdn (Optional[str]) – fqdn to be searched for in the queue

Return type:

Optional[tuple[Any]]

Returns:

tuple containing the values of the state_change dictionary with matching keys (or None)

class MockChangeEventCallback(event_name, called_timeout=5.0, not_called_timeout=0.5, filter_for_change=False)

This class implements a mock change event callback.

It is a special case of a MockCallable where the callable expects to be called with event_name, event_value and event_quality arguments (which is how ska_sat_lmc.device_proxy.SatDeviceProxy calls its change event callbacks).

__init__(event_name, called_timeout=5.0, not_called_timeout=0.5, filter_for_change=False)

Initialise a new instance.

Parameters:

event_name (str) – the name of the event for which this callable is a callback
called_timeout (float) – how long to wait for a call to occur when we are expecting one. It makes sense to wait a long time for the expected call, as it will generally arrive much much sooner anyhow, and failure for the call to arrive in time will cause the assertion to fail. The default is 5 seconds.
not_called_timeout (float) – how long to wait for a callback when we are not expecting one. Since we need to wait the full timeout period in order to determine that a callback has not arrived, asserting that a call has not been made can severely slow down your tests. By keeping this timeout quite short, we can speed up our tests, at the risk of prematurely passing an assertion. The default is 0.5
filter_for_change (bool) – filtered?

assert_last_change_event(value, _do_assert=True, quality=tango.AttrQuality.ATTR_VALID)

Assert the arguments of the last call to this mock callback.

The “last” call is the last call before an attempt to get the next event times out.

This is useful for situations where we know a device may fire several events, and we don’t know or care about the exact order of events, but we do know what the final event should be. For example, when we tell MccsController to turn on, it has to turn many devices on, which have to turn many devices on, etc. With so m

Parameters:

value (Any) – the asserted value of the change event
quality (AttrQuality) – the asserted quality of the change event. This is optional, with a default of ATTR_VALID.
_do_assert (bool) – option to not perform an assert (useful for debugging).

Raises:

AssertionError – if the callback has not been called.

Return type:

None

assert_next_change_event(value, quality=tango.AttrQuality.ATTR_VALID)

Assert the arguments of the next call to this mock callback.

If the call has not been made, this method will wait up to the specified timeout for a call to arrive.

Parameters:

value (Any) – the asserted value of the change event
quality (AttrQuality) – the asserted quality of the change event. This is optional, with a default of ATTR_VALID.

Raises:

AssertionError – if the callback has not been called.

Return type:

None

assert_not_called()

Assert if not called.

Raises:: AssertionError – change event callback
Return type:: None

get_next_change_event()

Return the attribute value in the next call to this mock change event callback.

This is useful for situations where you do not know exactly what the value will be, so you cannot use the assert_next_change_event() method. Instead you want to assert some specific properties on the arguments.

Raises:: AssertionError – if the callback has not been called
Return type:: Any
Returns:: an (args, kwargs) tuple