Testing

Testing SOCS Agent code is critical to ensuring code functions properly, especially when making changes after initial creation of the Agent. Testing within SOCS comes in two forms, unit tests and integration tests.

Unit tests test functionality of the Agent code directly, without running the Agent itself (or any supporting parts, such as the crossbar server, or a piece of hardware to connect to.)

Integration tests run a small OCS network, starting up the crossbar server, and likely require some code emulating the piece of hardware the OCS Agent is built to communicate with. Integration tests are more involved than unit tests, requiring more setup and thus taking longer to execute than unit tests. Both are important for fully testing the functionality of your Agent.

Running Tests

We use pytest as the test runner for SOCS. To run all of the tests, from with in the socs/tests/ directory, run pytest:

$ python3 -m pytest --cov

This will run every test, both unit and integration tests. Integration tests depend on mocked up versions of the hardware the agents in question interface with. Testing against actual hardware has previously been done in the hardware/ directory. These hardware tests will not automatically be run.

Test Selection

You typically will not want to run all tests all the time. There are many ways to limit which tests run. Here are some examples.

Run only one test file:

$ python3 -m pytest --cov socs agents/test_ls372_agent.py

Run tests based on test name(s):

$ python3 -m pytest --cov -k 'test_ls372_init_lakeshore_task'

Note that this will match to the beginning of the test names, so the above will match ‘test_ls372_init_lakeshore_task’ as well as ‘test_ls372_init_lakeshore_task_already_initialzed’.

Custom Markers

There are some custom markers for tests in SOCS, ‘integtest’ for integration tests, and ‘spt3g’ for tests dependent on spt3g_software. These markers can be used to select or deselect tests.

To run only the unit tests run:

$ python3 -m pytest --cov -m 'not integtest'

To run only the integration tests:

$ python3 -m pytest --cov -m 'integtest'

Note

The integration tests depend on ‘–cov’ being used, so all examples here throw that flag. You could omit it when running unit tests, but it’s often useful to view coverage results anyway.

You can view the available markers with:

$ python3 -m pytest --markers
@pytest.mark.integtest: marks tests as integration test (deselect with '-m "not integtest"')
@pytest.mark.spt3g: marks tests that depend on spt3g (deselect with '-m "not spt3g"')

Code Coverage

Code coverage measures how much of the code the tests cover. This is typically reported as a percentage of lines executed during testing. Coverage is measured with Coverage.py.

To obtain code coverage:

$ python3 -m pytest --cov --cov-report=html

You can then view the coverage report in the htmlcov/ directory. Coverage for SOCS is also automatically reported to coveralls.io.

Testing Against Hardware

Warning

Testing against hardware is not well supported at this time. Tests early on in development were initially built against actual hardware, but have since been neglected.

If you are designing and running tests against actual hardware you should store and run them directly from the hardware directory, which is configured to be ignored by the automatic test discovery in pytest.

When running against hardware, call only the test you’d like to run. For instance, to test just the Lakeshore 372:

$ cd hardware/
$ python3 -m pytest test_ls372.py

Writing Tests

When writing integration tests for an Agent we need to mock the communication with the hardware the Agent interfaces with. We can do that with the DeviceEmulator. socs provides a function for creating pytest fixtures that yield a Device Emulator. Here’s an example of how to use it:

# define the fixture, including any responses needed for Agent startup
emulator = create_device_emulator({'*IDN?': 'LSCI,MODEL425,4250022,1.0'}, relay_type='serial')

@pytest.mark.integtest
def test_ls425_operational_status(wait_for_crossbar, emulator, run_agent, client):
    # define the responses needed for this test
    responses = {'OPST?': '001'}
    emulator.define_responses(responses)

    # run task that sends the 'OPST?' command
    resp = client.operational_status()
    assert resp.status == ocs.OK
    assert resp.session['op_code'] == OpCode.SUCCEEDED.value

In the example, we initialize the emulator with any commands and responses needed for Agent startup, and specify that the Agent communicates via serial. Then, within the test, we define commands and their responses needed during testing. In this case that is the OPST? command and the expected response of 001. Once these are saved with socs.testing.device_emulator.DeviceEmulator.define_responses() we can run our Agent Tasks/Processes for testing.

API

socs.testing.device_emulator.create_device_emulator(responses, relay_type, port=9001, encoding='utf-8', reconnect=False)[source]

Create a device emulator fixture.

This provides a device emulator that can be used to mock a device during testing.

Parameters:

responses (dict) – Dictionary with commands as keys, and responses as values. See DeviceEmulator for details.
relay_type (str) – Communication relay type. Either ‘serial’ or ‘tcp’.
port (int) – Port for the TCP relay to listen for connections on. Defaults to 9001. Only used if relay_type is ‘tcp’.
encoding (str) – Encoding for the messages and responses. See socs.testing.device_emulator.DeviceEmulator() for more details.
reconnect (bool) – If True, on TCP client disconnect, the emulator will listen for new incoming connections instead of quitting

Returns:

A pytest fixture that creates a Device emulator of the specified type.

Return type:

function

class socs.testing.device_emulator.DeviceEmulator(responses, encoding='utf-8', reconnect=False)[source]

A mocked device which knows how to respond on various communication channels.

Parameters:

responses (dict) – Initial responses, any response required by Agent startup, if any.
encoding (str) – Encoding for the messages and responses. DeviceEmulator will try to encode and decode messages with the given encoding. No encoding is used if set to None. That can be useful if you need to use raw data from your hardware. Defaults to ‘utf-8’.
reconnect (bool) – If True, on TCP client disconnect, the emulator will listen for new incoming connections instead of quitting

responses

Current set of responses the DeviceEmulator would give. Should all be strings, not bytes-like.

Type:: dict

default_response

Default response to send if a command is unrecognized. No response is sent and an error message is logged if a command is unrecognized and the default response is set to None. Defaults to None.

Type:: str

encoding

Encoding for the messages and responses, set by the encoding argument.

Type:: str

_type

Relay type, either ‘serial’ or ‘tcp’.

Type:: str

_read

Used to stop the background reading of data recieved on the relay.

Type:: bool

_conn

TCP connection for use in ‘tcp’ relay.

Type:: socket.socket

create_serial_relay()[source]

Create the serial relay, emulating a hardware device connected over serial.

This first uses socat to setup a relay. It then connects to the “internal” end of the relay, ready to receive communications sent to the “responder” end of the relay. This end of the relay is located at ./responder. You will need to configure your Agent to use that path for communication.

Next it creates a thread to read commands sent to the serial relay in the background. This allows responses to be defined within a test using DeviceEmulator.define_responses() after instantiation of the DeviceEmulator object within a given test.

get_response(msg)[source]

Determine the response to a given message.

Parameters:

msg (str) – Command string to get the response for.

Returns:

Response string. Will return None if a valid response is not: found.

Return type:

str

shutdown()[source]: Shutdown communication on the configured relay. This will stop any attempt to read communication on the relay, as well as shutdown the relay itself.

create_tcp_relay(port)[source]

Create the TCP relay, emulating a hardware device connected over TCP.

Creates a thread to read commands sent to the TCP relay in the background. This allows responses to be defined within a test using DeviceEmulator.define_responses() after instantiation of the DeviceEmulator object within a given test.

Parameters:: port (int) – Port for the TCP relay to listen for connections on.

Notes

This will not return until the socket is properly bound to the given port. If this setup is not working it is likely another device emulator instance is not yet finished or has not been properly shutdown.

update_responses(responses: Dict)[source]

Updates the current responses. See define_responses for more detail.

Parameters:: responses (dict) – Dict of commands to use to update the current responses.

define_responses(responses, default_response=None)[source]

Define what responses are available to reply with on the configured communication relay.

Parameters:

responses (dict) – Dictionary of commands: response. Values can be a list, in which case the responses in the list are popped and given in order until depleted.
default_response (str) – Default response to send if a command is unrecognized. No response is sent and an error message is logged if a command is unrecognized and the default response is set to None. Defaults to None.

Examples

The given responses might look like:

>>> responses = {'KRDG? 1': '+1.7E+03'}
>>> responses = {'*IDN?': 'LSCI,MODEL425,4250022,1.0',
                 'RDGFIELD?': ['+1.0E-01', '+1.2E-01', '+1.4E-01']}

Notes

The DeviceEmulator will handle encoding/decoding. The responses defined should all be strings, not bytes-like, unless you set encoding=None.