Commit c34f5f58 authored by Mark Hewitt's avatar Mark Hewitt
Browse files

Merge branch 'feature/workflow-commands' into 'main'

API documentation pages & Workflow commands

See merge request !12
parents 74c95250 9190f1b6
Pipeline #85536 passed with stages
in 2 minutes and 54 seconds
......@@ -19,6 +19,7 @@ from click.testing import CliRunner
from osducli.__main__ import cli
from osducli.commands.legal._const import (
LEGAL_DOCUMENTATION,
LEGAL_SERVICE_NAME,
LEGAL_STATUS_PATH,
LEGAL_SWAGGER_PATH,
......@@ -40,6 +41,7 @@ class ServiceInfoTests(unittest.TestCase):
CONFIG_LEGAL_URL,
LEGAL_STATUS_PATH,
LEGAL_SWAGGER_PATH,
LEGAL_DOCUMENTATION,
)
self.assertEqual(result.exit_code, 0)
......
......@@ -19,6 +19,7 @@ from click.testing import CliRunner
from osducli.__main__ import cli
from osducli.commands.schema._const import (
SCHEMA_DOCUMENTATION,
SCHEMA_SERVICE_NAME,
SCHEMA_STATUS_PATH,
SCHEMA_SWAGGER_PATH,
......@@ -40,6 +41,7 @@ class ServiceInfoTests(unittest.TestCase):
CONFIG_SCHEMA_URL,
SCHEMA_STATUS_PATH,
SCHEMA_SWAGGER_PATH,
SCHEMA_DOCUMENTATION,
)
self.assertEqual(result.exit_code, 0)
......
......@@ -19,6 +19,7 @@ from click.testing import CliRunner
from osducli.__main__ import cli
from osducli.commands.search._const import (
SEARCH_DOCUMENTATION,
SEARCH_SERVICE_NAME,
SEARCH_STATUS_PATH,
SEARCH_SWAGGER_PATH,
......@@ -40,6 +41,7 @@ class ServiceInfoTests(unittest.TestCase):
CONFIG_SEARCH_URL,
SEARCH_STATUS_PATH,
SEARCH_SWAGGER_PATH,
SEARCH_DOCUMENTATION,
)
self.assertEqual(result.exit_code, 0)
......
......@@ -15,6 +15,7 @@
import mock
from osducli.commands.storage._const import (
STORAGE_DOCUMENTATION,
STORAGE_SERVICE_NAME,
STORAGE_STATUS_PATH,
STORAGE_SWAGGER_PATH,
......@@ -24,7 +25,7 @@ from tests.base_test_case import BaseTestCase
class ServiceInfoTests(BaseTestCase):
"""Test cases for schema info comand"""
"""Test cases for schema info command"""
# region CLI interface
......@@ -38,6 +39,7 @@ class ServiceInfoTests(BaseTestCase):
CONFIG_STORAGE_URL,
STORAGE_STATUS_PATH,
STORAGE_SWAGGER_PATH,
STORAGE_DOCUMENTATION,
)
# endregion CLI interface
......
{
"workflowId": "Osdu_ingest",
"workflowName": "Osdu_ingest",
"description": "R3 manifest processing with providing integrity",
"createdBy": "someone@womewhere.com",
"creationTimestamp": 1635143746147,
"version": 1,
"registrationInstructions": {
"dagType": "private"
}
}
\ No newline at end of file
[
{
"workflowId": "Osdu_ingest",
"workflowName": "Osdu_ingest",
"description": "R3 manifest processing with providing integrity",
"createdBy": "someone@somewhere.com",
"creationTimestamp": 1635143746147,
"version": 1,
"registrationInstructions": {}
}
]
{
"description": "description2",
"registrationInstructions": {},
"workflowName": "name2"
}
\ No newline at end of file
{
"workflowId": "test",
"workflowName": "test",
"description": "mydesc",
"createdBy": "someone@somewhere.com",
"creationTimestamp": 1641555758166,
"version": 1,
"registrationInstructions": {}
}
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Test cases for workflow get"""
import mock
from mock import patch
from nose2.tools import params
from osducli.click_cli import State
from osducli.cliclient import CliOsduClient
from osducli.commands.workflow.get import get_workflow
from tests.base_test_case import BaseTestCase
from tests.helpers import MockConfig
class StatusTests(BaseTestCase):
"""Test cases for workflow get"""
# region CLI interface
@params("name1", "name2")
def test_checkrefs_default_params(self, name: str):
"""Test cli with required parameters"""
self.assert_cli_calls_invoked_function(
["workflow", "get", "--name", name],
"osducli.commands.workflow.get.get_workflow",
mock.ANY,
name,
)
# endregion CLI interface
@params("name1", "name2")
def test_parameters(self, name): # pylint: disable=R0201
"""Test parameters"""
api_response = self.get_local_json_file("get-response.json", __file__)
expected_output = api_response
with patch(
"osducli.cliclient.CliOsduClient.cli_get_returning_json", return_value=api_response
) as mock_cli_get_returning_json:
state = State()
state.config = MockConfig
state.output = "json"
results = get_workflow(state, name)
mock_cli_get_returning_json.assert_called_once_with("workflow_url", "workflow/" + name)
self.assertDictEqual(expected_output, results)
def test_user_friendly_output(self): # pylint: disable=R0201
"""Test user friendly output"""
data = self.get_local_json_file("get-response.json", __file__)
with patch(
"osducli.cliclient.CliOsduClient.cli_get_returning_json", return_value=data
) as _:
state = State()
state.config = MockConfig
_ = get_workflow(state, "some-dummy-id")
# TO DO - Verify stdout.
@patch.object(CliOsduClient, "cli_get_returning_json", side_effect=SystemExit(1))
def test_exit_on_error(self, mock_cli_get_returning_json): # pylint: disable=W0613
"""Test any exit error is propogated"""
state = State()
state.config = MockConfig
with self.assertRaises(SystemExit) as sysexit:
_ = get_workflow(state, "some-dummy-id")
self.assertEqual(sysexit.exception.code, 1)
if __name__ == "__main__":
import nose2
nose2.main()
......@@ -10,38 +10,37 @@
# See the License for the specific language governing permissions and
# limitations under the License.
"""Test cases for service_info"""
import unittest
"""Test cases for service info"""
import mock
from click.testing import CliRunner
from osducli.__main__ import cli
from osducli.commands.workflow._const import (
WORKFLOW_SERVICE_NAME,
WORKFLOW_STATUS_PATH,
WORKFLOW_SWAGGER_PATH,
)
from osducli.config import CONFIG_WORKFLOW_URL
from tests.base_test_case import BaseTestCase
class ServiceInfoTests(BaseTestCase):
"""Test cases for status info command"""
class ServiceInfoTests(unittest.TestCase):
"""Test cases for status info comand"""
# region CLI interface
def test_info(self):
"""Test json output"""
with mock.patch("osducli.commands.workflow.info.info", return_value=None) as mock_info:
runner = CliRunner()
result = runner.invoke(cli, ["workflow", "info"])
mock_info.assert_called_once_with(
mock.ANY,
WORKFLOW_SERVICE_NAME,
CONFIG_WORKFLOW_URL,
WORKFLOW_STATUS_PATH,
WORKFLOW_SWAGGER_PATH,
)
self.assertEqual(result.exit_code, 0)
"""Test cli with required parameters"""
self.assert_cli_calls_invoked_function(
["workflow", "info"],
"osducli.commands.workflow.info.info",
mock.ANY,
WORKFLOW_SERVICE_NAME,
CONFIG_WORKFLOW_URL,
WORKFLOW_STATUS_PATH,
WORKFLOW_SWAGGER_PATH,
)
# endregion CLI interface
if __name__ == "__main__":
......
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Test cases for workflow list"""
import mock
from mock import patch
from osducli.click_cli import State
from osducli.cliclient import CliOsduClient
from osducli.commands.workflow.list import list_workflows
from tests.base_test_case import BaseTestCase
from tests.helpers import MockConfig
class StatusTests(BaseTestCase):
"""Test cases for workflow list"""
# region CLI interface
def test_checkrefs_default_params(self):
"""Test cli with required parameters"""
self.assert_cli_calls_invoked_function(
["workflow", "list"],
"osducli.commands.workflow.list.list_workflows",
mock.ANY,
)
# endregion CLI interface
def test_parameters(self): # pylint: disable=R0201
"""Test parameters"""
api_response = self.get_local_json_file("list-response.json", __file__)
expected_output = api_response
with patch(
"osducli.cliclient.CliOsduClient.cli_get_returning_json", return_value=api_response
) as mock_cli_get_returning_json:
state = State()
state.config = MockConfig
state.output = "json"
results = list_workflows(state)
mock_cli_get_returning_json.assert_called_once_with("workflow_url", "workflow?prefix=")
self.assertEqual(expected_output, results)
@patch.object(CliOsduClient, "cli_get_returning_json", side_effect=SystemExit(1))
def test_exit_on_error(self, mock_cli_get_returning_json): # pylint: disable=W0613
"""Test any exit error is propogated"""
state = State()
state.config = MockConfig
with self.assertRaises(SystemExit) as sysexit:
_ = list_workflows(state)
self.assertEqual(sysexit.exception.code, 1)
if __name__ == "__main__":
import nose2
nose2.main()
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Test cases for workflow register"""
import mock
from mock import patch
from nose2.tools import params
from osducli.click_cli import State
from osducli.cliclient import CliOsduClient
from osducli.commands.workflow.register import register_workflow
from tests.base_test_case import BaseTestCase
from tests.helpers import MockConfig
class StatusTests(BaseTestCase):
"""Test cases for workflow register"""
# region CLI interface
@params(
("name1", "description2"),
("name2", "description2"),
)
def test_checkrefs_default_params(self, name: str, description: str):
"""Test cli with required parameters"""
self.assert_cli_calls_invoked_function(
["workflow", "register", "-n", name, "-d", description],
"osducli.commands.workflow.register.register_workflow",
mock.ANY,
name,
description,
)
# endregion CLI interface
@params(
("name1", "description1"),
("name2", "description2"),
)
def test_parameters(self, name: str, description: str): # pylint: disable=R0201
"""Test parameters"""
expected_api_request = self.get_local_json_file("register-request.json", __file__)
expected_api_request["description"] = description
expected_api_request["workflowName"] = name
api_response = self.get_local_json_file("register-response.json", __file__)
expected_output = api_response
with patch(
"osducli.cliclient.CliOsduClient.cli_post_returning_json", return_value=api_response
) as mock_cli_post_returning_json:
state = State()
state.config = MockConfig
state.output = "json"
results = register_workflow(state, name, description)
mock_cli_post_returning_json.assert_called_once_with(
"workflow_url", "workflow", expected_api_request
)
self.assertEqual(expected_output, results)
@patch.object(CliOsduClient, "cli_post_returning_json", side_effect=SystemExit(1))
def test_exit_on_error(self, mock_cli_post_returning_json): # pylint: disable=W0613
"""Test any exit error is propogated"""
state = State()
state.config = MockConfig
with self.assertRaises(SystemExit) as sysexit:
_ = register_workflow(state, "name", "description")
self.assertEqual(sysexit.exception.code, 1)
if __name__ == "__main__":
import nose2
nose2.main()
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Test cases for workflow runs"""
import mock
from mock import patch
from nose2.tools import params
from osducli.click_cli import State
from osducli.cliclient import CliOsduClient
from osducli.commands.workflow.runs import runs
from tests.base_test_case import BaseTestCase
from tests.helpers import MockConfig
class StatusTests(BaseTestCase):
"""Test cases for workflow runs"""
# region CLI interface
@params("name1", "name2")
def test_checkrefs_default_params(self, name: str):
"""Test cli with required parameters"""
self.assert_cli_calls_invoked_function(
["workflow", "runs", "--name", name],
"osducli.commands.workflow.runs.runs",
mock.ANY,
name,
)
# endregion CLI interface
@params("name1", "name2")
def test_parameters(self, name): # pylint: disable=R0201
"""Test parameters"""
api_response = self.get_local_json_file("get-response.json", __file__)
expected_output = api_response
with patch(
"osducli.cliclient.CliOsduClient.cli_get_returning_json", return_value=api_response
) as mock_cli_get_returning_json:
state = State()
state.config = MockConfig
state.output = "json"
results = runs(state, name)
mock_cli_get_returning_json.assert_called_once_with(
"workflow_url", f"workflow/{name}/workflowRun"
)
self.assertDictEqual(expected_output, results)
def test_user_friendly_output(self): # pylint: disable=R0201
"""Test user friendly output"""
data = self.get_local_json_file("get-response.json", __file__)
with patch(
"osducli.cliclient.CliOsduClient.cli_get_returning_json", return_value=data
) as _:
state = State()
state.config = MockConfig
_ = runs(state, "some-dummy-id")
# TO DO - Verify stdout.
@patch.object(CliOsduClient, "cli_get_returning_json", side_effect=SystemExit(1))
def test_exit_on_error(self, mock_cli_get_returning_json): # pylint: disable=W0613
"""Test any exit error is propogated"""
state = State()
state.config = MockConfig
with self.assertRaises(SystemExit) as sysexit:
_ = runs(state, "some-dummy-id")
self.assertEqual(sysexit.exception.code, 1)
if __name__ == "__main__":
import nose2
nose2.main()
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Test cases for workflow status"""
import mock
from nose2.tools import params
from tests.base_test_case import BaseTestCase
class StatusTests(BaseTestCase):
"""Test cases for workflow status"""
# region CLI interface
@params("runid1", "runid2")
def test_checkrefs_default_params(self, runid: str):
"""Test cli with required parameters"""
self.assert_cli_calls_invoked_function(
["workflow", "status", "-r", runid],
"osducli.commands.workflow.status.status",
mock.ANY,
runid,
None,
False,
)
# endregion CLI interface
if __name__ == "__main__":
import nose2
nose2.main()
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Test cases for workflow register"""
import mock
from mock import patch
from nose2.tools import params
from osducli.click_cli import State
from osducli.cliclient import CliOsduClient
from osducli.commands.workflow.unregister import unregister_workflow
from tests.base_test_case import BaseTestCase
from tests.helpers import MockConfig
class StatusTests(BaseTestCase):
"""Test cases for workflow register"""
# region CLI interface
@params("name1", "name2")
def test_checkrefs_default_params(self, name: str):
"""Test cli with required parameters"""
self.assert_cli_calls_invoked_function(
["workflow", "unregister", "-n", name],
"osducli.commands.workflow.unregister.unregister_workflow",