Restore temporarily bco API test

This commit is contained in:
davelopez
2022-10-14 12:20:24 +02:00
parent e75923fce3
commit ffd29f01b7
2 changed files with 45 additions and 0 deletions
+9
View File
@@ -2205,6 +2205,15 @@ steps:
assert "## Workflow Inputs" in markdown_content
assert "## About This Report" in markdown_content
@skip_without_tool("cat1")
def test_export_invocation_bco(self):
with self.dataset_populator.test_history() as history_id:
summary = self._run_workflow(WORKFLOW_SIMPLE, test_data={"input1": "hello world"}, history_id=history_id)
invocation_id = summary.invocation_id
bco = self.workflow_populator.get_biocompute_object(invocation_id)
self.workflow_populator.validate_biocompute_object(bco)
assert bco["provenance_domain"]["name"] == "Simple Workflow"
@skip_without_tool("__APPLY_RULES__")
def test_workflow_run_apply_rules(self):
with self.dataset_populator.test_history() as history_id:
+36
View File
@@ -1547,6 +1547,42 @@ class BaseWorkflowPopulator(BasePopulator):
api_asserts.assert_status_code_is_ok(create_response)
return create_response.json()
def get_biocompute_object(self, invocation_id):
bco_response = self._get(f"invocations/{invocation_id}/biocompute")
bco_response.raise_for_status()
return bco_response.json()
def validate_biocompute_object(
self, bco, expected_schema_version="https://w3id.org/ieee/ieee-2791-schema/2791object.json"
):
# TODO: actually use jsonref and jsonschema to validate this someday
api_asserts.assert_has_keys(
bco,
"object_id",
"spec_version",
"etag",
"provenance_domain",
"usability_domain",
"description_domain",
"execution_domain",
"parametric_domain",
"io_domain",
"error_domain",
)
assert bco["spec_version"] == expected_schema_version
api_asserts.assert_has_keys(bco["description_domain"], "keywords", "xref", "platform", "pipeline_steps")
api_asserts.assert_has_keys(
bco["execution_domain"],
"script",
"script_driver",
"software_prerequisites",
"external_data_endpoints",
"environment_variables",
)
for p in bco["parametric_domain"]:
api_asserts.assert_has_keys(p, "param", "value", "step")
api_asserts.assert_has_keys(bco["io_domain"], "input_subdomain", "output_subdomain")
def invoke_workflow_raw(self, workflow_id: str, request: dict, assert_ok: bool = False) -> Response:
url = f"workflows/{workflow_id}/invocations"
invocation_response = self._post(url, data=request)