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 a MUSCLE3 Instance that acts as the “tester” component, which is connected to the implementation under test.

cleanup() None[source]

Clean up the tester instance.

Safe to call even if the instance was already shut down due to a timeout or deadlock error.

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 (without ‘receive_’ prefix).

  • slot – Optional slot number for vector ports.

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

Raises:

RuntimeError – If a deadlock is detected or the connection to the implementation was lost while waiting for a message.

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 (without ‘send_’ prefix).

  • message – The message to send.

  • slot – Optional slot number for vector ports.