Skip to content
Permalink

Comparing changes

This is a direct comparison between two commits made in this repository or its related repositories. View the default comparison for this range or learn more about diff comparisons.

Open a pull request

Create a new pull request by comparing changes across two branches. If you need to, you can also . Learn more about diff comparisons here.
base repository: galaxyproject/planemo
Failed to load repositories. Confirm that selected base ref is valid, then try again.
Loading
base: 18ce0b1a51de9691b1252d1b0ad061e683af80ed
Choose a base ref
..
head repository: galaxyproject/planemo
Failed to load repositories. Confirm that selected head ref is valid, then try again.
Loading
compare: d97057a381a9d93e45ff3839930d0e264d789b22
Choose a head ref
Showing with 42 additions and 0 deletions.
  1. +42 −0 planemo/commands/cmd_workflow_test_check.py
42 changes: 42 additions & 0 deletions planemo/commands/cmd_workflow_test_check.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
"""Module describing the planemo ``workflow_test_check`` command."""
import click

from planemo import options
from planemo.cli import command_function
from planemo.engine.factory import engine_context
from planemo.galaxy.activity import invocation_to_run_response
from planemo.galaxy.test.actions import handle_reports_and_summary
from planemo.runnable import definition_to_test_case
from planemo.runnable_resolve import for_runnable_identifier
from planemo.test.results import StructuredData


@click.command("workflow_test_check")
@options.optional_tools_arg(multiple=False, allow_uris=False, metavar="TEST DEFINITION")
@options.required_workflow_arg()
@options.galaxy_url_option(required=True)
@options.galaxy_user_key_option(required=True)
@options.test_index_option()
@options.test_options()
@command_function
def cli(ctx, path, workflow_identifier, test_index, **kwds):
"""Run defined tests against existing workflow invocation."""
with engine_context(ctx, engine="external_galaxy", **kwds) as engine, engine.ensure_runnables_served([]) as config:
user_gi = config.user_gi
invocation = user_gi.invocations.show_invocation(workflow_identifier)
runnable = for_runnable_identifier(ctx, invocation["workflow_id"], kwds)
test_cases = definition_to_test_case(path, runnable)
assert (
len(test_cases) >= test_index
), f"Selected test case {test_index}, but only found {len(test_cases)} test case(s)."
test_case = test_cases[test_index - 1]
run_response = invocation_to_run_response(ctx, user_gi=config.user_gi, runnable=runnable, invocation=invocation)
structured_data = test_case.structured_test_data(run_response)
test_data = {
"version": "0.1",
"tests": [structured_data],
}
structured_results = StructuredData(data=test_data)
structured_results.calculate_summary_data()
return_value = handle_reports_and_summary(ctx, structured_results.structured_data, kwds=kwds)
ctx.exit(return_value)