galaxy_test.base package

Submodules

galaxy_test.base.api module

galaxy_test.base.api.celery_config()

class galaxy_test.base.api.UsesCeleryTasks

Bases: object

classmethod handle_galaxy_config_kwds(config: Dict[str, Any]) → None

celery_worker_parameters()

celery_parameters()

class galaxy_test.base.api.HasAnonymousGalaxyInteractor(*args, **kwargs)

Bases: Protocol

property anonymous_galaxy_interactor: ApiTestInteractor

Return an optionally anonymous galaxy interactor.

__init__(*args, **kwargs)

class galaxy_test.base.api.UsesApiTestCaseMixin

Bases: object

url: str

tearDown()

property anonymous_galaxy_interactor: ApiTestInteractor

Return an optionally anonymous galaxy interactor.

Lighter requirements for use with API requests that may not required an API key.

property galaxy_interactor: ApiTestInteractor

class galaxy_test.base.api.ApiTestInteractor(test_case, api_key=None)

Bases: TestCaseGalaxyInteractor

Specialized variant of the API interactor (originally developed for tool functional tests) for testing the API generally.

__init__(test_case, api_key=None)

cookies: RequestsCookieJar | None

get(*args, **kwds)

head(*args, **kwds)

post(*args, **kwds)

delete(*args, **kwds)

patch(*args, **kwds)

put(*args, **kwds)

api_key: str | None

keep_outputs_dir: str | None

class galaxy_test.base.api.AnonymousGalaxyInteractor(test_case)

Bases: ApiTestInteractor

__init__(test_case)

api_key: str | None

cookies: RequestsCookieJar | None

keep_outputs_dir: str | None

galaxy_test.base.api_asserts module

Utility methods for making assertions about Galaxy API responses, etc…

galaxy_test.base.api_asserts.assert_status_code_is(response: Response, expected_status_code: int, failure_message: str | None = None)

Assert that the supplied response has the expect status code.

galaxy_test.base.api_asserts.assert_status_code_is_ok(response: Response, failure_message: str | None = None)

Assert that the supplied response is okay.

The easier alternative response.raise_for_status() might be preferable generally.

See also

requests.Response.raise_for_status()

galaxy_test.base.api_asserts.assert_has_keys(response: dict, *keys: str)

Assert that the supplied response (dict) has the supplied keys.

galaxy_test.base.api_asserts.assert_not_has_keys(response: dict, *keys: str)

Assert that the supplied response (dict) does not have the supplied keys.

galaxy_test.base.api_asserts.assert_error_code_is(response: Response | dict, error_code: int | ErrorCode)

Assert that the supplied response has the supplied Galaxy error code.

Galaxy error codes can be imported from galaxy.exceptions.error_codes to test against.

galaxy_test.base.api_asserts.assert_object_id_error(response: Response)

galaxy_test.base.api_asserts.assert_error_message_contains(response: Response | dict, expected_contains: str)

galaxy_test.base.api_asserts.assert_has_key(response: dict, *keys: str)

Assert that the supplied response (dict) has the supplied keys.

galaxy_test.base.api_util module

galaxy_test.base.api_util.get_admin_api_key() → str

Test admin API key to use for functional tests.

This key should be configured as a admin API key and should be able to create additional users and keys.

galaxy_test.base.api_util.get_user_api_key() → str | None

Test user API key to use for functional tests.

If set, this should drive API based testing - if not set an admin API key will be used to create a new user and API key for tests.

galaxy_test.base.api_util.baseauth_headers(username: str, password: str) → Dict[str, str]

galaxy_test.base.api_util.random_name(prefix: str | None = None, suffix: str | None = None, len: int = 10) → str

galaxy_test.base.constants module

Just constants useful for testing across test types.

galaxy_test.base.env module

Base utilities for working Galaxy test environments.

galaxy_test.base.env.setup_keep_outdir() → str

galaxy_test.base.env.target_url_parts() → Tuple[str, str | None, str]

galaxy_test.base.env.get_ip_address(ifname: str) → str

galaxy_test.base.instrument module

galaxy_test.base.interactor module

class galaxy_test.base.interactor.TestCaseGalaxyInteractor(functional_test_case, test_user=None, api_key=None)

Bases: GalaxyInteractorApi

__init__(functional_test_case, test_user=None, api_key=None)

api_key: str | None

cookies: RequestsCookieJar | None

keep_outputs_dir: str | None

galaxy_test.base.populators module

Abstractions used by the Galaxy testing frameworks for interacting with the Galaxy API.

These abstractions are geared toward testing use cases and populating fixtures. For a more general framework for working with the Galaxy API checkout bioblend.

The populators are broken into different categories of data one might want to populate and work with (datasets, histories, libraries, and workflows). Within each populator type abstract classes describe high-level functionality that depend on abstract HTTP verbs executions (e.g. methods for executing GET, POST, DELETE). The abstract classes are galaxy_test.base.populators.BaseDatasetPopulator, galaxy_test.base.populators.BaseWorkflowPopulator, and galaxy_test.base.populators.BaseDatasetCollectionPopulator.

There are a few different concrete ways to supply these low-level verb executions. For instance galaxy_test.base.populators.DatasetPopulator implements the abstract galaxy_test.base.populators.BaseDatasetPopulator by leveraging a galaxy interactor galaxy.tool_util.interactor.GalaxyInteractorApi. It is non-intuitive that the Galaxy testing framework uses the tool testing code inside Galaxy’s code base for a lot of heavy lifting. This is due to the API testing framework organically growing from the tool testing framework that predated it and then the tool testing framework being extracted for re-use in Planemo, etc..

These other two concrete implementation of the populators are much more direct and intuitive. galaxy_test.base.populators.GiDatasetPopulator, et. al. are populators built based on Bioblend gi objects to build URLs and describe API keys. galaxy_test.selenium.framework.SeleniumSessionDatasetPopulator, et al. are populators built based on Selenium sessions to leverage Galaxy cookies for auth for instance.

All three of these implementations are now effectively light wrappers around requests. Not leveraging requests directly is a bit ugly and this ugliness again stems from these organically growing from a framework that originally didn’t use requests at all.

API tests and Selenium tests routinely use requests directly and that is totally fine, requests should just be filtered through the verb abstractions if that functionality is then added to populators to be shared across tests or across testing frameworks.

galaxy_test.base.populators.flakey(method)

galaxy_test.base.populators.skip_without_tool(tool_id: str)

Decorate an API test method as requiring a specific tool.

Have test framework skip the test case if the tool is unavailable.

galaxy_test.base.populators.skip_without_asgi(method)

galaxy_test.base.populators.skip_without_datatype(extension: str)

Decorate an API test method as requiring a specific datatype.

Have test framework skip the test case if the datatype is unavailable.

galaxy_test.base.populators.skip_without_visualization_plugin(plugin_name: str)

galaxy_test.base.populators.summarize_instance_history_on_error(method)

galaxy_test.base.populators.conformance_tests_gen(directory, filename='conformance_tests.yaml')

class galaxy_test.base.populators.CwlRun(dataset_populator, history_id)

Bases: object

__init__(dataset_populator, history_id)

get_output_as_object(output_name, download_folder=None)

abstract wait()

Wait for the completion of the job(s) generated by this run.

class galaxy_test.base.populators.CwlToolRun(dataset_populator, history_id, run_response)

Bases: CwlRun

__init__(dataset_populator, history_id, run_response)

property job_id

wait()

Wait for the completion of the job(s) generated by this run.

class galaxy_test.base.populators.CwlWorkflowRun(dataset_populator, workflow_populator, history_id, workflow_id, invocation_id)

Bases: CwlRun

__init__(dataset_populator, workflow_populator, history_id, workflow_id, invocation_id)

wait()

Wait for the completion of the job(s) generated by this run.

class galaxy_test.base.populators.BasePopulator

Bases: object

galaxy_interactor: ApiTestInteractor

class galaxy_test.base.populators.BaseDatasetPopulator

Bases: BasePopulator

Abstract description of API operations optimized for testing Galaxy - implementations must implement _get, _post and _delete.

new_dataset(history_id: str, content=None, wait: bool = False, fetch_data=True, to_posix_lines=True, auto_decompress=True, **kwds) → Dict[str, Any]

Create a new history dataset instance (HDA).

Returns:

a dictionary describing the new HDA

new_dataset_request(history_id: str, content=None, wait: bool = False, fetch_data=True, **kwds) → Response

Lower-level dataset creation that returns the upload tool response object.

new_bam_dataset(history_id: str, test_data_resolver)

fetch(payload: dict, assert_ok: bool = True, timeout: int | float = 60, wait: bool | None = None)

fetch_hdas(history_id: str, items: List[Dict[str, Any]], wait: bool = True) → List[Dict[str, Any]]

fetch_hda(history_id: str, item: Dict[str, Any], wait: bool = True) → Dict[str, Any]

create_deferred_hda(history_id, uri: str, ext: str | None = None) → Dict[str, Any]

export_dataset_to_remote_file(history_id: str, content: str, name: str, target_uri: str)

tag_dataset(history_id, hda_id, tags, raise_on_error=True)

create_from_store_raw(payload: Dict[str, Any]) → Response

create_from_store_raw_async(payload: Dict[str, Any]) → Response

create_from_store(store_dict: Dict[str, Any] | None = None, store_path: str | None = None, model_store_format: str | None = None) → Dict[str, Any]

create_from_store_async(store_dict: Dict[str, Any] | None = None, store_path: str | None = None, model_store_format: str | None = None) → Dict[str, Any]

create_contents_from_store_raw(history_id: str, payload: Dict[str, Any]) → Response

create_contents_from_store(history_id: str, store_dict: Dict[str, Any] | None = None, store_path: str | None = None) → List[Dict[str, Any]]

download_contents_to_store(history_id: str, history_content: Dict[str, Any], extension='.tgz') → str

reupload_contents(history_content: Dict[str, Any])

wait_for_tool_run(history_id: str, run_response: Response, timeout: int | float = 60, assert_ok: bool = True)

check_run(run_response: Response) → dict

wait_for_history(history_id: str, assert_ok: bool = False, timeout: int | float = 60) → str

wait_for_history_jobs(history_id: str, assert_ok: bool = False, timeout: int | float = 60)

wait_for_jobs(jobs: List[dict] | List[str], assert_ok: bool = False, timeout: int | float = 60, ok_states=None)

wait_for_job(job_id: str, assert_ok: bool = False, timeout: int | float = 60, ok_states=None)

get_job_details(job_id: str, full: bool = False) → Response

compute_hash(dataset_id: str, hash_function: str | None = 'MD5', extra_files_path: str | None = None, wait: bool = True) → Response

cancel_history_jobs(history_id: str, wait=True) → None

history_jobs(history_id: str) → List[Dict[str, Any]]

history_jobs_for_tool(history_id: str, tool_id: str) → List[Dict[str, Any]]

invocation_jobs(invocation_id: str) → List[Dict[str, Any]]

active_history_jobs(history_id: str) → list

cancel_job(job_id: str) → Response

delete_history(history_id: str) → None

delete_dataset(history_id: str, content_id: str, purge: bool = False, stop_job: bool = False, wait_for_purge: bool = False, use_query_params: bool = False) → Response

wait_for_purge(history_id, content_id)

create_tool_from_path(tool_path: str) → Dict[str, Any]

create_tool(representation, tool_directory: str | None = None) → Dict[str, Any]

list_dynamic_tools() → list

show_dynamic_tool(uuid) → dict

deactivate_dynamic_tool(uuid) → dict

test_history_for(method) → Generator[str, None, None]

test_history(require_new: bool = True) → Generator[str, None, None]

new_history(name='API Test History', **kwds) → str

copy_history(history_id, name='API Test Copied History', **kwds) → Response

fetch_payload(history_id: str, content: str, auto_decompress: bool = False, file_type: str = 'txt', dbkey: str = '?', name: str = 'Test_Dataset', **kwds) → dict

upload_payload(history_id: str, content: str | None = None, **kwds) → dict

get_remote_files(target: str = 'ftp') → dict

run_tool_payload(tool_id: str | None, inputs: dict, history_id: str, **kwds) → dict

build_tool_state(tool_id: str, history_id: str)

run_tool_raw(tool_id: str | None, inputs: dict, history_id: str, **kwds) → Response

run_tool(tool_id: str, inputs: dict, history_id: str, **kwds)

tools_post(payload: dict, url='tools') → Response

materialize_dataset_instance(history_id: str, id: str, source: str = 'hda')

get_history_dataset_content(history_id: str, wait=True, filename=None, type='text', to_ext=None, raw=False, **kwds)

display_chunk(dataset_id: str, offset: int = 0, ck_size: int | None = None) → Dict[str, Any]

get_history_dataset_source_transform_actions(history_id: str, **kwd) → Set[str]

get_history_dataset_details(history_id: str, keys: str | None = None, **kwds) → Dict[str, Any]

get_history_dataset_details_raw(history_id: str, dataset_id: str, keys: str | None = None) → Response

get_history_dataset_extra_files(history_id: str, **kwds) → list

get_history_collection_details(history_id: str, **kwds) → dict

run_collection_creates_list(history_id: str, hdca_id: str) → Response

new_error_dataset(history_id: str) → str

report_job_error_raw(job_id: str, dataset_id: str, message: str = '', email: str | None = None) → Response

report_job_error(job_id: str, dataset_id: str, message: str = '', email: str | None = None) → Response

run_detect_errors(history_id: str, exit_code: int, stdout: str = '', stderr: str = '') → dict

run_exit_code_from_file(history_id: str, hdca_id: str) → dict

get_history_contents(history_id: str, data=None) → List[Dict[str, Any]]

ds_entry(history_content: dict) → dict

dataset_storage_info(dataset_id: str) → Dict[str, Any]

dataset_storage_info_raw(dataset_id: str) → Response

get_roles() → list

get_configuration(admin=False) → Dict[str, Any]

user_email() → str

user_id() → str

user_private_role_id() → str

get_usage() → List[Dict[str, Any]]

get_usage_for(label: str | None) → Dict[str, Any]

update_user(properties: Dict[str, Any]) → Dict[str, Any]

set_user_preferred_object_store_id(store_id: str | None) → None

update_user_raw(properties: Dict[str, Any]) → Response

total_disk_usage() → float

update_object_store_id(dataset_id: str, object_store_id: str)

create_role(user_ids: list, description: str | None = None) → dict

create_quota(quota_payload: dict) → dict

get_quotas() → list

make_private(history_id: str, dataset_id: str) → dict

make_dataset_public_raw(history_id: str, dataset_id: str) → Response

update_permissions_raw(history_id: str, dataset_id: str, payload: dict) → Response

make_public(history_id: str) → dict

validate_dataset(history_id: str, dataset_id: str) → Dict[str, Any]

validate_dataset_and_wait(history_id, dataset_id) → str | None

setup_history_for_export_testing(history_name)

prepare_export(history_id, data)

export_url(history_id: str, data, check_download: bool = True) → str

get_export_url(export_url) → Response

import_history(import_data)

wait_for_history_with_name(history_name: str, desc: str) → Dict[str, Any]

import_history_and_wait_for_name(import_data, history_name)

history_names() → Dict[str, Dict]

rename_history(history_id: str, new_name: str)

update_history(history_id: str, payload: Dict[str, Any]) → Response

get_histories()

wait_on_history_length(history_id: str, wait_on_history_length: int)

wait_on_download(download_request_response: Response) → Response

assert_download_request_ok(download_request_response: Response) → UUID

Assert response is valid and okay and extract storage request ID.

wait_for_download_ready(storage_request_id: UUID)

wait_on_task(async_task_response: Response)

wait_on_task_id(task_id: str)

wait_on_download_request(storage_request_id: UUID) → Response

history_length(history_id)

reimport_history(history_id, history_name, wait_on_history_length, export_kwds, task_based=False)

get_random_name(prefix: str | None = None, suffix: str | None = None, len: int = 10) → str

wait_for_dataset(history_id: str, dataset_id: str, assert_ok: bool = False, timeout: int | float = 60) → str

create_object_store_raw(payload: Dict[str, Any]) → Response

create_object_store(payload: Dict[str, Any]) → Dict[str, Any]

upgrade_object_store_raw(id: str, payload: Dict[str, Any]) → Response

upgrade_object_store(id: str, payload: Dict[str, Any]) → Dict[str, Any]

update_object_store_raw(id: str, payload: Dict[str, Any]) → Response

update_object_store(id: str, payload: Dict[str, Any]) → Dict[str, Any]

selectable_object_stores() → List[Dict[str, Any]]

selectable_object_store_ids() → List[str]

new_page(slug: str = 'mypage', title: str = 'MY PAGE', content_format: str = 'html', content: str | None = None) → Dict[str, Any]

new_page_raw(slug: str = 'mypage', title: str = 'MY PAGE', content_format: str = 'html', content: str | None = None) → Response

new_page_payload(slug: str = 'mypage', title: str = 'MY PAGE', content_format: str = 'html', content: str | None = None) → Dict[str, str]

export_history_to_uri_async(history_id: str, target_uri: str, model_store_format: str = 'tgz', include_files: bool = True)

import_history_from_uri_async(target_uri: str, model_store_format: str)

download_history_to_store(history_id: str, extension: str = 'tgz', serve_file: bool = False)

get_history_export_tasks(history_id: str)

make_page_public(page_id: str) → Dict[str, Any]

wait_for_export_task_on_record(export_record)

archive_history(history_id: str, export_record_id: str | None = None, purge_history: bool | None = False) → Response

restore_archived_history(history_id: str, force: bool | None = None) → Response

get_archived_histories(query: str | None = None) → List[Dict[str, Any]]

galaxy_interactor: ApiTestInteractor

class galaxy_test.base.populators.GalaxyInteractorHttpMixin

Bases: object

galaxy_interactor: ApiTestInteractor

class galaxy_test.base.populators.DatasetPopulator(galaxy_interactor: ApiTestInteractor)

Bases: GalaxyInteractorHttpMixin, BaseDatasetPopulator

__init__(galaxy_interactor: ApiTestInteractor) → None

galaxy_interactor: ApiTestInteractor

class galaxy_test.base.populators.BaseWorkflowPopulator

Bases: BasePopulator

dataset_populator: BaseDatasetPopulator

dataset_collection_populator: BaseDatasetCollectionPopulator

load_workflow(name: str, content: str = '{\n    "a_galaxy_workflow": "true", \n    "annotation": "simple workflow",\n    "format-version": "0.1", \n    "name": "TestWorkflow1", \n    "steps": {\n        "0": {\n            "annotation": "input1 description", \n            "id": 0, \n            "input_connections": {}, \n            "inputs": [\n                {\n                    "description": "input1 description", \n                    "name": "WorkflowInput1"\n                }\n            ], \n            "name": "Input dataset", \n            "outputs": [], \n            "position": {\n                "left": 199.55555772781372, \n                "top": 200.66666460037231\n            }, \n            "tool_errors": null, \n            "tool_id": null, \n            "tool_state": "{\\"name\\": \\"WorkflowInput1\\"}", \n            "tool_version": null, \n            "type": "data_input", \n            "user_outputs": []\n        }, \n        "1": {\n            "annotation": "", \n            "id": 1, \n            "input_connections": {}, \n            "inputs": [\n                {\n                    "description": "", \n                    "name": "WorkflowInput2"\n                }\n            ], \n            "name": "Input dataset", \n            "outputs": [], \n            "position": {\n                "left": 206.22221422195435, \n                "top": 327.33335161209106\n            }, \n            "tool_errors": null, \n            "tool_id": null, \n            "tool_state": "{\\"name\\": \\"WorkflowInput2\\"}", \n            "tool_version": null, \n            "type": "data_input", \n            "user_outputs": []\n        }, \n        "2": {\n            "annotation": "", \n            "id": 2, \n            "input_connections": {\n                "input1": {\n                    "id": 0, \n                    "output_name": "output"\n                }, \n                "queries_0|input2": {\n                    "id": 1, \n                    "output_name": "output"\n                }\n            }, \n            "inputs": [], \n            "name": "Concatenate datasets", \n            "outputs": [\n                {\n                    "name": "out_file1", \n                    "type": "input"\n                }\n            ], \n            "position": {\n                "left": 419.33335876464844, \n                "top": 200.44446563720703\n            }, \n            "post_job_actions": {}, \n            "tool_errors": null, \n            "tool_id": "cat1", \n            "tool_state": "{\\"__page__\\": 0, \\"__rerun_remap_job_id__\\": null, \\"input1\\": \\"null\\", \\"queries\\": \\"[{\\\\\\"input2\\\\\\": null, \\\\\\"__index__\\\\\\": 0}]\\"}", \n            "tool_version": "1.0.0", \n            "type": "tool", \n            "user_outputs": []\n        }\n    }\n}\n', add_pja=False) → dict

load_random_x2_workflow(name: str) → dict

load_workflow_from_resource(name: str, filename: str | None = None) → dict

simple_workflow(name: str, **create_kwds) → str

import_workflow_from_path_raw(from_path: str, object_id: str | None = None) → Response

import_workflow_from_path(from_path: str, object_id: str | None = None) → str

create_workflow(workflow: Dict[str, Any], **create_kwds) → str

create_workflow_response(workflow: Dict[str, Any], **create_kwds) → Response

upload_yaml_workflow(yaml_content: str | PathLike | dict, **kwds) → str

wait_for_invocation(workflow_id: str | None, invocation_id: str, timeout: int | float = 60, assert_ok: bool = True) → str

workflow_invocations(workflow_id: str, include_nested_invocations=True) → List[Dict[str, Any]]

cancel_invocation(invocation_id: str)

history_invocations(history_id: str, include_nested_invocations: bool = True) → List[Dict[str, Any]]

wait_for_history_workflows(history_id: str, assert_ok: bool = True, timeout: int | float = 60, expected_invocation_count: int | None = None) → None

wait_for_workflow(workflow_id: str | None, invocation_id: str, history_id: str, assert_ok: bool = True, timeout: int | float = 60) → None

Wait for a workflow invocation to completely schedule and then history to be complete.

get_invocation(invocation_id, step_details=False)

download_invocation_to_store(invocation_id, include_files=False, extension='tgz')

download_invocation_to_uri(invocation_id, target_uri, extension='tgz')

create_invocation_from_store_raw(history_id: str, store_dict: Dict[str, Any] | None = None, store_path: str | None = None, model_store_format: str | None = None) → Response

create_invocation_from_store(history_id: str, store_dict: Dict[str, Any] | None = None, store_path: str | None = None, model_store_format: str | None = None) → Response

validate_biocompute_object(bco, expected_schema_version='https://w3id.org/ieee/ieee-2791-schema/2791object.json')

get_ro_crate(invocation_id, include_files=False)

validate_invocation_crate_directory(crate_directory)

invoke_workflow_raw(workflow_id: str, request: dict, assert_ok: bool = False) → Response

invoke_workflow(workflow_id: str, history_id: str | None = None, inputs: dict | None = None, request: dict | None = None, inputs_by: str = 'step_index') → Response

invoke_workflow_and_assert_ok(workflow_id: str, history_id: str | None = None, inputs: dict | None = None, request: dict | None = None, inputs_by: str = 'step_index') → str

invoke_workflow_and_wait(workflow_id: str, history_id: str | None = None, inputs: dict | None = None, request: dict | None = None) → Response

workflow_report_json(workflow_id: str, invocation_id: str) → dict

workflow_report_pdf(workflow_id: str, invocation_id: str) → Response

download_workflow(workflow_id: str, style: str | None = None, history_id: str | None = None) → dict

set_tags(workflow_id: str, tags: List[str]) → None

update_workflow(workflow_id: str, workflow_object: dict) → Response

refactor_workflow(workflow_id: str, actions: list, dry_run: bool | None = None, style: str | None = None) → Response

export_for_update(workflow_id)

run_workflow(has_workflow: str | PathLike | dict, test_data: str | dict | None = None, history_id: str | None = None, wait: bool = True, source_type: str | None = None, jobs_descriptions=None, expected_response: int = 200, assert_ok: bool = True, client_convert: bool | None = None, extra_invocation_kwds: Dict[str, Any] | None = None, round_trip_format_conversion: bool = False, invocations: int = 1, raw_yaml: bool = False)

High-level wrapper around workflow API, etc. to invoke format 2 workflows.

dump_workflow(workflow_id, style=None)

workflow_inputs(workflow_id: str) → Dict[str, Dict[str, Any]]

build_ds_map(workflow_id: str, label_map: Dict[str, Any]) → str

setup_workflow_run(workflow: Dict[str, Any] | None = None, inputs_by: str = 'step_id', history_id: str | None = None, workflow_id: str | None = None) → Tuple[Dict[str, Any], str, str]

get_invocation_jobs(invocation_id: str) → List[Dict[str, Any]]

wait_for_invocation_and_jobs(history_id: str, workflow_id: str, invocation_id: str, assert_ok: bool = True) → None

index(show_shared: bool | None = None, show_published: bool | None = None, sort_by: str | None = None, sort_desc: bool | None = None, limit: int | None = None, offset: int | None = None, search: str | None = None, skip_step_counts: bool | None = None)

index_ids(show_shared: bool | None = None, show_published: bool | None = None, sort_by: str | None = None, sort_desc: bool | None = None, limit: int | None = None, offset: int | None = None, search: str | None = None)

share_with_user(workflow_id: str, user_id_or_email: str)

class galaxy_test.base.populators.RunJobsSummary(history_id, workflow_id, invocation_id, inputs, jobs, invocation, workflow_request)

Bases: tuple

history_id: str

Alias for field number 0

workflow_id: str

Alias for field number 1

invocation_id: str

Alias for field number 2

inputs: dict

Alias for field number 3

jobs: list

Alias for field number 4

invocation: dict

Alias for field number 5

workflow_request: dict

Alias for field number 6

jobs_for_tool(tool_id)

class galaxy_test.base.populators.WorkflowPopulator(galaxy_interactor)

Bases: GalaxyInteractorHttpMixin, BaseWorkflowPopulator, ImporterGalaxyInterface

__init__(galaxy_interactor)

galaxy_interactor: ApiTestInteractor

import_workflow(workflow, **kwds) → Dict[str, Any]

Import a workflow via POST /api/workflows or comparable interface into Galaxy.

import_tool(tool) → Dict[str, Any]

Import a workflow via POST /api/workflows or comparable interface into Galaxy.

build_module(step_type: str, content_id: str | None = None, inputs: Dict[str, Any] | None = None)

scaling_workflow_yaml(**kwd)

class galaxy_test.base.populators.CwlPopulator(dataset_populator: DatasetPopulator, workflow_populator: WorkflowPopulator)

Bases: object

__init__(dataset_populator: DatasetPopulator, workflow_populator: WorkflowPopulator)

get_conformance_test(version: str, doc: str)

run_cwl_job(artifact: str, job_path: str | None = None, job: Dict | None = None, test_data_directory: str | None = None, history_id: str | None = None, assert_ok: bool = True) → CwlRun

Parameters:

artifact – CWL tool id, or (absolute or relative) path to a CWL tool or workflow file

run_conformance_test(version: str, doc: str)

class galaxy_test.base.populators.LibraryPopulator(galaxy_interactor)

Bases: object

__init__(galaxy_interactor)

get_libraries()

new_private_library(name)

create_from_store_raw(payload: Dict[str, Any]) → Response

create_from_store(store_dict: Dict[str, Any] | None = None, store_path: str | None = None) → List[Dict[str, Any]]

new_library(name)

fetch_single_url_to_folder(file_type='auto', assert_ok=True)

get_permissions(library_id, scope: str | None = 'current', is_library_access: bool | None = False, page: int | None = 1, page_limit: int | None = 1000, q: str | None = None, admin: bool | None = True)

set_permissions(library_id, role_id=None)

Old legacy way of setting permissions.

set_permissions_with_action(library_id, role_id=None, action=None)

set_access_permission(library_id, role_id, action=None)

set_add_permission(library_id, role_id, action=None)

set_manage_permission(library_id, role_id, action=None)

set_modify_permission(library_id, role_id, action=None)

user_email()

user_private_role_id()

create_dataset_request(library, **kwds)

new_library_dataset(name, **create_dataset_kwds)

wait_on_library_dataset(library_id, dataset_id)

raw_library_contents_create(library_id, payload, files=None)

show_ld_raw(library_id: str, library_dataset_id: str) → Response

show_ld(library_id: str, library_dataset_id: str) → Dict[str, Any]

show_ldda(ldda_id)

new_library_dataset_in_private_library(library_name='private_dataset', wait=True)

get_library_contents(library_id: str) → List[Dict[str, Any]]

get_library_contents_with_path(library_id: str, path: str) → Dict[str, Any]

setup_fetch_to_folder(test_name)

class galaxy_test.base.populators.BaseDatasetCollectionPopulator

Bases: object

dataset_populator: BaseDatasetPopulator

create_list_from_pairs(history_id, pairs, name='Dataset Collection from pairs')

nested_collection_identifiers(history_id: str, collection_type)

create_nested_collection(history_id, collection_type, name=None, collection=None, element_identifiers=None)

Create a nested collection either from collection or using collection_type).

create_list_of_pairs_in_history(history_id, **kwds)

create_list_of_list_in_history(history_id: str, **kwds)

create_pair_in_history(history_id: str, wait: bool = False, **kwds)

create_list_in_history(history_id: str, wait: bool = False, **kwds)

upload_collection(history_id: str, collection_type, elements, wait: bool = False, **kwds)

create_list_payload(history_id: str, **kwds)

create_pair_payload(history_id: str, **kwds)

wait_for_fetched_collection(fetch_response: Dict[str, Any] | Response)

pair_identifiers(history_id: str, contents=None, wait: bool = False)

list_identifiers(history_id: str, contents=None)

wait_for_dataset_collection(create_payload: dict, assert_ok: bool = False, timeout: int | float = 60) → None

class galaxy_test.base.populators.DatasetCollectionPopulator(galaxy_interactor: ApiTestInteractor)

Bases: BaseDatasetCollectionPopulator

__init__(galaxy_interactor: ApiTestInteractor)

dataset_populator: BaseDatasetPopulator

galaxy_test.base.populators.load_data_dict(history_id: str, test_data: Dict[str, Any], dataset_populator: BaseDatasetPopulator, dataset_collection_populator: BaseDatasetCollectionPopulator) → Tuple[Dict[str, Any], Dict[str, Any], bool]

Load a dictionary as inputs to a workflow (test data focused).

galaxy_test.base.populators.stage_inputs(galaxy_interactor: ApiTestInteractor, history_id: str, job: Dict[str, Any], use_path_paste: bool = True, use_fetch_api: bool = True, to_posix_lines: bool = True, tool_or_workflow: typing_extensions.Literal[tool, workflow] = 'workflow', job_dir: str | None = None) → Tuple[Dict[str, Any], List[Dict[str, Any]]]

Alternative to load_data_dict that uses production-style workflow inputs.

galaxy_test.base.populators.stage_rules_example(galaxy_interactor: ApiTestInteractor, history_id: str, example: Dict[str, Any]) → Dict[str, Any]

Wrapper around stage_inputs for staging collections defined by rules spec DSL.

galaxy_test.base.populators.wait_on_state(state_func: Callable, desc: str = 'state', skip_states=None, ok_states=None, assert_ok: bool = False, timeout: int | float = 60) → str

class galaxy_test.base.populators.GiHttpMixin

Bases: object

Mixin for adapting Galaxy testing populators helpers to bioblend.

class galaxy_test.base.populators.GiDatasetPopulator(gi)

Bases: GiHttpMixin, BaseDatasetPopulator

Implementation of BaseDatasetPopulator backed by bioblend.

__init__(gi)

Construct a dataset populator from a bioblend GalaxyInstance.

class galaxy_test.base.populators.GiDatasetCollectionPopulator(gi)

Bases: GiHttpMixin, BaseDatasetCollectionPopulator

Implementation of BaseDatasetCollectionPopulator backed by bioblend.

__init__(gi)

Construct a dataset collection populator from a bioblend GalaxyInstance.

class galaxy_test.base.populators.GiWorkflowPopulator(gi)

Bases: GiHttpMixin, BaseWorkflowPopulator

Implementation of BaseWorkflowPopulator backed by bioblend.

__init__(gi)

Construct a workflow populator from a bioblend GalaxyInstance.

galaxy_test.base.populators.wait_on(function: Callable, desc: str, timeout: int | float = 60)

galaxy_test.base.populators.wait_on_assertion(function: Callable, desc: str, timeout: int | float = 60)

galaxy_test.base.rules_test_data module

galaxy_test.base.rules_test_data.check_example_1(hdca, dataset_populator)

galaxy_test.base.rules_test_data.check_example_2(hdca, dataset_populator)

galaxy_test.base.rules_test_data.check_example_3(hdca, dataset_populator)

galaxy_test.base.rules_test_data.check_example_4(hdca, dataset_populator)

galaxy_test.base.rules_test_data.check_example_5(hdca, dataset_populator)

galaxy_test.base.rules_test_data.check_example_6(hdca, dataset_populator)

galaxy_test.base.testcase module

class galaxy_test.base.testcase.FunctionalTestCase

Bases: TestCase

Base class for tests targetting actual Galaxy servers.

Subclass should override galaxy_driver_class if a Galaxy server needs to be launched to run the test, this base class assumes a server is already running.

galaxy_driver_class: type | None = None

host: str

port: str | None

url: str

keepOutdir: str

test_data_resolver: TestDataResolver

setUp() → None

classmethod setUpClass()

Configure and start Galaxy for a test.

classmethod tearDownClass()

Shutdown Galaxy server and cleanup temp directory.

get_filename(filename: str) → str

pytestmark = [Mark(name='usefixtures', args=('embedded_driver',), kwargs={})]

galaxy_test.base.uses_shed module

galaxy_test.base.workflow_fixtures module