libmuscle.pytest.implementation_tester module

class libmuscle.pytest.implementation_tester.ImplementationTester(default_timeout: float, muscle_manager_address: str, test_ymmsl_config: Configuration)[source]

Bases: object

The ImplementationTester creates the MUSCLE3 Instances that act as the “tester” components, which are connected to the implementation under test, the testee.

There are up to two tester components, both with only O_I and S ports: - The parent tester is connected to the F_INIT and O_F ports of the

testee, and acts as its parent (“macro”).

  • The sibling tester is connected to the O_I and S ports of the testee, and interacts with it as a peer on a matching timeline.

Sends and receives are forwarded to the component that has a port with the given name. Every time the parent tester starts a new round of messages to the testee’s F_INIT ports, it also sends a message to the sibling, so that the sibling’s reuse loop runs in step with that of the testee.

cleanup() → None[source]

Clean up the tester instances.

Safe to call even if the instances were already shut down after an error.

Raises:

RuntimeError – If the test did not finish its last round of messages.

receive(port_name: str, slot: int | None = None, *, timeout: float | None = None) → Message[source]

Receive a message from the specified port.

Parameters:

  • port_name – Name of the port to receive from.

  • slot – Optional slot number for vector ports.

  • timeout – Timeout in seconds. If None, uses default_timeout.

Raises:

  • ValueError – If the testee has no port with this name.

  • RuntimeError – If receiving is not allowed yet, a deadlock is detected, or the connection to the testee was lost.

On an error, all tester instances are shut down.

send(port_name: str, message: Message, slot: int | None = None) → None[source]

Send a message on the specified port.

Parameters:

  • port_name – Name of the port to send on.

  • message – The message to send.

  • slot – Optional slot number for vector ports.

Raises:

  • ValueError – If the testee has no port with this name.

  • RuntimeError – If the message cannot be sent, e.g. because the previous round of messages is not finished.

On an error, all tester instances are shut down.

libmuscle.pytest.implementation_tester.PARENT_SETTINGS_PORT_NAME = '__settings_in__'

Port of the parent tester connected to the testee’s muscle_settings_in.

libmuscle.pytest.implementation_tester.PARENT_TESTER_NAME = 'muscle3_tester_parent'

Tester connected to the F_INIT and O_F ports of the testee.

libmuscle.pytest.implementation_tester.SIBLING_SETTINGS_PORT_NAME = '__sibling_settings_out__'

Port of the parent tester connected to the sibling’s muscle_settings_in.

libmuscle.pytest.implementation_tester.SIBLING_TESTER_NAME = 'muscle3_tester_sibling'

Tester connected to the O_I and S ports of the testee.