libmuscle.pytest package

class libmuscle.pytest.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.

class libmuscle.pytest.MuscleTester(run_dir: Path)[source]

Bases: object

Helper class to test an implementation.

Note: You don’t need to construct a MuscleTester directly; use the muscle3_tester pytest fixture instead.

cleanup() → None[source]

Stop the manager process and clean up all resources.

Stops the ImplementationTester, restores the monkeypatched ReceiveTimeoutHandler.on_timeout(), and shuts down the manager subprocess.

start_implementation(ymmsl_source: str | Path, implementation: str, *, default_timeout: float = 60) → ImplementationTester[source]

Start a MUSCLE3 manager and return an ImplementationTester.

A tester component is added and connected to all ports of the implementation defined in the yMMSL source. A subprocess is started in which the MUSCLE3 manager runs, and its address is retrieved. A monkeypatch overwrites ReceiveTimeoutHandler.on_timeout() so that a RuntimeError is raised when a receive timeout is reached, causing the test simulation to quit. Finally, an ImplementationTester is created from the manager address and the generated test yMMSL configuration.

Parameters:

  • ymmsl_source – Either a string containing the yMMSL, or a pathlib.Path pointing to a file containing the yMMSL.

  • implementation – Name of the implementation to test.

  • default_timeout – Timeout (seconds) for message operations.

Returns:

An ImplementationTester connected to the running manager.

Raises:

RuntimeError – If the ImplementationTester could not be initialized, for example because the executable under test does not exist and never registered with the manager.

Submodules