libmuscle.pytest.muscle_tester module

class libmuscle.pytest.muscle_tester.MuscleTester(run_dir: Path)

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

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

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.

libmuscle.pytest.muscle_tester.make_server_process(ymmsl_config: Configuration, run_dir: Path, start_instances: bool) → Generator[str, None, None]

libmuscle.pytest.muscle_tester.raise_error(*args: object) → None

libmuscle.pytest.muscle_tester.start_mmp_server(control_pipe: Tuple[Connection, Connection], ymmsl_config: Configuration, run_dir: RunDir, env: dict[str, str], start_instances: bool) → None