jobflow_remote.testing package

Submodules

• jobflow_remote.testing.cli module
  • run_check_cli()

Module contents

A series of toy workflows that can be used for testing.

class jobflow_remote.testing.EnumMaker(e: jobflow_remote.testing.TestEnum = <TestEnum.A: 'A'>, name: str = 'enum maker')

Bases: Maker

e: TestEnum = 'A'

make()

Make a job or a flow - must be overridden with a concrete implementation.

name: str = 'enum maker'

class jobflow_remote.testing.MyShellIO(blocking=False, stdout_path='stdout', stderr_path='stderr', ps_username_strict=False, username_maxchars=None)

Bases: ShellIO

Construct the ShellIO object.

Notes

On Linux, the ‘ps’ command truncates usernames to 7-8 characters, appending a “+” if truncated. To use the full username, set this variable to the desired maximum length. The ‘user’ field in the ps output will then be displayed as “user:NN”, where NN is this value. Previously, the default for the max number of characters for the username was 32. On macOS, usernames are not truncated and there is no option to fix the field width, so the default is now None.

Parameters:

• blocking (bool) – Whether the execution should be blocking.

• stdout_path (str or Path) – Path to the standard output file.

• stderr_path (str or Path) – Path to the standard error file.

• ps_username_strict (bool) – If set to True, the parsing of the username will raise an error if the username is truncated.

• username_maxchars (int or None) – Width of the username field for the ps command. Default is None (which is 7-8 characters on Linux). On MacOS, setting the field explicitly is not allowed but the username is never truncated.

USERNAME_MAXCHARS = 1

header_template: str = 'exec > $${qout_path}\nexec 2> $${qerr_path}\n\necho $${job_name}\n$${qverbatim}\n'

class jobflow_remote.testing.TestEnum(value)

Bases: Enum

A = 'A'

B = 'B'

jobflow_remote.testing.add(a, b)

Adds two numbers together and writes the answer to a file.

jobflow_remote.testing.add_big(a: float, b: float)

Adds two numbers together and inflates the answer to a large list and tries to store that within the defined store.

jobflow_remote.testing.add_big_undefined_store(a: float, b: float)

Adds two numbers together and writes the answer to an artificially large file which is attempted to be stored in a undefined store.

jobflow_remote.testing.add_sleep(a, b)

Adds two numbers together and sleeps for “b” seconds.

jobflow_remote.testing.always_fails() → NoReturn

A job that always fails.

jobflow_remote.testing.arithmetic(a: float | list[float], b: float | list[float], op: Callable | None = None) → float | None

jobflow_remote.testing.check_env_var() → str

jobflow_remote.testing.create_detour(detour_job: Job)

Create a detour based on the passed Job.

jobflow_remote.testing.current_jobdoc()

jobflow_remote.testing.ignore_input(a: int) → int

Can receive an input, but ignores it.

Allows to test flows with failed parents

jobflow_remote.testing.no_resolve(ref)

A job that does not resolve the reference in input

jobflow_remote.testing.onmissing_none(args)

A job that can run even if references are missing

jobflow_remote.testing.replace_and_stop_children(a: int = 1, b: int = 1) → Response[None]

jobflow_remote.testing.replace_and_stop_jobflow(a=1, b=1) → Response[None]

jobflow_remote.testing.self_replace(n: int)

Create a replace Job with the same job n times.

jobflow_remote.testing.stop_jobflow(x=None) → Response[None]

jobflow_remote.testing.write_file(n) → None