cis_interface.communication.tests package¶

Submodules¶

cis_interface.communication.tests.test_AsciiFileComm module¶

class cis_interface.communication.tests.test_AsciiFileComm.TestAsciiFileComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_FileComm.TestFileComm

Test for AsciiFileComm communication class.

attr_list = ['name', 'address', 'direction', 'serializer', 'recv_timeout', 'close_on_eof_recv', 'opp_address', 'opp_comms', 'maxMsgSize', 'fd', 'read_meth', 'append', 'in_temp', 'open_as_binary', 'newline', 'is_series', 'platform_newline', 'comment']¶

comm = 'AsciiFileComm'¶

test_send_recv_comment()[source]¶

Test send/recv with commented message.

cis_interface.communication.tests.test_AsciiMapComm module¶

class cis_interface.communication.tests.test_AsciiMapComm.TestAsciiMapComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_FileComm.TestFileComm

Test for AsciiMapComm communication class.

comm = 'AsciiMapComm'¶

cis_interface.communication.tests.test_AsciiTableComm module¶

class cis_interface.communication.tests.test_AsciiTableComm.TestAsciiTableComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_AsciiFileComm.TestAsciiFileComm

Test for AsciiTableComm communication class.

comm = 'AsciiTableComm'¶

test_send_recv_comment()[source]¶

Disabled: Test send/recv with commented message.

class cis_interface.communication.tests.test_AsciiTableComm.TestAsciiTableComm_AsArray(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_AsciiTableComm.TestAsciiTableComm

Test for AsciiTableComm communication class.

testing_option_kws = {'as_array': True}¶

cis_interface.communication.tests.test_AsciiTableComm.test_AsciiTableComm_nofmt()[source]¶

Test read of asciitable without format.

cis_interface.communication.tests.test_AsyncComm module¶

class cis_interface.communication.tests.test_AsyncComm.TestAsyncComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_CommBase.TestCommBase

Tests for AsyncComm communication class.

attr_list = ['name', 'address', 'direction', 'serializer', 'recv_timeout', 'close_on_eof_recv', 'opp_address', 'opp_comms', 'maxMsgSize', 'dont_backlog', 'backlog_send_ready', 'backlog_recv_ready']¶

comm = 'AsyncComm'¶

test_send_recv_after_close()[source]¶

Test that send/recv after close returns false.

test_send_recv_direct()[source]¶

Test send/recv direct.

cis_interface.communication.tests.test_CommBase module¶

class cis_interface.communication.tests.test_CommBase.TestCommBase(*args, **kwargs)[source]¶

Bases: cis_interface.tests.CisTestClassInfo

Tests for CommBase communication class.

send_inst_kwargs¶

Keyword arguments for send half of the comm pair.

Type:	dict

assert_msg_equal(x, y)[source]¶

Assert that two messages are equivalent.

attr_list = ['name', 'address', 'direction', 'serializer', 'recv_timeout', 'close_on_eof_recv', 'opp_address', 'opp_comms', 'maxMsgSize']¶

cleanup_comm_classes¶

Comm classes that should be cleaned up following the test.

Type:	list

cls¶

Communication class.

Type:	str

comm = 'CommBase'¶

create_instance()[source]¶

Create a new instance of the class.

do_send_recv(send_meth='send', recv_meth='recv', msg_send=None, msg_recv=None, n_msg_send_meth='n_msg_send', n_msg_recv_meth='n_msg_recv', reverse_comms=False, send_kwargs=None, recv_kwargs=None, n_send=1, n_recv=1, close_on_send_eof=None, close_on_recv_eof=None)[source]¶

Generic send/recv of a message.

get_fresh_error_instance(recv=False)[source]¶

Get comm instance with ErrorClass parent class.

inst_args¶

Arguments for tested class.

Type:	list

inst_kwargs¶

Keyword arguments for tested class.

Type:	dict

is_installed¶

Is the communication class installed.

Type:	bool

map_sent2recv(obj)[source]¶

Convert a sent object into a received one.

maxMsgSize¶

Maximum message size.

Type:	int

mod¶

Absolute module import.

Type:	str

msg_long¶

Small test message for sending.

Type:	str

name¶

Name of the test connection.

Type:	str

recv_instance¶

Alias for instance.

remove_instance(inst)[source]¶

Remove an instance.

send_inst_kwargs

Keyword arguments for send instance.

Type:	dict

setup(*args, **kwargs)[source]¶

Initialize comm object pair.

teardown(*args, **kwargs)[source]¶

Destroy comm object pair.

test_attributes()[source]¶

Assert that the instance has all of the required attributes.

test_drain_messages()[source]¶

Test waiting for messages to drain.

test_empty_obj_recv()[source]¶

Test identification of empty message.

test_eof()[source]¶

Test send/recv of EOF message.

test_eof_no_close()[source]¶

Test send/recv of EOF message with no close.

test_error_name()[source]¶

Test error on missing address.

test_error_recv()[source]¶

Test error on recv.

test_error_send()[source]¶

Test error on send.

test_invalid_direction()[source]¶

Check that error raised for invalid direction.

test_msg¶

Test message that should be used for any send/recv tests.

Type:	str

test_msg_array¶

Test message that should be used for any send/recv tests.

Type:	str

test_purge(nrecv=1)[source]¶

Test purging messages from the comm.

test_recv_nomsg()[source]¶

Test recieve when there is no waiting message.

test_send_recv()[source]¶

Test send/recv of a small message.

test_send_recv_after_close()[source]¶

Test that opening twice dosn’t cause errors and that send/recv after close returns false.

test_send_recv_array()[source]¶

Test send/recv of a array message.

test_send_recv_dict()[source]¶

Test send/recv message as dict.

test_send_recv_nolimit()[source]¶

Test send/recv of a large message.

test_work_comm()[source]¶

Test creating/removing a work comm.

cis_interface.communication.tests.test_CommBase.test_registry()[source]¶

Test registry of comm.

cis_interface.communication.tests.test_FileComm module¶

class cis_interface.communication.tests.test_FileComm.TestFileComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_CommBase.TestCommBase

Test for FileComm communication class.

attr_list = ['name', 'address', 'direction', 'serializer', 'recv_timeout', 'close_on_eof_recv', 'opp_address', 'opp_comms', 'maxMsgSize', 'fd', 'read_meth', 'append', 'in_temp', 'open_as_binary', 'newline', 'is_series', 'platform_newline']¶

comm = 'FileComm'¶

send_inst_kwargs¶

Keyword arguments for send instance.

Type:	dict

teardown()[source]¶

Remove the file.

test_append()[source]¶

Test open of file comm with append.

test_invalid_read_meth()[source]¶

Test raise of error on invalid read_meth.

test_recv_nomsg()[source]¶

Test recieve when there is no waiting message.

test_remaining_bytes()[source]¶

Test remaining_bytes.

test_send_recv_nolimit()[source]¶

Disabled: Test send/recv of a large message.

test_series()[source]¶

Test sending/receiving to/from a series of files.

test_work_comm()[source]¶

Disabled: Test creating/removing a work comm.

class cis_interface.communication.tests.test_FileComm.TestFileComm_ascii(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_FileComm.TestFileComm

Test for FileComm communication class with open_as_binary = False.

testing_option_kws = {'open_as_binary': False}¶

class cis_interface.communication.tests.test_FileComm.TestFileComm_readline(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_FileComm.TestFileComm

Test for FileComm communication class with read_meth = ‘readline’.

inst_kwargs¶

Keyword arguments for tested class.

Type:	dict

testing_options¶

Testing options.

Type:	dict

cis_interface.communication.tests.test_FileComm.test_wait_for_creation()[source]¶

Test FileComm waiting for creation.

cis_interface.communication.tests.test_ForkComm module¶

class cis_interface.communication.tests.test_ForkComm.TestForkComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_CommBase.TestCommBase

Tests for ForkComm communication class.

attr_list = ['name', 'address', 'direction', 'serializer', 'recv_timeout', 'close_on_eof_recv', 'opp_address', 'opp_comms', 'maxMsgSize', 'comm_list', 'curr_comm_index']¶

cleanup_comm_classes¶

Comm classes that should be cleaned up following the test.

Type:	list

comm = 'ForkComm'¶

do_send_recv(*args, **kwargs)[source]¶

Generic send/recv of a message.

ncomm = 2¶

send_inst_kwargs¶

Keyword arguments for send instance.

Type:	dict

test_error_name()[source]¶

Test error on missing address.

test_error_recv()[source]¶

Disabled: Test error on recv.

test_error_send()[source]¶

Disabled: Test error on send.

test_purge(**kwargs)[source]¶

Test purging messages from the comm.

test_work_comm()[source]¶

Disabled: Test creating/removing a work comm.

class cis_interface.communication.tests.test_ForkComm.TestForkCommList(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_ForkComm.TestForkComm

Tests for ForkComm communication class with construction from address.

inst_kwargs¶

Keyword arguments for tested class.

Type:	list

cis_interface.communication.tests.test_IPCComm module¶

class cis_interface.communication.tests.test_IPCComm.TestIPCComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_AsyncComm.TestAsyncComm

Test for IPCComm communication class.

attr_list = ['name', 'address', 'direction', 'serializer', 'recv_timeout', 'close_on_eof_recv', 'opp_address', 'opp_comms', 'maxMsgSize', 'dont_backlog', 'backlog_send_ready', 'backlog_recv_ready', 'q']¶

comm = 'IPCComm'¶

cis_interface.communication.tests.test_IPCComm.test_ipc_queues()[source]¶

Test list of ipc queues.

cis_interface.communication.tests.test_IPCComm.test_ipcrm()[source]¶

Test removal of ipc objects.

cis_interface.communication.tests.test_IPCComm.test_ipcrm_not_isntalled()[source]¶

Test ipcrm if IPC library is not installed.

cis_interface.communication.tests.test_IPCComm.test_ipcrm_queues()[source]¶

Test removal of ipc queues.

cis_interface.communication.tests.test_IPCComm.test_ipcrm_queues_not_isntalled()[source]¶

Test ipcrm_queues if IPC library is not installed.

cis_interface.communication.tests.test_IPCComm.test_ipcs()[source]¶

Test list of ipc objects.

cis_interface.communication.tests.test_IPCComm.test_ipcs_not_isntalled()[source]¶

Test return of ipcs if IPC library is not installed.

cis_interface.communication.tests.test_IPCComm.test_not_running()[source]¶

Test raise of an error if a IPC library is not installed.

cis_interface.communication.tests.test_IPCComm.test_queue()[source]¶

Test creation/removal of queue.

cis_interface.communication.tests.test_IPCComm.test_queue_not_installed()[source]¶

Test return of get_queue if IPC library is not installed.

cis_interface.communication.tests.test_JSONFileComm module¶

class cis_interface.communication.tests.test_JSONFileComm.TestJSONFileComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_FileComm.TestFileComm

Test for JSONFileComm communication class.

comm = 'JSONFileComm'¶

cis_interface.communication.tests.test_MatFileComm module¶

class cis_interface.communication.tests.test_MatFileComm.TestMatFileComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_FileComm.TestFileComm

Test for MatFileComm communication class.

comm = 'MatFileComm'¶

cis_interface.communication.tests.test_ObjFileComm module¶

class cis_interface.communication.tests.test_ObjFileComm.TestObjFileComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_PlyFileComm.TestPlyFileComm

Test for ObjFileComm communication class.

comm = 'ObjFileComm'¶

cis_interface.communication.tests.test_PandasFileComm module¶

class cis_interface.communication.tests.test_PandasFileComm.TestPandasFileComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_AsciiTableComm.TestAsciiTableComm

Test for PandasFileComm communication class.

comm = 'PandasFileComm'¶

class cis_interface.communication.tests.test_PandasFileComm.TestPandasFileComm_nonames(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_PandasFileComm.TestPandasFileComm

Test for PandasFileComm communication class without field names sent.

testing_option_kws = {'no_names': True}¶

class cis_interface.communication.tests.test_PandasFileComm.TestPandasFileComm_single(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_PandasFileComm.TestPandasFileComm

Test for PandasFileComm communication class with field names sent.

get_options()[source]¶

Get testing options.

test_send_dict_default()[source]¶

Test automated conversion of dictionary to pandas data frame.

cis_interface.communication.tests.test_PickleFileComm module¶

class cis_interface.communication.tests.test_PickleFileComm.TestPickleFileComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_FileComm.TestFileComm

Test for PickleFileComm communication class.

comm = 'PickleFileComm'¶

cis_interface.communication.tests.test_PlyFileComm module¶

class cis_interface.communication.tests.test_PlyFileComm.TestPlyFileComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_FileComm.TestFileComm

Test for PlyFileComm communication class.

comm = 'PlyFileComm'¶

cis_interface.communication.tests.test_RMQAsyncComm module¶

class cis_interface.communication.tests.test_RMQAsyncComm.TestRMQAsyncComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_RMQComm.TestRMQComm

Test for RMQAsyncComm communication class.

attr_list = ['name', 'address', 'direction', 'serializer', 'recv_timeout', 'close_on_eof_recv', 'opp_address', 'opp_comms', 'maxMsgSize', 'dont_backlog', 'backlog_send_ready', 'backlog_recv_ready', 'connection', 'channel', 'times_connected', 'rmq_thread', 'rmq_lock']¶

comm = 'RMQAsyncComm'¶

test_reconnect()[source]¶

Test reconnect after unexpected disconnect.

cis_interface.communication.tests.test_RMQAsyncComm.test_not_running()[source]¶

Test raise of an error if a RMQ server is not running.

cis_interface.communication.tests.test_RMQComm module¶

class cis_interface.communication.tests.test_RMQComm.TestRMQComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_AsyncComm.TestAsyncComm

Test for RMQComm communication class.

attr_list = ['name', 'address', 'direction', 'serializer', 'recv_timeout', 'close_on_eof_recv', 'opp_address', 'opp_comms', 'maxMsgSize', 'dont_backlog', 'backlog_send_ready', 'backlog_recv_ready', 'connection', 'channel']¶

comm = 'RMQComm'¶

timeout = 10.0¶

cis_interface.communication.tests.test_RMQComm.test_not_running()[source]¶

Test raise of an error if a RMQ server is not running.

cis_interface.communication.tests.test_ServerComm module¶

class cis_interface.communication.tests.test_ServerComm.TestServerComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_CommBase.TestCommBase

Tests for ServerComm communication class.

attr_list = ['name', 'address', 'direction', 'serializer', 'recv_timeout', 'close_on_eof_recv', 'opp_address', 'opp_comms', 'maxMsgSize', 'response_kwargs', 'icomm', 'ocomm']¶

comm = 'ServerComm'¶

send_inst_kwargs¶

Keyword arguments for send instance.

Type:	dict

test_call()[source]¶

Test RPC call.

test_call_alias()[source]¶

Test RPC call aliases.

test_call_nolimit()[source]¶

Test RPC nolimit call.

test_close_in_thread()[source]¶

Test close of comm in thread.

test_eof_no_close()[source]¶

Test send/recv of EOF message with no close.

test_error_recv()[source]¶

Disabled: Test error on recv.

test_error_send()[source]¶

Disabled: Test error on send.

test_invalid_direction()[source]¶

Disabled: Test of error on incorrect direction.

test_newcomm_server()[source]¶

Test creation of server using newcomm.

test_work_comm()[source]¶

Disabled: Test creating/removing a work comm.

cis_interface.communication.tests.test_YAMLFileComm module¶

class cis_interface.communication.tests.test_YAMLFileComm.TestYAMLFileComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_FileComm.TestFileComm

Test for YAMLFileComm communication class.

comm = 'YAMLFileComm'¶

cis_interface.communication.tests.test_ZMQComm module¶

class cis_interface.communication.tests.test_ZMQComm.TestZMQComm(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_AsyncComm.TestAsyncComm

Test for ZMQComm communication class.

attr_list = ['name', 'address', 'direction', 'serializer', 'recv_timeout', 'close_on_eof_recv', 'opp_address', 'opp_comms', 'maxMsgSize', 'dont_backlog', 'backlog_send_ready', 'backlog_recv_ready', 'context', 'socket', 'socket_type_name', 'socket_type', 'protocol', 'host', 'port']¶

comm = 'ZMQComm'¶

description_prefix¶

String prefix to prepend docstr test message with.

protocol = None¶

send_inst_kwargs¶

Keyword arguments for send instance.

socket_type = None¶

test_eof_no_close()[source]¶

Test send/recv of EOF message with no close.

test_send_recv_nolimit()[source]¶

Send/recv of large message.

class cis_interface.communication.tests.test_ZMQComm.TestZMQCommINPROC(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_ZMQComm.TestZMQComm

Test for ZMQComm communication class with INPROC socket.

protocol = 'inproc'¶

class cis_interface.communication.tests.test_ZMQComm.TestZMQCommIPC(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_ZMQComm.TestZMQComm

Test for ZMQComm communication class with IPC socket.

protocol = 'ipc'¶

class cis_interface.communication.tests.test_ZMQComm.TestZMQCommIPC_client(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_ZMQComm.TestZMQComm_client, cis_interface.communication.tests.test_ZMQComm.TestZMQCommIPC

Test for ZMQComm communication class with IPC socket.

class cis_interface.communication.tests.test_ZMQComm.TestZMQCommPAIR(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_ZMQComm.TestZMQComm

Test for ZMQComm communication class with PAIR/PAIR socket.

socket_type = 'PAIR'¶

class cis_interface.communication.tests.test_ZMQComm.TestZMQCommPUB(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_ZMQComm.TestZMQComm

Test for ZMQComm communication class with PUB/SUB socket.

socket_type = 'PUB'¶

class cis_interface.communication.tests.test_ZMQComm.TestZMQCommPUSH(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_ZMQComm.TestZMQComm

Test for ZMQComm communication class with PUSH/PULL socket.

socket_type = 'PUSH'¶

class cis_interface.communication.tests.test_ZMQComm.TestZMQCommPUSH_INPROC(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_ZMQComm.TestZMQCommINPROC

Test for ZMQComm communication class with INPROC PUSH/PULL socket.

socket_type = 'PUSH'¶

class cis_interface.communication.tests.test_ZMQComm.TestZMQCommREQ(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_ZMQComm.TestZMQComm

Test for ZMQComm communication class with REP/REQ socket.

socket_type = 'REQ'¶

class cis_interface.communication.tests.test_ZMQComm.TestZMQCommROUTER(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_ZMQComm.TestZMQComm

Test for ZMQComm communication class with DEALER/ROUTER socket.

setup(*args, **kwargs)[source]¶

Initialize comm object pair with sleep after setup to ensure dealer has connected.

socket_type = 'ROUTER'¶

test_router_recv()[source]¶

Test router receipt of message from the dealer with an identity.

class cis_interface.communication.tests.test_ZMQComm.TestZMQCommTCP(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_ZMQComm.TestZMQComm

Test for ZMQComm communication class with TCP socket.

protocol = 'tcp'¶

class cis_interface.communication.tests.test_ZMQComm.TestZMQComm_client(*args, **kwargs)[source]¶

Bases: cis_interface.communication.tests.test_ZMQComm.TestZMQComm

Test for ZMQComm communication class for client/server.

send_inst_kwargs¶

Keyword arguments for send instance.

cis_interface.communication.tests.test_ZMQComm.test_error_on_send_open_twice()[source]¶

Test creation of the same send socket twice for an error.

cis_interface.communication.tests.test_ZMQComm.test_format_address()[source]¶

Test format/parse of address.

cis_interface.communication.tests.test_ZMQComm.test_get_socket_type_mate()[source]¶

Test socket type matching.

cis_interface.communication.tests.test_ZMQComm.test_invalid_protocol()[source]¶

Test raise of an error in the event of an invalid protocol.

Module contents¶