fm_weck package

Subpackages

• fm_weck.grpc_service package
  • Subpackages
    • fm_weck.grpc_service.proto package
      • Submodules
      • fm_weck.grpc_service.proto.fm_weck_service_pb2 module
      • fm_weck.grpc_service.proto.fm_weck_service_pb2_grpc module
      • Module contents
  • Submodules
  • fm_weck.grpc_service.fm_weck_client module
    • cancel_run()
    • client_expert_run()
    • client_get_run()
    • client_run()
    • query_files()
    • start_expert_run()
    • start_run()
    • wait_on_run()
  • fm_weck.grpc_service.fm_weck_server module
    • FmWeckRemote
      • FmWeckRemote.cancelRun()
      • FmWeckRemote.cleanupRun()
      • FmWeckRemote.queryFiles()
      • FmWeckRemote.startExpertRun()
      • FmWeckRemote.startRun()
      • FmWeckRemote.waitOnRun()
    • serve()
  • fm_weck.grpc_service.request_handling module
    • RunHandler
      • RunHandler.cancel_run()
      • RunHandler.cleanup()
      • RunHandler.close()
      • RunHandler.failed()
      • RunHandler.failure()
      • RunHandler.get_c_program()
      • RunHandler.get_custom_property()
      • RunHandler.get_custom_tool()
      • RunHandler.get_file()
      • RunHandler.get_property()
      • RunHandler.get_tool()
      • RunHandler.glob()
      • RunHandler.is_canceled()
      • RunHandler.is_running()
      • RunHandler.join()
      • RunHandler.mp
      • RunHandler.output
      • RunHandler.output_files
      • RunHandler.ready()
      • RunHandler.start()
      • RunHandler.start_expert()
      • RunHandler.successful()
    • StillRunningError
    • worker()
  • fm_weck.grpc_service.run_store module
    • active_runs()
    • add_run()
    • get_run()
    • remove_run()
  • fm_weck.grpc_service.server_utils module
    • read_file()
  • Module contents
• fm_weck.resources package
  • Module contents
    • fm_tools_choice_map()
    • iter_fm_data()
    • iter_properties()
    • property_choice_map()

Submodules

fm_weck.cache_mgr module

fm_weck.cache_mgr.ask_and_clear(cache_location: str)

fm_weck.cache_mgr.clear_cache(cache_location: str)

fm_weck.capture module

class fm_weck.capture.Capture(target: TextIOBase, stream: Literal['stdout', 'stderr', 'stdin'] = 'stdout')

Bases: object

fm_weck.cli module

class fm_weck.cli.ShellCompletion

Bases: object

static properties_completer(prefix, parsed_args, **kwargs)

static tool_completer(prefix, parsed_args, **kwargs)

static versions_completer(prefix, parsed_args, **kwargs)

class fm_weck.cli.ToolQualifier(qualifier: str)

Bases: object

tool: str | Path

version: str | None

fm_weck.cli.add_shared_args_for_client(parser)

fm_weck.cli.add_shared_args_for_run_modes(parser)

fm_weck.cli.add_shared_args_for_zenodo_modes(parser)

fm_weck.cli.add_tool_arg(parser, nargs: str | None = '?')

fm_weck.cli.check_client_options(timelimit_arg, output_path=None)

fm_weck.cli.cli(raw_args: list[str])

fm_weck.cli.get_version_for_competition_year(tool_path: Path, year: int) → str | None

Find the tool version used in a competition for the given year. Searches for SV-COMP or Test-Comp participation entries.

Parameters:

• tool_path – Path to the tool’s YAML file

• year – Competition year (e.g., 2025)

Returns:

Version string if found, None otherwise

fm_weck.cli.list_known_properties()

fm_weck.cli.list_known_tools()

fm_weck.cli.log_no_image_error(tool, config)

fm_weck.cli.main_clear_cache(args: Namespace)

fm_weck.cli.main_client(args: Namespace)

fm_weck.cli.main_client_expert(args: Namespace)

fm_weck.cli.main_client_get_run(args: Namespace)

fm_weck.cli.main_client_query_files(args: Namespace)

fm_weck.cli.main_config(args: Namespace)

fm_weck.cli.main_install(args: Namespace)

fm_weck.cli.main_manual(args: Namespace)

fm_weck.cli.main_publish(args: Namespace)

fm_weck.cli.main_pull(args: Namespace)

fm_weck.cli.main_push(args: Namespace)

fm_weck.cli.main_run(args: Namespace)

fm_weck.cli.main_runexec(args: Namespace)

fm_weck.cli.main_server(args: Namespace)

fm_weck.cli.main_shell(args: Namespace)

fm_weck.cli.main_smoke_test(args: Namespace)

fm_weck.cli.main_versions(args: Namespace)

fm_weck.cli.parse(raw_args: list[str]) → Tuple[Callable[[], None], Namespace]

fm_weck.cli.resolve_property(prop_name: str) → Path

fm_weck.cli.resolve_property_for_server(prop_name: str) → Path | str

fm_weck.cli.resolve_tool(tool: ToolQualifier) → Path

fm_weck.cli.set_log_options(loglevel: str | None, logfile: str | None, config: dict[str, Any])

fm_weck.config module

class fm_weck.config.Config

Bases: object

The config singleton holds the configuration for the weck tool.

as_absolute_path(*args, **kwargs) → Path

Makes sure relative Paths in the config are relative to the config file.

property cache_location: Path

Makes sure relative Paths in the config are relative to the config file.

defaults() → dict[str, Any]

from_defaults_or_none(key: str) → Any

get(key: str, default: _T = None) → _T

get_checksum_db() → Path

get_shelve_path_for_benchexec() → Path

get_shelve_path_for_property(path: Path) → Path

get_shelve_space_for(fm_data: FmToolVersion) → Path

is_dry_run() → bool

load(config: Path | None = None) → dict[str, Any]

make_script_available(target_name: str = 'run_with_overlay.sh') → Path | None

mounts() → Iterable[Tuple[Path, Path]]

set_default_image(image: str) → None

set_dry_run(dry_run: bool) → None

fm_weck.config.parse_fm_data(fm_data: Path, version: str | None) → FmToolVersion

fm_weck.engine module

class fm_weck.engine.Docker(image: str | FmImageConfig)

Bases: Engine

class DockerBuildCommand(containerfile: Path, needs_sudo: bool = False)

Bases: BuildCommand

engine()

base_command()

benchexec_capabilities()

image_from(containerfile: Path)

setup_command()

class fm_weck.engine.Engine(image: str | FmImageConfig)

Bases: ABC

class BuildCommand(containerfile: Path, **kwargs)

Bases: ABC

base_image(image: str)

build()

build_args: List[str] = []

containerfile: Path

engine()

inspect_image(image: str)

load_image(image: str | Path)

packages(packages: Iterable[str])

save_image(image: str, output_path: Path)

tag_exists(tag: str) → bool

tag_image(image: str, tag: str)

add_benchexec_capabilities: bool = False

add_container_long_opt(arg: list[str])

Add a long option to the container command. If the first element of the list does not start with “–”, it will be prepended. Example: add_container_long_opt([”–option”, “value”]) -> –option value add_container_long_opt([“option”, “value”]) -> –option value

add_environment()

add_mounting_capabilities: bool = True

assemble_command(command: Iterable[str]) → list[str]

assemble_smoke_test_command(command: list[str | Path]) → list[str | Path]

base_command()

abstractmethod benchexec_capabilities()

dry_run: bool = False

static extract_image(fm: str | Path, version: str, config: dict) → str

from_config() → Engine

from_config(fm: Path, version: str, config: Config)

from_config(fm: str, version: str, config: Config)

from_config(fm: FmToolVersion, config: Config)

get_workdir() → str

handle_io: bool = True

image: str | None = None

abstractmethod image_from(containerfile: Path) → BuildCommand

interactive: bool = False

interactive_command()

mount(source: str, target: str)

mounting_capabilities()

overlay_tool_dir: str | None = None

print_output_to_stdout: bool = True

run(*command: str, timeout_sec: float | None = None) → RunResult

set_log_file(log_file: Path)

set_output_dir(output_dir: Path)

setup_command()

use_overlay(overlay_dir: str)

work_dir: Path = PosixPath('/home/cwd')

class fm_weck.engine.Podman(image: str | FmImageConfig)

Bases: Engine

class PodmanBuildCommand(containerfile: Path)

Bases: BuildCommand

benchexec_capabilities()

image_from(containerfile: Path)

class fm_weck.engine.Runexec(image: str | FmImageConfig)

Bases: Engine

base_command()

benchexec_capabilities()

image_from(containerfile: Path)

run(*command, timeout_sec: float | None = None) → RunResult

setup_command()

fm_weck.exceptions module

class fm_weck.exceptions.Failure(kind: str, message: str, detail: str | None = None)

Bases: object

detail: str | None = None

kind: str

message: str

exception fm_weck.exceptions.NoImageError

Bases: Exception

exception fm_weck.exceptions.RunFailedError(exit_code, command, output=None)

Bases: Exception

exception fm_weck.exceptions.ZenodoError

Bases: Exception

fm_weck.exceptions.failure_from_exception(e: BaseException) → Failure

fm_weck.exceptions.failure_to_error_code(failure: Failure)

fm_weck.file_util module

fm_weck.file_util.copy_ensuring_unix_line_endings(src: Path, dst: Path) → None

fm_weck.file_util.ensure_linux_style(path: str) → str

Ensure that the given path uses Linux-style forward slashes.

fm_weck.image_mgr module

class fm_weck.image_mgr.ImageMgr

Bases: object

The image manager singleton is responsible for preparing the images for the container.

check_if_doi_exists_locally(engine: Engine, doi: str) → str

Check if an image tagged with the given DOI already exists locally.

Returns:

The DOI if it does not exist locally, otherwise empty string.

load_image(engine: Engine, image_tar: Path, tag: str | None = None) → None

prepare_image(engine: Engine, image: FmImageConfig) → str

prepare_image_for_zenodo(engine: Engine, image_name: str) → tuple[str, str, Path]

Prepare an image for Zenodo upload, by retreiving the image tarball.

Returns:

A tuple containing the normalized image name and tarball path.

Raises:

ValueError – If the image is invalid or does not exist.

static print_dockerfile_warning()

tag_image(engine: Engine, image: str, tag: str) → None

fm_weck.run_result module

class fm_weck.run_result.RunResult(command: tuple[str, ...] | list[str], exit_code: int, raw_output: str)

Bases: object

as_benchexec_run()

command: tuple[str, ...] | list[str]

determine_result(tool: BaseTool2 | FmToolVersion) → str

exit_code: int

raw_output: str

fm_weck.runexec_mode module

fm_weck.runexec_mode.run_runexec(benchexec_package: Path | None, use_image: str | None, configuration: Config, extra_container_args: list[list[str]], command: list[str], mount_rw: list[Path] | None = None) → RunResult | None

fm_weck.runexec_util module

fm_weck.runexec_util.mountable_absolute_paths_of_command(cwd: Path, command: List[str]) → Iterable[Path]

Iterate over all arguments in command and find those that are paths. The paths are returned as absolute paths, that already exist on the host.

fm_weck.serve module

fm_weck.serve.check_cache_entry(shelve_space: Path, checksum: str, config: Config) → bool

fm_weck.serve.install_fm_tool(fm_tool: Path | FmToolVersion, version: str | None, configuration: Config, install_path: Path) → None

fm_weck.serve.map_doi(fm_data: FmToolVersion, tool_path: Path)

fm_weck.serve.run_guided(fm_tool: Path | FmToolVersion, version: str | None, configuration: Config, prop: Path | None, program_files: list[Path], additional_args: list[str], witness: Path | None = None, data_model: DataModel | None = None, offline_mode: bool = False, log_output_to: Path | None = None, output_files_to: Path | None = None, timeout_sec: float | None = None, print_tool_output_to_console: bool = True) → RunResult

fm_weck.serve.run_manual(fm_tool: Path | FmToolVersion, version: str | None, configuration: Config, command: list[str], offline_mode: bool = False, use_overlay: bool = False, log_output_to: Path | None = None, output_files_to: Path | None = None, timeout_sec: float | None = None, print_tool_output_to_console: bool = True) → RunResult

fm_weck.serve.setup_fm_tool(fm_tool: Path | FmToolVersion, version: str | None, configuration: Config, offline_mode: bool = False) → Tuple[FmToolVersion, Path]

fm_weck.serve.update_checksum(shelve_space: Path, checksum: str, config: Config)

fm_weck.smoke_test_mode module

exception fm_weck.smoke_test_mode.NoSmokeTestFileError

Bases: SmokeTestError

Exception raised when no smoke test file is found.

exception fm_weck.smoke_test_mode.SmokeTestError

Bases: Exception

Custom exception for smoke test errors.

exception fm_weck.smoke_test_mode.SmokeTestFileIsEmptyError

Bases: SmokeTestError

Exception raised when the smoke test file is empty.

exception fm_weck.smoke_test_mode.SmokeTestFileIsNotExecutableError(smoke_test_file: Path)

Bases: SmokeTestError

Exception raised when the smoke test file is not executable.

smoke_test_file: Path

fm_weck.smoke_test_mode.locate_and_check_smoke_test_file(shelve_space: Path) → str

Check if the smoke test file exists, is executable, and is not empty.

Parameters:

shelve_space – Path to the shelve space directory.

Returns:

The relative path to the smoke test file as a string.

Raises:

• NoSmokeTestFileError – If the smoke test file does not exist.

• SmokeTestFileIsEmptyError – If the smoke test file is empty.

• SmokeTestFileIsNotExecutableError – If the smoke test file is not executable.

fm_weck.smoke_test_mode.run_smoke_test(fm_data, shelve_space, config)

fm_weck.smoke_test_mode.run_smoke_test_gitlab_ci(fm_data: FmToolVersion, tool_dir: Path)

Run smoke test in GitLab CI mode. This mode directly installs required packages using apt instead of building/pulling images.

Parameters:

• fm_data – The FmToolVersion object containing tool information

• tool_dir – The directory containing the tool’s smoketest.sh script

fm_weck.tmp_file module

class fm_weck.tmp_file.NTempFile(name, mode='wb')

Bases: object

Custom temporary file context manager. It creates a temporary file and deletes it after the context manager is closed. The file is not kept open in order to achieve compatibility with Windows.

Inspired by https://stackoverflow.com/a/63173312

write(data)

Write data to the temporary file.

fm_weck.version_listing module

class fm_weck.version_listing.VersionListing(tool_paths: list)

Bases: object

print_versions()

tool_and_version: dict[str, list] = {}

fm_weck.zenodo module

class fm_weck.zenodo.TokenAuth(token: str)

Bases: Auth

auth_flow(request)

Execute the authentication flow.

To dispatch a request, yield it:

` yield request `

The client will .send() the response back into the flow generator. You can access it like so:

` response = yield request `

A return (or reaching the end of the generator) will result in the client returning the last response obtained from the server.

You can dispatch as many requests as is necessary.

class fm_weck.zenodo.ZenodoMgr(engine, access_token: str | None, image: str | None, doi: str | None, is_sandbox: bool = False, config: Config | None = None)

Bases: object

Zenodo API manager.

Parameters:

• engine (object) – The fm-weck engine object.

• access_token (str, optional) – The access token for authentication with the API. Defaults to None.

• image (str, optional) – The image to be published. Required if no doi is given. Defaults to None.

• doi (str, optional) – The DOI of the image to be pulled. Required if no image is given. Defaults to None.

Raises:

ValueError – If a chosen image is invalid or does not exist.

static load_zenodo_token(config)

Load the Zenodo token from the Zenodo config file.

Raises:

ZenodoError – If no Zenodo token is found.

publish()

Publish an image on Zenodo.

Raises:

ZenodoError – If Zenodo API returns an error.

pull(download_path: str | None, force_pull: bool = False)

Download an image from Zenodo using its DOI.

Raises:

• ValueError – If no DOI is provided.

• ZenodoError – If Zenodo API returns an error.

push()

Upload an image to Zenodo and leave it in draft mode (unpublished).

Raises:

ZenodoError – If Zenodo API returns an error.

static save_zenodo_token(config, token)

class fm_weck.zenodo.ZenodoSession(access_token: str | None = None, is_sandbox: bool = False)

Bases: object

check_existing_records(metadata: dict)

Check if a record with the same name already exists in the depositions. If it exists, prompt whether the image should be uploaded as a new version.

Returns:

The deposition ID in case of a positive prompt response, or an empty string.

delete_deposition(deposition_id: str)

download(doi: str, download_path: str | None, no_progress_bar: bool = False) → Path

Download an image from Zenodo using its DOI.

download_with_progress_bar(url: str)

get_upload_metadata(image_name: str, image_version: str) → dict

prepare_new_version_upload(metadata: dict, deposition_id: str) → None

Set up the metadata for the new version and delete the old versions’ files from that version.

publish(image_info: tuple)

Publish an image on Zenodo.

resolve_download_info(doi: str) → str

Extract the download URL from the deposition metadata.

Returns:

The download URL of the image.

save_image(image_name: str, image: bytes, save_path)

upload(image_info: tuple[str, str, Path]) → str

Upload an image to Zenodo and leave it in draft mode (unpublished).

upload_with_progress_bar(image_path, upload_name, bucket_url)

Module contents