Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_get_run(self, mock_wes_client):
mock_run_log = {'run_id': 'foo',
'request': {},
'state': '',
'run_log': {},
'task_logs': [],
'outputs': {}}
mock_response = BravadoResponseMock(result=mock_run_log)
mock_wes_client.GetRunLog.return_value.response = mock_response
wes_instance = WES(wes_id='mock_wes',
api_client=mock_wes_client)
test_run_log = wes_instance.get_run(id='foo')
assert isinstance(test_run_log, dict)
assert test_run_log == mock_run_log
def test_get_run_status(self, mock_wes_client):
mock_run_status = {'run_id': 'foo',
'state': ''}
mock_response = BravadoResponseMock(result=mock_run_status)
mock_wes_client.GetRunStatus.return_value.response = mock_response
wes_instance = WES(wes_id='mock_wes',
api_client=mock_wes_client)
test_run_status = wes_instance.get_run_status(id='foo')
assert isinstance(test_run_status, dict)
assert test_run_status == mock_run_status
def test_get_metadata(self, mock_trs_client):
mock_metadata = {
'version': '1.0.0',
'api_version': '1.0.0',
'country': '',
'friendly_name': 'mock_trs'
}
mock_response = BravadoResponseMock(result=mock_metadata)
mock_trs_client.metadataGet.return_value.response = mock_response
trs_instance = TRS(trs_id='mock_trs', api_client=mock_trs_client)
test_metadata = trs_instance.get_metadata()
assert isinstance(test_metadata, dict)
assert test_metadata == mock_metadata
def test_get_workflow_descriptor(self, mock_trs_client):
mock_workflow_descriptor = {'content': '', 'url': ''}
mock_response = BravadoResponseMock(result=mock_workflow_descriptor)
operator = mock_trs_client.toolsIdVersionsVersionIdTypeDescriptorGet
operator.return_value.response = mock_response
trs_instance = TRS(trs_id='mock_trs', api_client=mock_trs_client)
test_workflow_descriptor = trs_instance.get_workflow_descriptor(
id='mock_wf',
version_id='test',
type='CWL'
)
assert isinstance(test_workflow_descriptor, dict)
assert test_workflow_descriptor == mock_workflow_descriptor
def test_get_workflow(self, mock_trs_client):
mock_workflow = {
'url': '',
'id': 'mock_wf',
'organization': '',
'author': '',
'toolclass': {},
'versions': []
}
mock_response = BravadoResponseMock(result=mock_workflow)
mock_trs_client.toolsIdGet.return_value.response = mock_response
trs_instance = TRS(trs_id='mock_trs', api_client=mock_trs_client)
test_workflow = trs_instance.get_workflow('mock_wf')
assert isinstance(test_workflow, dict)
assert test_workflow == mock_workflow
def test_get_workflow_descriptor_relative(self, mock_trs_client):
mock_workflow_descriptor = {'content': '', 'url': ''}
mock_response = BravadoResponseMock(result=mock_workflow_descriptor)
operator = mock_trs_client.toolsIdVersionsVersionIdTypeDescriptorRelativePathGet
operator.return_value.response = mock_response
trs_instance = TRS(trs_id='mock_trs', api_client=mock_trs_client)
test_workflow_descriptor = trs_instance.get_workflow_descriptor_relative(
id='mock_wf',
version_id='test',
type='CWL',
relative_path=''
)
assert isinstance(test_workflow_descriptor, dict)
assert test_workflow_descriptor == mock_workflow_descriptor
def test_list_runs(self, mock_wes_client):
mock_runs = ['foo', 'bar']
mock_response = BravadoResponseMock(result=mock_runs)
mock_wes_client.ListRuns.return_value.response = mock_response
wes_instance = WES(wes_id='mock_wes',
api_client=mock_wes_client)
test_runs = wes_instance.list_runs()
assert isinstance(test_runs, list)
assert test_runs == mock_runs
def test_run_workflow(self, mock_wes_client):
mock_run_id = {'run_id': 'foo'}
mock_response = BravadoResponseMock(result=mock_run_id)
mock_wes_client.RunWorkflow.return_value.response = mock_response
wes_instance = WES(wes_id='mock_wes',
api_client=mock_wes_client)
test_run_id = wes_instance.run_workflow(request={})
assert isinstance(test_run_id, dict)
assert test_run_id == mock_run_id
def test_get_workflow_versions(self, mock_trs_client):
mock_workflow_versions = [
{'url': '', 'id': ''},
{'url': '', 'id': ''}
]
mock_response = BravadoResponseMock(result=mock_workflow_versions)
operator = mock_trs_client.toolsIdVersionsGet
operator.return_value.response = mock_response
trs_instance = TRS(trs_id='mock_trs', api_client=mock_trs_client)
test_workflow_versions = trs_instance.get_workflow_versions(
id='mock_wf'
)
assert isinstance(test_workflow_versions, list)
assert test_workflow_versions == mock_workflow_versions
def test_get_service_info_bravado(self, mock_wes_client):
mock_service_info = {'workflow_type_versions': ['CWL', 'WDL']}
mock_response = BravadoResponseMock(result=mock_service_info)
mock_wes_client.GetServiceInfo.return_value.response = mock_response
wes_instance = WES(wes_id='mock_wes',
api_client=mock_wes_client)
test_service_info = wes_instance.get_service_info()
assert isinstance(test_service_info, dict)
assert test_service_info == mock_service_info