openstack/python-magnumclient
Code Issues Proposed changes

Split v1 shell sub-command into specific files

Browse Source 
v1 shell has many sub command. It is difficult to add more test
and functions. Actually we have forgotten to add some tests for
these sub-commands.
This patch splits these.

Change-Id: I6b80b621311442df987e6ff031e8c77cf4bb7095
Closes-Bug: #1515109
This commit is contained in:
OTSUKA, Yuanying 2015-11-11 14:32:10 +09:00
parent ba92b81215
commit 8480e0b0f1
23 changed files with 2126 additions and 1595 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
magnumclient/common/utils.py
View File

@ -107,3 +107,7 @@ def format_labels(lbls, parse_comma=True):
labels[k].append(v)
return labels
def print_list_field(field):
return lambda obj: ', '.join(getattr(obj, field))

18
magnumclient/shell.py
View File

@ -52,7 +52,7 @@ except ImportError:
from magnumclient.openstack.common.apiclient import auth
from magnumclient.openstack.common.apiclient import exceptions as exc
from magnumclient.openstack.common import cliutils
from magnumclient.v1 import client
from magnumclient.v1 import client as client_v1
from magnumclient.v1 import shell as shell_v1
from magnumclient import version
@ -335,13 +335,14 @@ class OpenStackMagnumShell(object):
subparsers = parser.add_subparsers(metavar='<subcommand>')
try:
actions_module = {
'1': shell_v1,
actions_modules = {
'1': shell_v1.COMMAND_MODULES,
}[version]
except KeyError:
actions_module = shell_v1
actions_modules = shell_v1.COMMAND_MODULES
self._find_actions(subparsers, actions_module)
for actions_module in actions_modules:
self._find_actions(subparsers, actions_module)
self._find_actions(subparsers, self)
self._add_bash_completion_subparser(subparsers)
@ -508,6 +509,13 @@ class OpenStackMagnumShell(object):
'--os-password, env[OS_PASSWORD], or '
'prompted response')
try:
client = {
'1': client_v1,
}[options.magnum_api_version]
except KeyError:
client = client_v1
self.cs = client.Client(username=os_username,
api_key=os_password,
project_id=os_tenant_id,

71
magnumclient/tests/v1/shell_test_base.py Normal file
View File

@ -0,0 +1,71 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import re
import mock
from testtools import matchers
from magnumclient.tests import utils
FAKE_ENV = {'OS_USERNAME': 'username',
'OS_PASSWORD': 'password',
'OS_TENANT_NAME': 'tenant_name',
'OS_AUTH_URL': 'http://no.where/v2.0',
'BYPASS_URL': 'http://magnum'}
class TestCommandLineArgument(utils.TestCase):
_unrecognized_arg_error = [
'.*?^usage: ',
'.*?^error: unrecognized arguments:',
".*?^Try 'magnum help ' for more information.",
]
_mandatory_arg_error = [
'.*?^usage: ',
'.*?^error: (the following arguments|argument)',
".*?^Try 'magnum help ",
]
_few_argument_error = [
'.*?^usage: magnum ',
'.*?^error: (the following arguments|too few arguments)',
".*?^Try"
]
def setUp(self):
super(TestCommandLineArgument, self).setUp()
self.make_env(fake_env=FAKE_ENV)
session_client = mock.patch(
'magnumclient.common.httpclient.SessionClient')
session_client.start()
loader = mock.patch('keystoneauth1.loading.get_plugin_loader')
loader.start()
session = mock.patch('keystoneauth1.session.Session')
session.start()
self.addCleanup(session_client.stop)
self.addCleanup(loader.stop)
self.addCleanup(session.stop)
def _test_arg_success(self, command):
stdout, stderr = self.shell(command)
def _test_arg_failure(self, command, error_msg):
stdout, stderr = self.shell(command, (2,))
for line in error_msg:
self.assertThat((stdout + stderr),
matchers.MatchesRegex(line,
re.DOTALL | re.MULTILINE))

243
magnumclient/tests/v1/test_baymodels_shell.py Normal file
View File

@ -0,0 +1,243 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import mock
from magnumclient.tests.v1 import shell_test_base
class ShellTest(shell_test_base.TestCommandLineArgument):
@mock.patch('magnumclient.v1.baymodels.BayModelManager.create')
def test_baymodel_create_success(self, mock_create):
self._test_arg_success('baymodel-create '
'--name test '
'--image-id test_image '
'--keypair-id test_keypair '
'--external-network-id test_net '
'--coe swarm '
'--dns-nameserver test_dns '
'--flavor-id test_flavor '
'--fixed-network public '
'--network-driver test_driver '
'--labels key=val '
'--master-flavor-id test_flavor '
'--docker-volume-size 10'
'--public')
self.assertTrue(mock_create.called)
self._test_arg_success('baymodel-create '
'--keypair-id test_keypair '
'--external-network-id test_net '
'--image-id test_image '
'--coe kubernetes '
'--name test ')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.create')
def test_baymodel_create_public_success(self, mock_create):
self._test_arg_success('baymodel-create '
'--name test --network-driver test_driver '
'--keypair-id test_keypair '
'--external-network-id test_net '
'--image-id test_image '
'--coe swarm'
'--public')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.create')
def test_baymodel_create_success_with_master_flavor(self, mock_create):
self._test_arg_success('baymodel-create '
'--name test '
'--image-id test_image '
'--keypair-id test_keypair '
'--external-network-id test_net '
'--coe swarm '
'--dns-nameserver test_dns '
'--master-flavor-id test_flavor')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.create')
def test_baymodel_create_docker_vol_size_success(self, mock_create):
self._test_arg_success('baymodel-create '
'--name test --docker-volume-size 4514 '
'--keypair-id test_keypair '
'--external-network-id test_net '
'--image-id test_image '
'--coe swarm'
)
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.create')
def test_baymodel_create_fixed_network_success(self, mock_create):
self._test_arg_success('baymodel-create '
'--name test --fixed-network private '
'--keypair-id test_keypair '
'--external-network-id test_net '
'--image-id test_image '
'--coe swarm')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.create')
def test_baymodel_create_network_driver_success(self, mock_create):
self._test_arg_success('baymodel-create '
'--name test --network-driver test_driver '
'--keypair-id test_keypair '
'--external-network-id test_net '
'--image-id test_image '
'--coe swarm')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.create')
def test_baymodel_create_ssh_authorized_key_success(self, mock_create):
self._test_arg_success('baymodel-create '
'--name test '
'--keypair-id test_keypair '
'--external-network-id test_net '
'--image-id test_image '
'--coe swarm '
'--ssh-authorized-key test_key '
)
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.create')
def test_baymodel_create_http_proxy_success(self, mock_create):
self._test_arg_success('baymodel-create '
'--name test --fixed-network private '
'--keypair-id test_keypair '
'--external-network-id test_net '
'--image-id test_image '
'--coe swarm '
'--http-proxy http_proxy ')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.create')
def test_baymodel_create_https_proxy_success(self, mock_create):
self._test_arg_success('baymodel-create '
'--name test --fixed-network private '
'--keypair-id test_keypair '
'--external-network-id test_net '
'--image-id test_image '
'--coe swarm '
'--https-proxy https_proxy ')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.create')
def test_baymodel_create_no_proxy_success(self, mock_create):
self._test_arg_success('baymodel-create '
'--name test --fixed-network private '
'--keypair-id test_keypair '
'--external-network-id test_net '
'--image-id test_image '
'--coe swarm '
'--no-proxy no_proxy ')
@mock.patch('magnumclient.v1.baymodels.BayModelManager.create')
def test_baymodel_create_labels_success(self, mock_create):
self._test_arg_success('baymodel-create '
'--name test '
'--labels key=val '
'--keypair-id test_keypair '
'--external-network-id test_net '
'--image-id test_image '
'--coe swarm')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.create')
def test_baymodel_create_separate_labels_success(self, mock_create):
self._test_arg_success('baymodel-create '
'--name test '
'--labels key1=val1 '
'--labels key2=val2 '
'--keypair-id test_keypair '
'--external-network-id test_net '
'--image-id test_image '
'--coe swarm')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.create')
def test_baymodel_create_combined_labels_success(self, mock_create):
self._test_arg_success('baymodel-create '
'--name test '
'--labels key1=val1,key2=val2 '
'--keypair-id test_keypair '
'--external-network-id test_net '
'--image-id test_image '
'--coe swarm')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.create')
def test_baymodel_create_failure_few_arg(self, mock_create):
self._test_arg_failure('baymodel-create '
'--name test', self._mandatory_arg_error)
self.assertFalse(mock_create.called)
self._test_arg_failure('baymodel-create '
'--image-id test', self._mandatory_arg_error)
self.assertFalse(mock_create.called)
self._test_arg_failure('baymodel-create '
'--keypair-id test', self._mandatory_arg_error)
self.assertFalse(mock_create.called)
self._test_arg_failure('baymodel-create '
'--external-network-id test',
self._mandatory_arg_error)
self.assertFalse(mock_create.called)
self._test_arg_failure('baymodel-create '
'--coe test', self._mandatory_arg_error)
self.assertFalse(mock_create.called)
self._test_arg_failure('baymodel-create', self._mandatory_arg_error)
self.assertFalse(mock_create.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.get')
def test_baymodel_show_success(self, mock_show):
self._test_arg_success('baymodel-show xxx')
self.assertTrue(mock_show.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.get')
def test_baymodel_show_failure_no_arg(self, mock_show):
self._test_arg_failure('baymodel-show', self._few_argument_error)
self.assertFalse(mock_show.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.delete')
def test_baymodel_delete_success(self, mock_delete):
self._test_arg_success('baymodel-delete xxx')
self.assertTrue(mock_delete.called)
self.assertEqual(1, mock_delete.call_count)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.delete')
def test_baymodel_delete_multiple_id_success(self, mock_delete):
self._test_arg_success('baymodel-delete xxx xyz')
self.assertTrue(mock_delete.called)
self.assertEqual(2, mock_delete.call_count)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.delete')
def test_baymodel_delete_failure_no_arg(self, mock_delete):
self._test_arg_failure('baymodel-delete', self._few_argument_error)
self.assertFalse(mock_delete.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.list')
def test_baymodel_list_success(self, mock_list):
self._test_arg_success('baymodel-list')
self.assertTrue(mock_list.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.list')
def test_baymodel_list_failure(self, mock_list):
self._test_arg_failure('baymodel-list --wrong',
self._unrecognized_arg_error)
self.assertFalse(mock_list.called)

158
magnumclient/tests/v1/test_bays_shell.py Normal file
View File

@ -0,0 +1,158 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import mock
from magnumclient import exceptions
from magnumclient.tests.v1 import shell_test_base
class ShellTest(shell_test_base.TestCommandLineArgument):
@mock.patch('magnumclient.v1.bays.BayManager.list')
def test_bay_list_success(self, mock_list):
self._test_arg_success('bay-list')
self.assertTrue(mock_list.called)
@mock.patch('magnumclient.v1.bays.BayManager.list')
def test_bay_list_failure(self, mock_list):
self._test_arg_failure('bay-list --wrong',
self._unrecognized_arg_error)
self.assertFalse(mock_list.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.get')
@mock.patch('magnumclient.v1.bays.BayManager.create')
def test_bay_create_success(self, mock_create, mock_get):
self._test_arg_success('bay-create --name test --baymodel xxx '
'--node-count 123 --timeout 15')
self.assertTrue(mock_create.called)
self._test_arg_success('bay-create --baymodel xxx')
self.assertTrue(mock_create.called)
self._test_arg_success('bay-create --name test --baymodel xxx')
self.assertTrue(mock_create.called)
self._test_arg_success('bay-create --baymodel xxx --node-count 123')
self.assertTrue(mock_create.called)
self._test_arg_success('bay-create --baymodel xxx --node-count 123 '
'--master-count 123')
self.assertTrue(mock_create.called)
self._test_arg_success('bay-create --baymodel xxx '
'--timeout 15')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.baymodels.BayModelManager.get')
@mock.patch('magnumclient.v1.bays.BayManager.create')
def test_bay_create_success_only_baymodel_arg(self, mock_create, mock_get):
self._test_arg_success('bay-create --baymodel xxx')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.bays.BayManager.create')
def test_bay_create_failure_only_name(self, mock_create):
self._test_arg_failure('bay-create --name test',
self._mandatory_arg_error)
self.assertFalse(mock_create.called)
@mock.patch('magnumclient.v1.bays.BayManager.create')
def test_bay_create_failure_only_node_count(self, mock_create):
self._test_arg_failure('bay-create --node-count 1',
self._mandatory_arg_error)
self.assertFalse(mock_create.called)
@mock.patch('magnumclient.v1.bays.BayManager.create')
def test_bay_create_failure_only_bay_create_timeout(self, mock_create):
self._test_arg_failure('bay-create --timeout 15',
self._mandatory_arg_error)
self.assertFalse(mock_create.called)
@mock.patch('magnumclient.v1.bays.BayManager.create')
def test_bay_create_failure_no_arg(self, mock_create):
self._test_arg_failure('bay-create',
self._mandatory_arg_error)
self.assertFalse(mock_create.called)
@mock.patch('magnumclient.v1.bays.BayManager.delete')
def test_bay_delete_success(self, mock_delete):
self._test_arg_success('bay-delete xxx')
self.assertTrue(mock_delete.called)
self.assertEqual(1, mock_delete.call_count)
@mock.patch('magnumclient.v1.bays.BayManager.delete')
def test_bay_delete_multiple_id_success(self, mock_delete):
self._test_arg_success('bay-delete xxx xyz')
self.assertTrue(mock_delete.called)
self.assertEqual(2, mock_delete.call_count)
@mock.patch('magnumclient.v1.bays.BayManager.delete')
def test_bay_delete_failure_no_arg(self, mock_delete):
self._test_arg_failure('bay-delete', self._few_argument_error)
self.assertFalse(mock_delete.called)
@mock.patch('magnumclient.v1.bays.BayManager.get')
def test_bay_show_success(self, mock_show):
self._test_arg_success('bay-show xxx')
self.assertTrue(mock_show.called)
@mock.patch('magnumclient.v1.bays.BayManager.get')
def test_bay_show_failure_no_arg(self, mock_show):
self._test_arg_failure('bay-show', self._few_argument_error)
self.assertFalse(mock_show.called)
@mock.patch('magnumclient.v1.bays.BayManager.update')
def test_bay_update_success(self, mock_update):
self._test_arg_success('bay-update test add test=test')
self.assertTrue(mock_update.called)
@mock.patch('magnumclient.v1.bays.BayManager.update')
def test_bay_update_success_many_attribute(self, mock_update):
self._test_arg_success('bay-update test add test=test test1=test1')
self.assertTrue(mock_update.called)
@mock.patch('magnumclient.v1.bays.BayManager.update')
def test_bay_update_failure_wrong_op(self, mock_update):
_error_msg = [
'.*?^usage: magnum bay-update ',
'.*?^error: argument <op>: invalid choice: ',
".*?^Try 'magnum help bay-update' for more information."
]
self._test_arg_failure('bay-update test wrong test=test', _error_msg)
self.assertFalse(mock_update.called)
@mock.patch('magnumclient.v1.bays.BayManager.update')
def test_bay_update_failure_wrong_attribute(self, mock_update):
_error_msg = [
'.*?^ERROR: Attributes must be a list of PATH=VALUE'
]
self.assertRaises(exceptions.CommandError, self._test_arg_failure,
'bay-update test add test', _error_msg)
self.assertFalse(mock_update.called)
@mock.patch('magnumclient.v1.bays.BayManager.update')
def test_bay_update_failure_few_args(self, mock_update):
_error_msg = [
'.*?^usage: magnum bay-update ',
'.*?^error: (the following arguments|too few arguments)',
".*?^Try 'magnum help bay-update' for more information."
]
self._test_arg_failure('bay-update', _error_msg)
self.assertFalse(mock_update.called)
self._test_arg_failure('bay-update test', _error_msg)
self.assertFalse(mock_update.called)
self._test_arg_failure('bay-update test add', _error_msg)
self.assertFalse(mock_update.called)

99
magnumclient/tests/v1/test_certificates_shell.py Normal file
View File

@ -0,0 +1,99 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import mock
from magnumclient.tests.v1 import shell_test_base
from magnumclient.v1 import certificates_shell
class ShellTest(shell_test_base.TestCommandLineArgument):
@mock.patch('magnumclient.v1.bays.BayManager.get')
@mock.patch('magnumclient.v1.certificates.CertificateManager.get')
def test_ca_show_success(self, mock_cert_get, mock_bay_get):
mockbay = mock.MagicMock()
mockbay.status = "CREATE_COMPLETE"
mock_bay_get.return_value = mockbay
self._test_arg_success('ca-show '
'--bay xxx')
self.assertTrue(mock_cert_get.called)
@mock.patch('magnumclient.v1.bays.BayManager.get')
@mock.patch('magnumclient.v1.certificates.CertificateManager.get')
def test_ca_show_with_wrong_status(self, mock_cert_get, mock_bay_get):
mockbay = mock.MagicMock()
mockbay.status = "FAILED"
mock_bay_get.return_value = mockbay
self._test_arg_success('ca-show '
'--bay xxx')
self.assertFalse(mock_cert_get.called)
@mock.patch('os.path.isfile')
@mock.patch('magnumclient.v1.bays.BayManager.get')
@mock.patch('magnumclient.v1.certificates.CertificateManager.create')
def test_ca_sign_success(
self, mock_cert_create, mock_bay_get, mock_isfile):
mock_isfile.return_value = True
mockbay = mock.MagicMock()
mockbay.status = "CREATE_COMPLETE"
mock_bay_get.return_value = mockbay
fake_csr = 'fake-csr'
file_mock = mock.mock_open(read_data=fake_csr)
with mock.patch.object(certificates_shell, 'open', file_mock):
self._test_arg_success('ca-sign '
'--csr path/csr.pem '
'--bay xxx')
self.assertTrue(mock_cert_create.called)
@mock.patch('os.path.isfile')
@mock.patch('magnumclient.v1.bays.BayManager.get')
@mock.patch('magnumclient.v1.certificates.CertificateManager.create')
def test_ca_sign_with_wrong_status(
self, mock_cert_create, mock_bay_get, mock_isfile):
mock_isfile.return_value = True
mockbay = mock.MagicMock()
mockbay.status = "FAILED"
mock_bay_get.return_value = mockbay
fake_csr = 'fake-csr'
file_mock = mock.mock_open(read_data=fake_csr)
with mock.patch.object(certificates_shell, 'open', file_mock):
self._test_arg_success('ca-sign '
'--csr path/csr.pem '
'--bay xxx')
self.assertFalse(mock_isfile.called)
self.assertFalse(file_mock.called)
self.assertFalse(mock_cert_create.called)
@mock.patch('os.path.isfile')
@mock.patch('magnumclient.v1.bays.BayManager.get')
@mock.patch('magnumclient.v1.certificates.CertificateManager.create')
def test_ca_sign_with_not_csr(
self, mock_cert_create, mock_bay_get, mock_isfile):
mock_isfile.return_value = False
mockbay = mock.MagicMock()
mockbay.status = "CREATE_COMPLETE"
mock_bay_get.return_value = mockbay
fake_csr = 'fake-csr'
file_mock = mock.mock_open(read_data=fake_csr)
with mock.patch.object(certificates_shell, 'open', file_mock):
self._test_arg_success('ca-sign '
'--csr path/csr.pem '
'--bay xxx')
mock_isfile.assert_called_once_with('path/csr.pem')
self.assertFalse(file_mock.called)
self.assertFalse(mock_cert_create.called)

197
magnumclient/tests/v1/test_containers_shell.py Normal file
View File

@ -0,0 +1,197 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import mock
from magnumclient.tests.v1 import shell_test_base
class ShellTest(shell_test_base.TestCommandLineArgument):
@mock.patch('magnumclient.v1.containers.ContainerManager.list')
def test_container_list_success(self, mock_list):
self._test_arg_success('container-list')
self.assertTrue(mock_list.called)
@mock.patch('magnumclient.v1.containers.ContainerManager.list')
def test_container_list_failure(self, mock_list):
self._test_arg_failure('container-list --wrong',
self._unrecognized_arg_error)
self.assertFalse(mock_list.called)
@mock.patch('magnumclient.v1.bays.BayManager.get')
@mock.patch('magnumclient.v1.containers.ContainerManager.create')
def test_container_create_success(self, mock_create, mock_bay_get):
mock_bay = mock.MagicMock()
mock_bay.status = "CREATE_COMPLETE"
mock_bay_get.return_value = mock_bay
self._test_arg_success('container-create '
'--image test-image '
'--bay test-bay')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.bays.BayManager.get')
@mock.patch('magnumclient.v1.containers.ContainerManager.create')
def test_container_create_failure_without_image(self, mock_create,
mock_bay_get):
mock_bay = mock.MagicMock()
mock_bay.status = "CREATE_COMPLETE"
mock_bay_get.return_value = mock_bay
self._test_arg_failure('container-create '
'--bay test-bay',
self._mandatory_arg_error)
self.assertFalse(mock_create.called)
@mock.patch('magnumclient.v1.containers.ContainerManager.delete')
def test_container_delete_success(self, mock_delete):
self._test_arg_success('container-delete xxx')
self.assertTrue(mock_delete.called)
self.assertEqual(1, mock_delete.call_count)
@mock.patch('magnumclient.v1.containers.ContainerManager.delete')
def test_container_delete_multiple_id_success(self, mock_delete):
self._test_arg_success('container-delete xxx xyz')
self.assertTrue(mock_delete.called)
self.assertEqual(2, mock_delete.call_count)
@mock.patch('magnumclient.v1.containers.ContainerManager.delete')
def test_container_delete_failure_no_arg(self, mock_delete):
self._test_arg_failure('container-delete', self._few_argument_error)
self.assertFalse(mock_delete.called)
@mock.patch('magnumclient.v1.containers.ContainerManager.get')
def test_container_show_success(self, mock_show):
self._test_arg_success('container-show xxx')
self.assertTrue(mock_show.called)
@mock.patch('magnumclient.v1.containers.ContainerManager.get')
def test_container_show_failure_no_arg(self, mock_show):
self._test_arg_failure('container-show', self._few_argument_error)
self.assertFalse(mock_show.called)
@mock.patch('magnumclient.v1.containers.ContainerManager.reboot')
def test_container_reboot_success(self, mock_reboot):
self._test_arg_success('container-reboot xxx')
self.assertTrue(mock_reboot.called)
self.assertEqual(1, mock_reboot.call_count)
@mock.patch('magnumclient.v1.containers.ContainerManager.reboot')
def test_container_reboot_multiple_id_success(self, mock_reboot):
self._test_arg_success('container-reboot xxx xyz')
self.assertTrue(mock_reboot.called)
self.assertEqual(2, mock_reboot.call_count)
@mock.patch('magnumclient.v1.containers.ContainerManager.reboot')
def test_container_reboot_failure_no_arg(self, mock_reboot):
self._test_arg_failure('container-reboot', self._few_argument_error)
self.assertFalse(mock_reboot.called)
@mock.patch('magnumclient.v1.containers.ContainerManager.stop')
def test_container_stop_success(self, mock_stop):
self._test_arg_success('container-stop xxx')
self.assertTrue(mock_stop.called)
self.assertEqual(1, mock_stop.call_count)
@mock.patch('magnumclient.v1.containers.ContainerManager.stop')
def test_container_stop_multiple_id_success(self, mock_stop):
self._test_arg_success('container-stop xxx xyz')
self.assertTrue(mock_stop.called)
self.assertEqual(2, mock_stop.call_count)
@mock.patch('magnumclient.v1.containers.ContainerManager.stop')
def test_container_stop_failure_no_arg(self, mock_stop):
self._test_arg_failure('container-stop', self._few_argument_error)
self.assertFalse(mock_stop.called)
@mock.patch('magnumclient.v1.containers.ContainerManager.start')
def test_container_start_success(self, mock_start):
self._test_arg_success('container-start xxx')
self.assertTrue(mock_start.called)
self.assertEqual(1, mock_start.call_count)
@mock.patch('magnumclient.v1.containers.ContainerManager.start')
def test_container_start_multiple_id_success(self, mock_start):
self._test_arg_success('container-start xxx xyz')
self.assertTrue(mock_start.called)
self.assertEqual(2, mock_start.call_count)
@mock.patch('magnumclient.v1.containers.ContainerManager.start')
def test_container_start_failure_no_arg(self, mock_start):
self._test_arg_failure('container-start', self._few_argument_error)
self.assertFalse(mock_start.called)
@mock.patch('magnumclient.v1.containers.ContainerManager.pause')
def test_container_pause_success(self, mock_pause):
self._test_arg_success('container-pause xxx')
self.assertTrue(mock_pause.called)
self.assertEqual(1, mock_pause.call_count)
@mock.patch('magnumclient.v1.containers.ContainerManager.pause')
def test_container_multiple_id_pause_success(self, mock_pause):
self._test_arg_success('container-pause xxx xyz')
self.assertTrue(mock_pause.called)
self.assertEqual(2, mock_pause.call_count)
@mock.patch('magnumclient.v1.containers.ContainerManager.pause')
def test_container_pause_failure_no_arg(self, mock_pause):
self._test_arg_failure('container-pause', self._few_argument_error)
self.assertFalse(mock_pause.called)
@mock.patch('magnumclient.v1.containers.ContainerManager.unpause')
def test_container_unpause_success(self, mock_unpause):
self._test_arg_success('container-unpause xxx')
self.assertTrue(mock_unpause.called)
self.assertEqual(1, mock_unpause.call_count)
@mock.patch('magnumclient.v1.containers.ContainerManager.unpause')
def test_container_unpause_multiple_id_success(self, mock_unpause):
self._test_arg_success('container-unpause xxx xyz')
self.assertTrue(mock_unpause.called)
self.assertEqual(2, mock_unpause.call_count)
@mock.patch('magnumclient.v1.containers.ContainerManager.unpause')
def test_container_unpause_failure_no_arg(self, mock_unpause):
self._test_arg_failure('container-unpause', self._few_argument_error)
self.assertFalse(mock_unpause.called)
@mock.patch('magnumclient.v1.containers.ContainerManager.logs')
def test_container_logs_success(self, mock_logs):
self._test_arg_success('container-logs xxx')
self.assertTrue(mock_logs.called)
@mock.patch('magnumclient.v1.containers.ContainerManager.logs')
def test_container_logs_failure_no_arg(self, mock_logs):
self._test_arg_failure('container-logs', self._few_argument_error)
self.assertFalse(mock_logs.called)
@mock.patch('magnumclient.v1.containers.ContainerManager.execute')
def test_container_execute_success(self, mock_execute):
self._test_arg_success('container-exec xxx '
'--command ls')
self.assertTrue(mock_execute.called)
@mock.patch('magnumclient.v1.containers.ContainerManager.execute')
def test_container_execute_failure_no_option(self, mock_execute):
self._test_arg_failure('container-exec xxx',
self._mandatory_arg_error)
self.assertFalse(mock_execute.called)
self._test_arg_failure('container-exec --command ls',
self._few_argument_error)
self.assertFalse(mock_execute.called)
@mock.patch('magnumclient.v1.containers.ContainerManager.execute')
def test_container_execute_failure_no_arg(self, mock_execute):
self._test_arg_failure('container-exec', self._few_argument_error)
self.assertFalse(mock_execute.called)

31
magnumclient/tests/v1/test_mservices_shell.py Normal file
View File

@ -0,0 +1,31 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import mock
from magnumclient.tests.v1 import shell_test_base
class ShellTest(shell_test_base.TestCommandLineArgument):
@mock.patch('magnumclient.v1.mservices.MServiceManager.list')
def test_magnum_service_list_success(self, mock_list):
self._test_arg_success('service-list')
self.assertTrue(mock_list.called)
@mock.patch('magnumclient.v1.mservices.MServiceManager.list')
def test_magnum_service_list_failure(self, mock_list):
self._test_arg_failure('service-list --wrong',
self._unrecognized_arg_error)
self.assertFalse(mock_list.called)

38
magnumclient/tests/v1/test_nodes_shell.py Normal file
View File

@ -0,0 +1,38 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import mock
from magnumclient.tests.v1 import shell_test_base
class ShellTest(shell_test_base.TestCommandLineArgument):
@mock.patch('magnumclient.v1.nodes.NodeManager.list')
def test_node_list_success(self, mock_list):
self._test_arg_success('node-list')
self.assertTrue(mock_list.called)
@mock.patch('magnumclient.v1.nodes.NodeManager.list')
def test_node_list_failure(self, mock_list):
self._test_arg_failure('node-list --wrong',
self._unrecognized_arg_error)
self.assertFalse(mock_list.called)
@mock.patch('magnumclient.v1.nodes.NodeManager.create')
def test_node_create_success(self, mock_create):
self._test_arg_success('node-create '
'--type test '
'--image-id test')
self.assertTrue(mock_create.called)

96
magnumclient/tests/v1/test_pods_shell.py Normal file
View File

@ -0,0 +1,96 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import mock
from magnumclient.tests.v1 import shell_test_base
class ShellTest(shell_test_base.TestCommandLineArgument):
@mock.patch('magnumclient.v1.pods.PodManager.list')
def test_pod_list_success(self, mock_list):
self._test_arg_success('pod-list bay_ident')
self.assertTrue(mock_list.called)
@mock.patch('magnumclient.v1.pods.PodManager.list')
def test_pod_list_failure(self, mock_list):
self._test_arg_failure('pod-list bay_ident --wrong',
self._unrecognized_arg_error)
self.assertFalse(mock_list.called)
@mock.patch('magnumclient.v1.bays.BayManager.get')
@mock.patch('magnumclient.v1.pods.PodManager.create')
def test_pod_create_success(self, mock_list, mock_get):
mockbay = mock.MagicMock()
mockbay.status = "CREATE_COMPLETE"
mock_get.return_value = mockbay
self._test_arg_success('pod-create '
'--bay xxx '
'--manifest test '
'--manifest-url test_url')
self.assertTrue(mock_list.called)
@mock.patch('magnumclient.v1.bays.BayManager.get')
@mock.patch('magnumclient.v1.pods.PodManager.create')
def test_pod_create_failure_few_arg(self, mock_list, mock_get):
self._test_arg_failure('pod-create '
'--manifest test '
'--manifest-url test_url',
self._mandatory_arg_error)
self.assertFalse(mock_list.called)
self._test_arg_failure('pod-create '
'bay xxx '
'--manifest-url test_url',
self._mandatory_arg_error)
self.assertFalse(mock_list.called)
@mock.patch('magnumclient.v1.pods.PodManager.delete')
def test_pod_delete_success(self, mock_delete):
self._test_arg_success('pod-delete xxx zzz')
self.assertTrue(mock_delete.called)
self.assertEqual(1, mock_delete.call_count)
@mock.patch('magnumclient.v1.pods.PodManager.delete')
def test_pod_delete_multiple_id_success(self, mock_delete):
self._test_arg_success('pod-delete xxx xyz zzz')
self.assertTrue(mock_delete.called)
self.assertEqual(2, mock_delete.call_count)
@mock.patch('magnumclient.v1.pods.PodManager.delete')
def test_pod_delete_failure_no_arg(self, mock_delete):
self._test_arg_failure('pod-delete', self._few_argument_error)
self.assertFalse(mock_delete.called)
@mock.patch('magnumclient.v1.pods.PodManager.update')
def test_pod_update_success(self, mock_update):
self._test_arg_success('pod-update xxx zzz replace xxx=xxx')
self.assertTrue(mock_update.called)
self.assertEqual(1, mock_update.call_count)
@mock.patch('magnumclient.v1.pods.PodManager.update')
def test_pod_update_failure_no_arg(self, mock_update):
self._test_arg_failure('pod-update', self._few_argument_error)
self.assertFalse(mock_update.called)
@mock.patch('magnumclient.v1.pods.PodManager.get')
def test_pod_show_success(self, mock_show):
self._test_arg_success('pod-show xxx zzz')
self.assertTrue(mock_show.called)
@mock.patch('magnumclient.v1.pods.PodManager.get')
def test_pod_show_failure_no_arg(self, mock_show):
self._test_arg_failure('pod-show', self._few_argument_error)
self.assertFalse(mock_show.called)

107
magnumclient/tests/v1/test_replicationcontrollers_shell.py Normal file
View File

@ -0,0 +1,107 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import mock
from magnumclient.tests.v1 import shell_test_base
class ShellTest(shell_test_base.TestCommandLineArgument):
@mock.patch('magnumclient.v1.replicationcontrollers.'
'ReplicationControllerManager.list')
def test_rc_list_success(self, mock_list):
self._test_arg_success('rc-list bay_ident')
self.assertTrue(mock_list.called)
@mock.patch('magnumclient.v1.replicationcontrollers.'
'ReplicationControllerManager.list')
def test_rc_list_failure(self, mock_list):
self._test_arg_failure('rc-list bay_ident --wrong',
self._unrecognized_arg_error)
self.assertFalse(mock_list.called)
@mock.patch('magnumclient.v1.bays.BayManager.get')
@mock.patch('magnumclient.v1.replicationcontrollers.'
'ReplicationControllerManager.create')
def test_rc_create_success(self, mock_create, mock_get):
mockbay = mock.MagicMock()
mockbay.status = "CREATE_COMPLETE"
mock_get.return_value = mockbay
self._test_arg_success('rc-create '
'--bay xxx '
'--manifest test '
'--manifest-url test_url')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.bays.BayManager.get')
@mock.patch('magnumclient.v1.replicationcontrollers.'
'ReplicationControllerManager.create')
def test_rc_create_failure_few_arg(self, mock_create, mock_get):
self._test_arg_failure('rc-create '
'--manifest test '
'--manifest-url test_url',
self._mandatory_arg_error)
self.assertFalse(mock_create.called)
self._test_arg_failure('rc-create '
'bay xxx '
'--manifest-url test_url',
self._mandatory_arg_error)
self.assertFalse(mock_create.called)
@mock.patch('magnumclient.v1.replicationcontrollers.'
'ReplicationControllerManager.delete')
def test_rc_delete_success(self, mock_delete):
self._test_arg_success('rc-delete xxx zzz')
self.assertTrue(mock_delete.called)
@mock.patch('magnumclient.v1.replicationcontrollers.'
'ReplicationControllerManager.delete')
def test_rc_delete_multiple_id_success(self, mock_delete):
self._test_arg_success('rc-delete xxx xyz zzz')
self.assertTrue(mock_delete.called)
self.assertEqual(2, mock_delete.call_count)
@mock.patch('magnumclient.v1.replicationcontrollers.'
'ReplicationControllerManager.delete')
def test_rc_delete_failure_no_arg(self, mock_delete):
self._test_arg_failure('rc-delete', self._few_argument_error)
self.assertFalse(mock_delete.called)
@mock.patch('magnumclient.v1.replicationcontrollers.'
'ReplicationControllerManager.get')
def test_rc_show_success(self, mock_show):
self._test_arg_success('rc-show xxx zzz')
self.assertTrue(mock_show.called)
@mock.patch('magnumclient.v1.replicationcontrollers.'
'ReplicationControllerManager.get')
def test_rc_show_failure_no_arg(self, mock_show):
self._test_arg_failure('rc-show', self._few_argument_error)
self.assertFalse(mock_show.called)
@mock.patch('magnumclient.v1.replicationcontrollers.'
'ReplicationControllerManager.update')
def test_rc_update_success(self, mock_update):
self._test_arg_success('rc-update xxx zzz replace xxx=xxx')
self.assertTrue(mock_update.called)
self.assertEqual(1, mock_update.call_count)
@mock.patch('magnumclient.v1.replicationcontrollers.'
'ReplicationControllerManager.update')
def test_rc_update_failure_no_arg(self, mock_update):
self._test_arg_failure('rc-update', self._few_argument_error)
self.assertFalse(mock_update.called)

89
magnumclient/tests/v1/test_services_shell.py Normal file
View File

@ -0,0 +1,89 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import mock
from magnumclient.tests.v1 import shell_test_base
class ShellTest(shell_test_base.TestCommandLineArgument):
@mock.patch('magnumclient.v1.services.ServiceManager.list')
def test_coe_service_list_success(self, mock_list):
self._test_arg_success('coe-service-list bay_ident')
self.assertTrue(mock_list.called)
@mock.patch('magnumclient.v1.services.ServiceManager.list')
def test_coe_service_list_failure(self, mock_list):
self._test_arg_failure('coe-service-list bay_ident --wrong',
self._unrecognized_arg_error)
self.assertFalse(mock_list.called)
@mock.patch('magnumclient.v1.bays.BayManager.get')
@mock.patch('magnumclient.v1.services.ServiceManager.create')
def test_coe_service_create_success(self, mock_create, mock_get):
mockbay = mock.MagicMock()
mockbay.status = "CREATE_COMPLETE"
mock_get.return_value = mockbay
self._test_arg_success('coe-service-create '
'--bay xxx '
'--manifest test '
'--manifest-url test_url')
self.assertTrue(mock_create.called)
@mock.patch('magnumclient.v1.services.ServiceManager.create')
def test_coe_service_create_failure_few_arg(self, mock_create):
self._test_arg_failure('coe-service-create '
'--manifest test '
'--manifest-url test_url',
self._mandatory_arg_error)
self.assertFalse(mock_create.called)
@mock.patch('magnumclient.v1.services.ServiceManager.delete')
def test_coe_service_delete_success(self, mock_delete):
self._test_arg_success('coe-service-delete xxx zzz')
self.assertTrue(mock_delete.called)
self.assertEqual(1, mock_delete.call_count)
@mock.patch('magnumclient.v1.services.ServiceManager.delete')
def test_coe_service_delete_multiple_id_success(self, mock_delete):
self._test_arg_success('coe-service-delete xxx xyz zzz')
self.assertTrue(mock_delete.called)
self.assertEqual(2, mock_delete.call_count)
@mock.patch('magnumclient.v1.services.ServiceManager.delete')
def test_coe_service_delete_failure_no_arg(self, mock_delete):
self._test_arg_failure('coe-service-delete', self._few_argument_error)
self.assertFalse(mock_delete.called)
@mock.patch('magnumclient.v1.services.ServiceManager.get')
def test_coe_service_show_success(self, mock_show):
self._test_arg_success('coe-service-show xxx zzz')
self.assertTrue(mock_show.called)
@mock.patch('magnumclient.v1.services.ServiceManager.get')
def test_coe_service_show_failure_no_arg(self, mock_show):
self._test_arg_failure('coe-service-show', self._few_argument_error)
self.assertFalse(mock_show.called)
@mock.patch('magnumclient.v1.services.ServiceManager.update')
def test_coe_service_update_success(self, mock_update):
self._test_arg_success('coe-service-update xxx zzz replace xxx=xxx')
self.assertTrue(mock_update.called)
self.assertEqual(1, mock_update.call_count)
@mock.patch('magnumclient.v1.services.ServiceManager.update')
def test_coe_service_update_failure_no_arg(self, mock_update):
self._test_arg_failure('coe-service-update', self._few_argument_error)
self.assertFalse(mock_update.called)

775
magnumclient/tests/v1/test_shell.py
View File

@ -1,775 +0,0 @@
# Copyright 2014 NEC Corporation. All rights reserved.
#
# 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.
import mock
from mock import mock_open
from magnumclient.common import utils as magnum_utils
from magnumclient.tests import base
from magnumclient.v1 import shell
class ShellTest(base.TestCase):
def setUp(self):
super(ShellTest, self).setUp()
def test_do_bay_list(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
shell.do_bay_list(client_mock, args)
client_mock.bays.list.assert_called_once_with()
def test_do_bay_create(self):
client_mock = mock.MagicMock()
baymodel = mock.MagicMock()
baymodel.uuid = 'uuid'
client_mock.baymodels.get.return_value = baymodel
args = mock.MagicMock()
node_count = 1
args.node_count = node_count
args.master_count = None
args.discovery_url = None
name = "test_bay"
args.name = name
baymodel_id_or_name = "test_baymodel_id"
args.baymodel = baymodel_id_or_name
args.timeout = None
shell.do_bay_create(client_mock, args)
client_mock.bays.create.assert_called_once_with(
name=name, node_count=node_count, baymodel_id=baymodel.uuid,
discovery_url=None, bay_create_timeout=None, master_count=None)
def test_do_bay_create_with_discovery_url(self):
client_mock = mock.MagicMock()
baymodel = mock.MagicMock()
baymodel.uuid = 'uuid'
client_mock.baymodels.get.return_value = baymodel
args = mock.MagicMock()
node_count = 1
args.node_count = node_count
args.master_count = None
discovery_url = 'discovery_url'
args.discovery_url = discovery_url
name = "test_bay"
args.name = name
baymodel_id_or_name = "test_baymodel_id"
args.baymodel = baymodel_id_or_name
args.timeout = None
shell.do_bay_create(client_mock, args)
client_mock.bays.create.assert_called_once_with(
name=name, node_count=node_count, baymodel_id=baymodel.uuid,
discovery_url=discovery_url, bay_create_timeout=None,
master_count=None)
def test_do_bay_create_with_bay_create_timeout(self):
client_mock = mock.MagicMock()
baymodel = mock.MagicMock()
baymodel.uuid = 'uuid'
client_mock.baymodels.get.return_value = baymodel
args = mock.MagicMock()
node_count = 1
args.node_count = node_count
args.master_count = None
name = "test_bay"
args.name = name
baymodel_id_or_name = "test_baymodel_id"
args.baymodel = baymodel_id_or_name
bay_create_timeout = 15
args.timeout = bay_create_timeout
args.discovery_url = None
shell.do_bay_create(client_mock, args)
client_mock.bays.create.assert_called_once_with(
name=name, node_count=node_count, baymodel_id=baymodel.uuid,
discovery_url=None, bay_create_timeout=bay_create_timeout,
master_count=None)
def test_do_bay_create_with_master_node_count(self):
client_mock = mock.MagicMock()
baymodel = mock.MagicMock()
baymodel.uuid = 'uuid'
client_mock.baymodels.get.return_value = baymodel
args = mock.MagicMock()
node_count = 1
args.node_count = node_count
master_count = 1
args.master_count = master_count
args.discovery_url = None
name = "test_bay"
args.name = name
baymodel_id_or_name = "test_baymodel_id"
args.baymodel = baymodel_id_or_name
args.timeout = None
shell.do_bay_create(client_mock, args)
client_mock.bays.create.assert_called_once_with(
name=name, node_count=node_count, baymodel_id=baymodel.uuid,
discovery_url=None, bay_create_timeout=None, master_count=1)
def test_do_bay_delete(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
bay_id = 'id'
args.bay = [bay_id]
shell.do_bay_delete(client_mock, args)
client_mock.bays.delete.assert_called_once_with(bay_id)
def test_do_bay_show(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
bay_id = 'id'
args.bay = bay_id
shell.do_bay_show(client_mock, args)
client_mock.bays.get.assert_called_once_with(bay_id)
def test_do_bay_update(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
bay_id = 'id'
args.bay = bay_id
op = 'add'
args.op = op
attributes = 'node_count=2'
args.attributes = attributes
shell.magnum_utils.args_array_to_patch = mock.MagicMock()
patch = [{'path': '/node_count', 'value': 2, 'op': 'add'}]
shell.magnum_utils.args_array_to_patch.return_value = patch
shell.do_bay_update(client_mock, args)
client_mock.bays.update.assert_called_once_with(bay_id, patch)
@mock.patch('os.path.isfile')
def test_do_ca_show(self, mock_isfile):
mock_isfile.return_value = True
client_mock = mock.MagicMock()
bay = mock.MagicMock()
bay.uuid = 'uuid'
bay.status = 'CREATE_COMPLETE'
client_mock.bays.get.return_value = bay
args = mock.MagicMock()
bay_id_or_name = "xxx"
args.bay = bay_id_or_name
shell.do_ca_show(client_mock, args)
client_mock.certificates.get.assert_called_once_with(
bay_uuid=bay.uuid)
@mock.patch('os.path.isfile')
def test_do_ca_show_wrong_status(self, mock_isfile):
mock_isfile.return_value = True
client_mock = mock.MagicMock()
bay = mock.MagicMock()
bay.uuid = 'uuid'
bay.status = 'XXX'
client_mock.bays.get.return_value = bay
args = mock.MagicMock()
bay_id_or_name = "xxx"
args.bay = bay_id_or_name
shell.do_ca_show(client_mock, args)
self.assertFalse(client_mock.certificates.get.called)
@mock.patch('os.path.isfile')
def test_do_ca_sign(self, mock_isfile):
mock_isfile.return_value = True
client_mock = mock.MagicMock()
bay = mock.MagicMock()
bay.uuid = 'uuid'
bay.status = 'CREATE_COMPLETE'
client_mock.bays.get.return_value = bay
args = mock.MagicMock()
bay_id_or_name = "xxx"
args.bay = bay_id_or_name
csr = "test_csr"
args.csr = csr
fake_csr = 'fake-csr'
mock_o = mock_open(read_data=fake_csr)
with mock.patch.object(shell, 'open', mock_o):
shell.do_ca_sign(client_mock, args)
mock_isfile.assert_called_once_with(csr)
mock_o.assert_called_once_with(csr, 'r')
client_mock.certificates.create.assert_called_once_with(
csr=fake_csr, bay_uuid=bay.uuid)
@mock.patch('os.path.isfile')
def test_do_ca_sign_wrong_status(self, mock_isfile):
mock_isfile.return_value = True
client_mock = mock.MagicMock()
bay = mock.MagicMock()
bay.uuid = 'uuid'
bay.status = 'XXX'
client_mock.bays.get.return_value = bay
args = mock.MagicMock()
bay_id_or_name = "xxx"
args.bay = bay_id_or_name
csr = "test_csr"
args.csr = csr
fake_csr = 'fake-csr'
mock_o = mock_open(read_data=fake_csr)
with mock.patch.object(shell, 'open', mock_o):
shell.do_ca_sign(client_mock, args)
self.assertFalse(mock_isfile.called)
self.assertFalse(mock_o.called)
self.assertFalse(client_mock.certificates.create.called)
@mock.patch('os.path.isfile')
def test_do_ca_sign_not_file(self, mock_isfile):
mock_isfile.return_value = False
client_mock = mock.MagicMock()
bay = mock.MagicMock()
bay.uuid = 'uuid'
bay.status = 'CREATE_COMPLETE'
client_mock.bays.get.return_value = bay
args = mock.MagicMock()
bay_id_or_name = "xxx"
args.bay = bay_id_or_name
csr = "test_csr"
args.csr = csr
fake_csr = 'fake-csr'
mock_o = mock_open(read_data=fake_csr)
with mock.patch.object(shell, 'open', mock_o):
shell.do_ca_sign(client_mock, args)
mock_isfile.assert_called_once_with(csr)
self.assertFalse(mock_o.called)
self.assertFalse(client_mock.certificates.create.called)
def test_do_baymodel_create(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
name = "test_baymodel"
args.name = name
image_id = "test_image"
args.image_id = image_id
flavor_id = "test_flavor"
args.flavor_id = flavor_id
master_flavor_id = "test_master_flavor"
args.master_flavor_id = master_flavor_id
keypair_id = "test_keypair"
args.keypair_id = keypair_id
external_network_id = "test_external_network_id"
args.external_network_id = external_network_id
dns_nameserver = "test_dns_nameserver"
args.dns_nameserver = dns_nameserver
docker_volume_size = "2051"
args.docker_volume_size = docker_volume_size
fixed_network = "private"
args.fixed_network = fixed_network
network_driver = "test_driver"
args.network_driver = network_driver
ssh_authorized_key = "test_key"
args.ssh_authorized_key = ssh_authorized_key
coe = 'swarm'
args.coe = coe
http_proxy = 'http_proxy'
args.http_proxy = http_proxy
https_proxy = 'https_proxy'
args.https_proxy = 'https_proxy'
no_proxy = 'no_proxy'
args.no_proxy = no_proxy
labels = ['key1=val1']
args.labels = labels
tls_disabled = True
args.tls_disabled = tls_disabled
public = True
args.public = public
shell.do_baymodel_create(client_mock, args)
client_mock.baymodels.create.assert_called_once_with(
name=name, image_id=image_id, flavor_id=flavor_id,
master_flavor_id=master_flavor_id, keypair_id=keypair_id,
external_network_id=external_network_id,
docker_volume_size=docker_volume_size,
fixed_network=fixed_network, dns_nameserver=dns_nameserver,
ssh_authorized_key=ssh_authorized_key, coe=coe,
http_proxy=http_proxy, https_proxy=https_proxy,
no_proxy=no_proxy, network_driver=network_driver,
labels=magnum_utils.format_labels(labels),
tls_disabled=tls_disabled, public=public)
def test_do_baymodel_delete(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
baymodel_id = 'id'
args.baymodels = [baymodel_id]
shell.do_baymodel_delete(client_mock, args)
client_mock.baymodels.delete.assert_called_once_with(baymodel_id)
def test_do_baymodel_show(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
baymodel_id = 'id'
args.baymodel = baymodel_id
shell.do_baymodel_show(client_mock, args)
client_mock.baymodels.get.assert_called_once_with(baymodel_id)
def test_do_baymodel_list(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
shell.do_baymodel_list(client_mock, args)
client_mock.baymodels.list.assert_called_once_with()
def test_do_baymodel_update(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
baymodel_id = 'id'
args.baymodel = baymodel_id
op = 'add'
args.op = op
attributes = 'name=updated_name'
args.attributes = attributes
shell.magnum_utils.args_array_to_patch = mock.MagicMock()
patch = [{'path': '/name', 'value': 'updated_name', 'op': 'add'}]
shell.magnum_utils.args_array_to_patch.return_value = patch
shell.do_baymodel_update(client_mock, args)
update = client_mock.baymodels.update
update.assert_called_once_with(baymodel_id, patch)
def test_do_node_list(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
shell.do_node_list(client_mock, args)
client_mock.nodes.list.assert_called_once_with()
def test_do_node_create(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
type = "test_type"
args.type = type
image_id = "test_image"
args.image_id = image_id
shell.do_node_create(client_mock, args)
client_mock.nodes.create.assert_called_once_with(
type=type, image_id=image_id)
def test_do_pod_list(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
bay = 'id'
args.bay = bay
shell.do_pod_list(client_mock, args)
client_mock.pods.list.assert_called_once_with(bay)
def test_do_pod_create(self):
client_mock = mock.MagicMock()
bay = mock.MagicMock()
bay.uuid = 'uuid'
bay.status = 'CREATE_COMPLETE'
client_mock.bays.get.return_value = bay
args = mock.MagicMock()
manifest_url = "test_url"
args.manifest_url = manifest_url
bay_id_or_name = "xxx"
args.bay = bay_id_or_name
manifest = "test_manifest"
args.manifest = manifest
shell.do_pod_create(client_mock, args)
client_mock.pods.create.assert_called_once_with(
manifest_url=manifest_url, bay_uuid=bay.uuid)
def test_do_pod_create_with_bay_in_wrong_status(self):
client_mock = mock.MagicMock()
bay = mock.MagicMock()
bay.uuid = 'uuid'
bay.status = 'XXX'
client_mock.bays.get.return_value = bay
args = mock.MagicMock()
manifest_url = "test_url"
args.manifest_url = manifest_url
bay_id_or_name = "xxx"
args.bay = bay_id_or_name
manifest = "test_manifest"
args.manifest = manifest
shell.do_pod_create(client_mock, args)
self.assertFalse(client_mock.pods.create.called)
def test_do_pod_update(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
pod_id = 'id'
args.pod = pod_id
bay_uuid = 'uuid'
args.bay = bay_uuid
op = 'add'
args.op = op
attributes = "labels={'name': 'value'}"
args.attributes = attributes
shell.magnum_utils.args_array_to_patch = mock.MagicMock()
patch = [{'path': '/labels', 'value': {'name': 'value'}, 'op': 'add'}]
shell.magnum_utils.args_array_to_patch.return_value = patch
shell.do_pod_update(client_mock, args)
client_mock.pods.update.assert_called_once_with(pod_id,
bay_uuid,
patch)
def test_do_pod_delete(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
pod_id = 'id'
bay_uuid = 'uuid'
args.pods = [pod_id]
args.bay = bay_uuid
shell.do_pod_delete(client_mock, args)
client_mock.pods.delete.assert_called_once_with(pod_id, bay_uuid)
def test_do_pod_show(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
pod_id = 'id'
args.pod = pod_id
bay_uuid = 'uuid'
args.bay = bay_uuid
shell.do_pod_show(client_mock, args)
client_mock.pods.get.assert_called_once_with(pod_id, bay_uuid)
def test_do_rc_list(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
bay = 'id'
args.bay = bay
shell.do_rc_list(client_mock, args)
client_mock.rcs.list.assert_called_once_with(bay)
def test_do_rc_create(self):
client_mock = mock.MagicMock()
bay = mock.MagicMock()
bay.uuid = 'uuid'
bay.status = 'CREATE_COMPLETE'
client_mock.bays.get.return_value = bay
args = mock.MagicMock()
manifest_url = "test_url"
args.manifest_url = manifest_url
bay_id_or_name = "xxx"
args.bay_id = bay_id_or_name
manifest = "test_manifest"
args.manifest = manifest
shell.do_rc_create(client_mock, args)
client_mock.rcs.create.assert_called_once_with(
manifest_url=manifest_url, bay_uuid=bay.uuid)
def test_do_rc_create_with_bay_status_wrong(self):
client_mock = mock.MagicMock()
bay = mock.MagicMock()
bay.uuid = 'uuid'
bay.status = 'XXX'
client_mock.bays.get.return_value = bay
args = mock.MagicMock()
manifest_url = "test_url"
args.manifest_url = manifest_url
bay_id_or_name = "xxx"
args.bay_id = bay_id_or_name
manifest = "test_manifest"
args.manifest = manifest
shell.do_rc_create(client_mock, args)
self.assertFalse(client_mock.rcs.create.called)
def test_do_rc_update(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
rc_id = 'id'
bay_uuid = 'uuid'
args.rc = rc_id
args.bay = bay_uuid
op = 'replace'
args.op = op
attributes = 'manifest={}'
args.attributes = attributes
shell.magnum_utils.args_array_to_patch = mock.MagicMock()
patch = [{'path': '/manifest', 'value': '{}', 'op': 'replace'}]
shell.magnum_utils.args_array_to_patch.return_value = patch
shell.do_rc_update(client_mock, args)
client_mock.rcs.update.assert_called_once_with(rc_id,
bay_uuid,
patch)
def test_do_rc_delete(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
rc_id = 'id'
bay_uuid = 'uuid'
args.rcs = [rc_id]
args.bay = bay_uuid
shell.do_rc_delete(client_mock, args)
client_mock.rcs.delete.assert_called_once_with(rc_id, bay_uuid)
def test_do_rc_show(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
rc_id = 'id'
bay_uuid = 'uuid'
args.rc = rc_id
args.bay = bay_uuid
shell.do_rc_show(client_mock, args)
client_mock.rcs.get.assert_called_once_with(rc_id, bay_uuid)
def test_do_coe_service_list(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
bay = 'id'
args.bay = bay
shell.do_coe_service_list(client_mock, args)
client_mock.services.list.assert_called_once_with(bay)
def test_do_coe_service_create(self):
client_mock = mock.MagicMock()
bay = mock.MagicMock()
bay.uuid = 'uuid'
bay.status = "CREATE_COMPLETE"
client_mock.bays.get.return_value = bay
args = mock.MagicMock()
manifest_url = "test_url"
args.manifest_url = manifest_url
bay_id_or_name = "xxx"
args.bay_id = bay_id_or_name
manifest = "test_manifest"
args.manifest = manifest
shell.do_coe_service_create(client_mock, args)
client_mock.services.create.assert_called_once_with(
manifest_url=manifest_url, bay_uuid=bay.uuid)
def test_do_service_create_with_bay_status_wrong(self):
client_mock = mock.MagicMock()
bay = mock.MagicMock()
bay.uuid = 'uuid'
bay.status = "XXX"
client_mock.bays.get.return_value = bay
args = mock.MagicMock()
manifest_url = "test_url"
args.manifest_url = manifest_url
bay_id_or_name = "xxx"
args.bay_id = bay_id_or_name
manifest = "test_manifest"
args.manifest = manifest
shell.do_coe_service_create(client_mock, args)
self.assertFalse(client_mock.services.create.called)
def test_do_coe_service_update(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
service_id = 'id'
args.service = service_id
bay_uuid = 'uuid'
args.bay = bay_uuid
op = 'replace'
args.op = op
attributes = 'manifest={}'
args.attributes = attributes
shell.magnum_utils.args_array_to_patch = mock.MagicMock()
patch = [{'path': '/manifest', 'value': '{}', 'op': 'replace'}]
shell.magnum_utils.args_array_to_patch.return_value = patch
shell.do_coe_service_update(client_mock, args)
client_mock.services.update.assert_called_once_with(service_id,
bay_uuid,
patch)
def test_do_coe_service_delete(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
service_id = 'id'
bay_uuid = 'uuid'
args.services = [service_id]
args.bay = bay_uuid
shell.do_coe_service_delete(client_mock, args)
client_mock.services.delete.assert_called_once_with(service_id,
bay_uuid)
def test_do_coe_service_show(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
service_id = 'id'
bay_uuid = 'uuid'
args.service = service_id
args.bay = bay_uuid
shell.do_coe_service_show(client_mock, args)
client_mock.services.get.assert_called_once_with(service_id,
bay_uuid)
def test_do_container_create(self):
client_mock = mock.MagicMock()
bay = mock.MagicMock()
bay.uuid = 'uuid'
bay.status = "CREATE_COMPLETE"
client_mock.bays.get.return_value = bay
args = mock.MagicMock()
name = "containe1"
args.name = name
image = "test_image"
args.image = image
bay_id_or_name = "xxx"
args.bay_id = bay_id_or_name
command = "test_command"
args.command = command
memory = "512m"
args.memory = memory
shell.do_container_create(client_mock, args)
client_mock.containers.create.assert_called_once_with(
name=name, image=image, bay_uuid=bay.uuid, command=command,
memory=memory)
def test_do_container_list(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
shell.do_container_list(client_mock, args)
client_mock.containers.list.assert_called_once_with()
def test_do_container_delete(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
container_id = "container_id"
args.containers = [container_id]
shell.do_container_delete(client_mock, args)
client_mock.containers.delete.assert_called_once_with(container_id)
def test_do_container_show(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
container_id = "container_id"
args.container = container_id
args.json = None
shell.do_container_show(client_mock, args)
client_mock.containers.get.assert_called_once_with(container_id)
def test_do_container_reboot(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
container_id = 'id'
args.containers = [container_id]
shell.do_container_reboot(client_mock, args)
client_mock.containers.reboot.assert_called_once_with(container_id)
def test_do_container_stop(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
container_id = 'id'
args.containers = [container_id]
shell.do_container_stop(client_mock, args)
client_mock.containers.stop.assert_called_once_with(container_id)
def test_do_container_start(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
container_id = 'id'
args.containers = [container_id]
shell.do_container_start(client_mock, args)
client_mock.containers.start.assert_called_once_with(container_id)
def test_do_container_pause(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
container_id = 'id'
args.containers = [container_id]
shell.do_container_pause(client_mock, args)
client_mock.containers.pause.assert_called_once_with(container_id)
def test_do_container_unpause(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
container_id = 'id'
args.containers = [container_id]
shell.do_container_unpause(client_mock, args)
client_mock.containers.unpause.assert_called_once_with(container_id)
def test_do_container_logs(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
container_id = 'id'
args.container = container_id
shell.do_container_logs(client_mock, args)
client_mock.containers.logs.assert_called_once_with(container_id)
def test_do_container_exec(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
container_id = 'id'
args.container = container_id
command = 'ls'
args.command = command
shell.do_container_exec(client_mock, args)
client_mock.containers.execute.assert_called_once_with(
container_id, command)
def test_do_service_list(self):
client_mock = mock.MagicMock()
args = mock.MagicMock()
shell.do_service_list(client_mock, args)
client_mock.mservices.list.assert_called_once_with()

173
magnumclient/v1/baymodels_shell.py Normal file
View File

@ -0,0 +1,173 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import os.path
from magnumclient.common import utils as magnum_utils
from magnumclient.openstack.common import cliutils as utils
def _show_baymodel(baymodel):
del baymodel._info['links']
utils.print_dict(baymodel._info)
@utils.arg('--name',
metavar='<name>',
help='Name of the bay to create.')
@utils.arg('--image-id',
required=True,
metavar='<image-id>',
help='The name or UUID of the base image to customize for the bay.')
@utils.arg('--keypair-id',
required=True,
metavar='<keypair-id>',
help='The name or UUID of the SSH keypair to load into the'
' Bay nodes.')
@utils.arg('--external-network-id',
required=True,
metavar='<external-network-id>',
help='The external Neutron network ID to connect to this bay'
' model.')
@utils.arg('--coe',
required=True,
metavar='<coe>',
help='Specify the Container Orchestration Engine to use.')
@utils.arg('--fixed-network',
metavar='<fixed-network>',
help='The private Neutron network name to connect to this bay'
' model.')
@utils.arg('--network-driver',
metavar='<network-driver>',
help='The network driver name for instantiating container'
' networks.')
@utils.arg('--ssh-authorized-key',
metavar='<ssh-authorized-key>',
help='The SSH authorized key to use')
@utils.arg('--dns-nameserver',
metavar='<dns-nameserver>',
default='8.8.8.8',
help='The DNS nameserver to use for this Bay.')
@utils.arg('--flavor-id',
metavar='<flavor-id>',
default='m1.medium',
help='The nova flavor id to use when launching the bay.')
@utils.arg('--master-flavor-id',
metavar='<master-flavor-id>',
help='The nova flavor id to use when launching the master node '
'of the bay.')
@utils.arg('--docker-volume-size',
metavar='<docker-volume-size>',
help='Specify the size of the docker volume to use.')
@utils.arg('--http-proxy',
metavar='<http-proxy>',
help='The http_proxy address to use for nodes in bay.')
@utils.arg('--https-proxy',
metavar='<https-proxy>',
help='The https_proxy address to use for nodes in bay.')
@utils.arg('--no-proxy',
metavar='<no-proxy>',
help='The no_proxy address to use for nodes in bay.')
@utils.arg('--labels', metavar='<KEY1=VALUE1,KEY2=VALUE2...>',
action='append', default=[],
help='Arbitrary labels in the form of key=value pairs '
'to associate with a baymodel. '
'May be used multiple times.')
@utils.arg('--tls-disabled',
action='store_true', default=False,
help='Disable TLS in the Bay.')
@utils.arg('--public',
action='store_true', default=False,
help='Make baymodel public.')
def do_baymodel_create(cs, args):
"""Create a baymodel."""
opts = {}
opts['name'] = args.name
opts['flavor_id'] = args.flavor_id
opts['master_flavor_id'] = args.master_flavor_id
opts['image_id'] = args.image_id
opts['keypair_id'] = args.keypair_id
opts['external_network_id'] = args.external_network_id
opts['fixed_network'] = args.fixed_network
opts['network_driver'] = args.network_driver
opts['dns_nameserver'] = args.dns_nameserver
opts['docker_volume_size'] = args.docker_volume_size
opts['ssh_authorized_key'] = args.ssh_authorized_key
opts['coe'] = args.coe
opts['http_proxy'] = args.http_proxy
opts['https_proxy'] = args.https_proxy
opts['no_proxy'] = args.no_proxy
opts['labels'] = magnum_utils.format_labels(args.labels)
opts['tls_disabled'] = args.tls_disabled
opts['public'] = args.public
baymodel = cs.baymodels.create(**opts)
_show_baymodel(baymodel)
@utils.arg('baymodels',
metavar='<baymodels>',
nargs='+',
help='ID or name of the (baymodel)s to delete.')
def do_baymodel_delete(cs, args):
"""Delete specified baymodel."""
for baymodel in args.baymodels:
try:
cs.baymodels.delete(baymodel)
except Exception as e:
print("Delete for baymodel %(baymodel)s failed: %(e)s" %
{'baymodel': baymodel, 'e': e})
@utils.arg('baymodel',
metavar='<baymodel>',
help='ID of the baymodel to show.')
def do_baymodel_show(cs, args):
"""Show details about the given baymodel."""
baymodel = cs.baymodels.get(args.baymodel)
_show_baymodel(baymodel)
def do_baymodel_list(cs, args):
"""Print a list of bay models."""
nodes = cs.baymodels.list()
columns = ('uuid', 'name')
utils.print_list(nodes, columns,
{'versions': magnum_utils.print_list_field('versions')})
@utils.arg('baymodel', metavar='<baymodel>', help="UUID or name of baymodel")
@utils.arg(
'op',
metavar='<op>',
choices=['add', 'replace', 'remove'],
help="Operations: 'add', 'replace' or 'remove'")
@utils.arg(
'attributes',
metavar='<path=value>',
nargs='+',
action='append',
default=[],
help="Attributes to add/replace or remove "
"(only PATH is necessary on remove)")
def do_baymodel_update(cs, args):
"""Updates one or more baymodel attributes."""
patch = magnum_utils.args_array_to_patch(args.op, args.attributes[0])
p = patch[0]
if p['path'] == '/manifest' and os.path.isfile(p['value']):
with open(p['value'], 'r') as f:
p['value'] = f.read()
baymodel = cs.baymodels.update(args.baymodel, patch)
_show_baymodel(baymodel)

110
magnumclient/v1/bays_shell.py Normal file
View File

@ -0,0 +1,110 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
from magnumclient.common import utils as magnum_utils
from magnumclient.openstack.common import cliutils as utils
def _show_bay(bay):
del bay._info['links']
utils.print_dict(bay._info)
def do_bay_list(cs, args):
"""Print a list of available bays."""
bays = cs.bays.list()
columns = ('uuid', 'name', 'node_count', 'master_count', 'status')
utils.print_list(bays, columns,
{'versions': magnum_utils.print_list_field('versions')})
@utils.arg('--name',
metavar='<name>',
help='Name of the bay to create.')
@utils.arg('--baymodel',
required=True,
metavar='<baymodel>',
help='ID or name of the baymodel.')
@utils.arg('--node-count',
metavar='<node-count>',
help='The bay node count.')
@utils.arg('--master-count',
metavar='<master-count>',
default=1,
help='The number of master nodes for the bay.')
@utils.arg('--discovery-url',
metavar='<discovery-url>',
help='Specifies custom discovery url for node discovery.')
@utils.arg('--timeout',
metavar='<timeout>',
help='The timeout for bay creation in minutes. Set '
'to 0 for no timeout. The default is no timeout.')
def do_bay_create(cs, args):
"""Create a bay."""
baymodel = cs.baymodels.get(args.baymodel)
opts = {}
opts['name'] = args.name
opts['baymodel_id'] = baymodel.uuid
opts['node_count'] = args.node_count
opts['master_count'] = args.master_count
opts['discovery_url'] = args.discovery_url
opts['bay_create_timeout'] = args.timeout
bay = cs.bays.create(**opts)
_show_bay(bay)
@utils.arg('bay',
metavar='<bay>',
nargs='+',
help='ID or name of the (bay)s to delete.')
def do_bay_delete(cs, args):
"""Delete specified bay."""
for id in args.bay:
try:
cs.bays.delete(id)
except Exception as e:
print("Delete for bay %(bay)s failed: %(e)s" %
{'bay': id, 'e': e})
@utils.arg('bay',
metavar='<bay>',
help='ID or name of the bay to show.')
def do_bay_show(cs, args):
"""Show details about the given bay."""
bay = cs.bays.get(args.bay)
_show_bay(bay)
@utils.arg('bay', metavar='<bay>', help="UUID or name of bay")
@utils.arg(
'op',
metavar='<op>',
choices=['add', 'replace', 'remove'],
help="Operations: 'add', 'replace' or 'remove'")
@utils.arg(
'attributes',
metavar='<path=value>',
nargs='+',
action='append',
default=[],
help="Attributes to add/replace or remove "
"(only PATH is necessary on remove)")
def do_bay_update(cs, args):
"""Update information about the given bay."""
patch = magnum_utils.args_array_to_patch(args.op, args.attributes[0])
bay = cs.bays.update(args.bay, patch)
_show_bay(bay)

73
magnumclient/v1/certificates_shell.py Normal file
View File

@ -0,0 +1,73 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import os.path
from magnumclient.openstack.common import cliutils as utils
def _show_cert(certificate):
print(certificate.pem)
@utils.arg('--bay',
required=True,
metavar='<bay>',
help='ID or name of the bay.')
def do_ca_show(cs, args):
"""Show details about the CA certificate for a bay."""
bay = cs.bays.get(args.bay)
if bay.status not in ['CREATE_COMPLETE', 'UPDATE_COMPLETE']:
print('Bay status for %s is: %s. We can not create a %s there'
' until the status is CREATE_COMPLETE or UPDATE_COMPLETE.' %
(bay.uuid, bay.status, 'certificate'))
return
opts = {
'bay_uuid': bay.uuid
}
cert = cs.certificates.get(**opts)
_show_cert(cert)
@utils.arg('--csr',
metavar='<csr>',
help='File path of the csr file to send to Magnum to get signed.')
@utils.arg('--bay',
required=True,
metavar='<bay>',
help='ID or name of the bay.')
def do_ca_sign(cs, args):
"""Generate the CA certificate for a bay."""
bay = cs.bays.get(args.bay)
if bay.status not in ['CREATE_COMPLETE', 'UPDATE_COMPLETE']:
print('Bay status for %s is: %s. We can not create a %s there'
' until the status is CREATE_COMPLETE or UPDATE_COMPLETE.' %
(bay.uuid, bay.status, 'certificate'))
return
opts = {
'bay_uuid': bay.uuid
}
if args.csr is None or not os.path.isfile(args.csr):
print('A CSR must be provided.')
return
with open(args.csr, 'r') as f:
opts['csr'] = f.read()
cert = cs.certificates.create(**opts)
_show_cert(cert)

187
magnumclient/v1/containers_shell.py Normal file
View File

@ -0,0 +1,187 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import json
from magnumclient.common import utils as magnum_utils
from magnumclient.openstack.common import cliutils as utils
def _show_container(container):
utils.print_dict(container._info)
@utils.arg('--name',
metavar='<name>',
help='name of the container')
@utils.arg('--image',
required=True,
metavar='<image>',
help='name or ID of the image')
@utils.arg('--bay',
required=True,
metavar='<bay>',
help='ID or name of the bay.')
@utils.arg('--command',
metavar='<command>',
help='Send command to the container')
@utils.arg('--memory',
metavar='<memory>',
help='The container memory size (format: <number><optional unit>, '
'where unit = b, k, m or g)')
def do_container_create(cs, args):
"""Create a container."""
bay = cs.bays.get(args.bay)
if bay.status not in ['CREATE_COMPLETE', 'UPDATE_COMPLETE']:
print('Bay status for %s is: %s. We can not create a %s there'
' until the status is CREATE_COMPLETE or UPDATE_COMPLETE.' %
(bay.uuid, bay.status, "pod"))
return
opts = {}
opts['name'] = args.name
opts['image'] = args.image
opts['bay_uuid'] = bay.uuid
opts['command'] = args.command
opts['memory'] = args.memory
_show_container(cs.containers.create(**opts))
def do_container_list(cs, args):
"""Print a list of available containers."""
containers = cs.containers.list()
columns = ('uuid', 'name', 'status')
utils.print_list(containers, columns,
{'versions': magnum_utils.print_list_field('versions')})
@utils.arg('containers',
metavar='<container>',
nargs='+',
help='ID or name of the (container)s to delete.')
def do_container_delete(cs, args):
"""Delete specified containers."""
for container in args.containers:
try:
cs.containers.delete(container)
except Exception as e:
print("Delete for container %(container)s failed: %(e)s" %
{'container': container, 'e': e})
@utils.arg('container',
metavar='<container>',
help='ID or name of the container to show.')
@utils.arg('--json',
action='store_true',
default=False,
help='Print JSON representation of the container.')
def do_container_show(cs, args):
"""Show details of a container."""
container = cs.containers.get(args.container)
if args.json:
print(json.dumps(container._info))
else:
_show_container(container)
@utils.arg('containers',
metavar='<container>',
nargs='+',
help='ID or name of the (container)s to start.')
def do_container_reboot(cs, args):
"""Reboot specified containers."""
for container in args.containers:
try:
cs.containers.reboot(container)
except Exception as e:
print("Reboot for container %(container)s failed: %(e)s" %
{'container': container, 'e': e})
@utils.arg('containers',
metavar='<container>',
nargs='+',
help='ID or name of the (container)s to start.')
def do_container_stop(cs, args):
"""Stop specified containers."""
for container in args.containers:
try:
cs.containers.stop(container)
except Exception as e:
print("Stop for container %(container)s failed: %(e)s" %
{'container': container, 'e': e})
@utils.arg('containers',
metavar='<container>',
nargs='+',
help='ID of the (container)s to start.')
def do_container_start(cs, args):
"""Start specified containers."""
for container in args.containers:
try:
cs.containers.start(container)
except Exception as e:
print("Start for container %(container)s failed: %(e)s" %
{'container': container, 'e': e})
@utils.arg('containers',
metavar='<container>',
nargs='+',
help='ID or name of the (container)s to start.')
def do_container_pause(cs, args):
"""Pause specified containers."""
for container in args.containers:
try:
cs.containers.pause(container)
except Exception as e:
print("Pause for container %(container)s failed: %(e)s" %
{'container': container, 'e': e})
@utils.arg('containers',
metavar='<container>',
nargs='+',
help='ID or name of the (container)s to start.')
def do_container_unpause(cs, args):
"""Unpause specified containers."""
for container in args.containers:
try:
cs.containers.unpause(container)
except Exception as e:
print("Unpause for container %(container)s failed: %(e)s" %
{'container': container, 'e': e})
@utils.arg('container',
metavar='<container>',
help='ID or name of the container to start.')
def do_container_logs(cs, args):
"""Get logs of a container."""
logs = cs.containers.logs(args.container)
print(logs)
@utils.arg('container',
metavar='<container>',
help='ID or name of the container to start.')
@utils.arg('--command',
required=True,
metavar='<command>',
help='The command to execute')
def do_container_exec(cs, args):
"""Execute command in a container."""
output = cs.containers.execute(args.container, args.command)
print(output)

25
magnumclient/v1/mservices_shell.py Normal file
View File

@ -0,0 +1,25 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
from magnumclient.common import utils as magnum_utils
from magnumclient.openstack.common import cliutils as utils
def do_service_list(cs, args):
"""Print a list of magnum services."""
mservices = cs.mservices.list()
columns = ('id', 'host', 'binary', 'state')
utils.print_list(mservices, columns,
{'versions': magnum_utils.print_list_field('versions')})

44
magnumclient/v1/nodes_shell.py Normal file
View File

@ -0,0 +1,44 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
from magnumclient.common import utils as magnum_utils
from magnumclient.openstack.common import cliutils as utils
def _show_node(node):
utils.print_dict(node._info)
def do_node_list(cs, args):
"""Print a list of configured nodes."""
nodes = cs.nodes.list()
columns = ('uuid', 'type', 'image_id')
utils.print_list(nodes, columns,
{'versions': magnum_utils.print_list_field('versions')})
@utils.arg('--type',
metavar='<type>',
help='Type of node to create (virt or bare).')
@utils.arg('--image-id',
metavar='<image-id>',
help='The name or UUID of the base image to use for the node.')
def do_node_create(cs, args):
"""Create a node."""
opts = {}
opts['type'] = args.type
opts['image_id'] = args.image_id
node = cs.nodes.create(**opts)
_show_node(node)

116
magnumclient/v1/pods_shell.py Normal file
View File

@ -0,0 +1,116 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import os.path
from magnumclient.common import utils as magnum_utils
from magnumclient.openstack.common import cliutils as utils
def _show_pod(pod):
del pod._info['links']
utils.print_dict(pod._info)
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_pod_list(cs, args):
"""Print a list of registered pods."""
pods = cs.pods.list(args.bay)
columns = ('uuid', 'name')
utils.print_list(pods, columns,
{'versions': magnum_utils.print_list_field('versions')})
@utils.arg('--manifest-url',
metavar='<manifest-url>',
help='Name/URL of the pod file to use for creating PODs.')
@utils.arg('--manifest',
metavar='<manifest>',
help='File path of the pod file to use for creating PODs.')
@utils.arg('--bay',
required=True,
metavar='<bay>',
help='ID or name of the bay.')
def do_pod_create(cs, args):
"""Create a pod."""
bay = cs.bays.get(args.bay)
if bay.status not in ['CREATE_COMPLETE', 'UPDATE_COMPLETE']:
print('Bay status for %s is: %s. We can not create a %s there'
' until the status is CREATE_COMPLETE or UPDATE_COMPLETE.' %
(bay.uuid, bay.status, "pod"))
return
opts = {}
opts['manifest_url'] = args.manifest_url
opts['bay_uuid'] = bay.uuid
if args.manifest is not None and os.path.isfile(args.manifest):
with open(args.manifest, 'r') as f:
opts['manifest'] = f.read()
node = cs.pods.create(**opts)
_show_pod(node)
pass
@utils.arg('pod', metavar='<pod-id>', help="UUID or name of pod")
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
@utils.arg(
'op',
metavar='<op>',
choices=['add', 'replace', 'remove'],
help="Operations: 'add', 'replace' or 'remove'")
@utils.arg(
'attributes',
metavar='<path=value>',
nargs='+',
action='append',
default=[],
help="Attributes to add/replace or remove "
"(only PATH is necessary on remove)")
def do_pod_update(cs, args):
"""Update information about the given pod."""
patch = magnum_utils.args_array_to_patch(args.op, args.attributes[0])
p = patch[0]
if p['path'] == '/manifest' and os.path.isfile(p['value']):
with open(p['value'], 'r') as f:
p['value'] = f.read()
pod = cs.pods.update(args.pod, args.bay, patch)
_show_pod(pod)
@utils.arg('pods',
metavar='<pods>',
nargs='+',
help='ID or name of the (pod)s to delete.')
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_pod_delete(cs, args):
"""Delete specified pod."""
for pod in args.pods:
try:
cs.pods.delete(pod, args.bay)
except Exception as e:
print("Delete for pod %(pod)s failed: %(e)s" %
{'pod': pod, 'e': e})
pass
@utils.arg('pod',
metavar='<pod>',
help='ID or name of the pod to show.')
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_pod_show(cs, args):
"""Show details about the given pod."""
pod = cs.pods.get(args.pod, args.bay)
_show_pod(pod)

117
magnumclient/v1/replicationcontrollers_shell.py Normal file
View File

@ -0,0 +1,117 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import os.path
from magnumclient.common import utils as magnum_utils
from magnumclient.openstack.common import cliutils as utils
def _show_rc(rc):
utils.print_dict(rc._info)
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_rc_list(cs, args):
"""Print a list of registered replication controllers."""
rcs = cs.rcs.list(args.bay)
columns = ('uuid', 'name')
utils.print_list(rcs, columns,
{'versions': magnum_utils.print_list_field('versions')})
@utils.arg('--manifest-url',
metavar='<manifest-url>',
help='Name/URL of the replication controller file to use for '
'creating replication controllers.')
@utils.arg('--manifest',
metavar='<manifest>',
help='File path of the replication controller file to use for '
'creating replication controllers.')
@utils.arg('--bay',
required=True,
metavar='<bay>',
help='ID or name of the bay.')
def do_rc_create(cs, args):
"""Create a replication controller."""
bay = cs.bays.get(args.bay)
if bay.status not in ['CREATE_COMPLETE', 'UPDATE_COMPLETE']:
print('Bay status for %s is: %s. We can not create a '
'replication controller in bay until the status '
'is CREATE_COMPLETE or UPDATE_COMPLETE.' %
(args.bay, bay.status))
return
opts = {}
opts['manifest_url'] = args.manifest_url
opts['bay_uuid'] = bay.uuid
if args.manifest is not None and os.path.isfile(args.manifest):
with open(args.manifest, 'r') as f:
opts['manifest'] = f.read()
rc = cs.rcs.create(**opts)
_show_rc(rc)
@utils.arg('rc', metavar='<rc>', help="UUID or name of replication controller")
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
@utils.arg(
'op',
metavar='<op>',
choices=['add', 'replace', 'remove'],
help="Operations: 'add', 'replace' or 'remove'")
@utils.arg(
'attributes',
metavar='<path=value>',
nargs='+',
action='append',
default=[],
help="Attributes to add/replace or remove "
"(only PATH is necessary on remove)")
def do_rc_update(cs, args):
"""Update information about the given replication controller."""
patch = magnum_utils.args_array_to_patch(args.op, args.attributes[0])
p = patch[0]
if p['path'] == '/manifest' and os.path.isfile(p['value']):
with open(p['value'], 'r') as f:
p['value'] = f.read()
rc = cs.rcs.update(args.rc, args.bay, patch)
_show_rc(rc)
@utils.arg('rcs',
metavar='<rcs>',
nargs='+',
help='ID or name of the replication (controller)s to delete.')
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_rc_delete(cs, args):
"""Delete specified replication controller."""
for rc in args.rcs:
try:
cs.rcs.delete(rc, args.bay)
except Exception as e:
print("Delete for rc %(rc)s failed: %(e)s" %
{'rc': rc, 'e': e})
@utils.arg('rc',
metavar='<rc>',
help='ID or name of the replication controller to show.')
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_rc_show(cs, args):
"""Show details about the given replication controller."""
rc = cs.rcs.get(args.rc, args.bay)
_show_rc(rc)

114
magnumclient/v1/services_shell.py Normal file
View File

@ -0,0 +1,114 @@
# Copyright 2015 NEC Corporation. All rights reserved.
#
# 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.
import os.path
from magnumclient.common import utils as magnum_utils
from magnumclient.openstack.common import cliutils as utils
def _show_coe_service(service):
utils.print_dict(service._info)
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_coe_service_list(cs, args):
"""Print a list of coe services."""
services = cs.services.list(args.bay)
columns = ('uuid', 'name', 'bay_uuid')
utils.print_list(services, columns,
{'versions': magnum_utils.print_list_field('versions')})
@utils.arg('--manifest-url',
metavar='<manifest-url>',
help='Name/URL of the serivce file to use for creating services.')
@utils.arg('--manifest',
metavar='<manifest>',
help='File path of the service file to use for creating services.')
@utils.arg('--bay',
required=True,
metavar='<bay>',
help='Id or name of the bay.')
def do_coe_service_create(cs, args):
"""Create a coe service."""
bay = cs.bays.get(args.bay)
if bay.status not in ['CREATE_COMPLETE', 'UPDATE_COMPLETE']:
print('Bay status for %s is: %s. We can not create a service in bay '
'until the status is CREATE_COMPLETE or UPDATE_COMPLETE.' %
(args.bay, bay.status))
return
opts = {}
opts['manifest_url'] = args.manifest_url
opts['bay_uuid'] = bay.uuid
if args.manifest is not None and os.path.isfile(args.manifest):
with open(args.manifest, 'r') as f:
opts['manifest'] = f.read()
service = cs.services.create(**opts)
_show_coe_service(service)
@utils.arg('service', metavar='<service>', help="UUID or name of service")
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
@utils.arg(
'op',
metavar='<op>',
choices=['add', 'replace', 'remove'],
help="Operations: 'add', 'replace' or 'remove'")
@utils.arg(
'attributes',
metavar='<path=value>',
nargs='+',
action='append',
default=[],
help="Attributes to add/replace or remove "
"(only PATH is necessary on remove)")
def do_coe_service_update(cs, args):
"""Update information about the given coe service."""
patch = magnum_utils.args_array_to_patch(args.op, args.attributes[0])
p = patch[0]
if p['path'] == '/manifest' and os.path.isfile(p['value']):
with open(p['value'], 'r') as f:
p['value'] = f.read()
service = cs.services.update(args.service, args.bay, patch)
_show_coe_service(service)
@utils.arg('services',
metavar='<services>',
nargs='+',
help='ID or name of the (service)s to delete.')
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_coe_service_delete(cs, args):
"""Delete specified coe service(s)."""
for service in args.services:
try:
cs.services.delete(service, args.bay)
except Exception as e:
print("Delete for service %(service)s failed: %(e)s" %
{'service': service, 'e': e})
@utils.arg('service',
metavar='<service>',
help='ID or name of the service to show.')
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_coe_service_show(cs, args):
"""Show details about the given coe service."""
service = cs.services.get(args.service, args.bay)
_show_coe_service(service)

836
magnumclient/v1/shell.py
View File

@ -13,818 +13,24 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import os.path
from magnumclient.common import utils as magnum_utils
from magnumclient.openstack.common import cliutils as utils
def _print_list_field(field):
return lambda obj: ', '.join(getattr(obj, field))
def _show_container(container):
utils.print_dict(container._info)
def _show_bay(bay):
del bay._info['links']
utils.print_dict(bay._info)
def _show_cert(certificate):
print(certificate.pem)
def _show_baymodel(baymodel):
del baymodel._info['links']
utils.print_dict(baymodel._info)
def _show_node(node):
utils.print_dict(node._info)
def _show_pod(pod):
del pod._info['links']
utils.print_dict(pod._info)
def _show_rc(rc):
utils.print_dict(rc._info)
def _show_coe_service(service):
utils.print_dict(service._info)
def do_bay_list(cs, args):
"""Print a list of available bays."""
bays = cs.bays.list()
columns = ('uuid', 'name', 'node_count', 'master_count', 'status')
utils.print_list(bays, columns,
{'versions': _print_list_field('versions')})
@utils.arg('--name',
metavar='<name>',
help='Name of the bay to create.')
@utils.arg('--baymodel',
required=True,
metavar='<baymodel>',
help='ID or name of the baymodel.')
@utils.arg('--node-count',
metavar='<node-count>',
help='The bay node count.')
@utils.arg('--master-count',
metavar='<master-count>',
default=1,
help='The number of master nodes for the bay.')
@utils.arg('--discovery-url',
metavar='<discovery-url>',
help='Specifies custom discovery url for node discovery.')
@utils.arg('--timeout',
metavar='<timeout>',
help='The timeout for bay creation in minutes. Set '
'to 0 for no timeout. The default is no timeout.')
def do_bay_create(cs, args):
"""Create a bay."""
baymodel = cs.baymodels.get(args.baymodel)
opts = {}
opts['name'] = args.name
opts['baymodel_id'] = baymodel.uuid
opts['node_count'] = args.node_count
opts['master_count'] = args.master_count
opts['discovery_url'] = args.discovery_url
opts['bay_create_timeout'] = args.timeout
bay = cs.bays.create(**opts)
_show_baymodel(bay)
@utils.arg('bay',
metavar='<bay>',
nargs='+',
help='ID or name of the (bay)s to delete.')
def do_bay_delete(cs, args):
"""Delete specified bay."""
for id in args.bay:
try:
cs.bays.delete(id)
except Exception as e:
print("Delete for bay %(bay)s failed: %(e)s" %
{'bay': id, 'e': e})
@utils.arg('bay',
metavar='<bay>',
help='ID or name of the bay to show.')
def do_bay_show(cs, args):
"""Show details about the given bay."""
bay = cs.bays.get(args.bay)
_show_bay(bay)
@utils.arg('bay', metavar='<bay>', help="UUID or name of bay")
@utils.arg(
'op',
metavar='<op>',
choices=['add', 'replace', 'remove'],
help="Operations: 'add', 'replace' or 'remove'")
@utils.arg(
'attributes',
metavar='<path=value>',
nargs='+',
action='append',
default=[],
help="Attributes to add/replace or remove "
"(only PATH is necessary on remove)")
def do_bay_update(cs, args):
"""Update information about the given bay."""
patch = magnum_utils.args_array_to_patch(args.op, args.attributes[0])
bay = cs.bays.update(args.bay, patch)
_show_bay(bay)
@utils.arg('--name',
metavar='<name>',
help='Name of the bay to create.')
@utils.arg('--image-id',
required=True,
metavar='<image-id>',
help='The name or UUID of the base image to customize for the bay.')
@utils.arg('--keypair-id',
required=True,
metavar='<keypair-id>',
help='The name or UUID of the SSH keypair to load into the'
' Bay nodes.')
@utils.arg('--external-network-id',
required=True,
metavar='<external-network-id>',
help='The external Neutron network ID to connect to this bay'
' model.')
@utils.arg('--coe',
required=True,
metavar='<coe>',
help='Specify the Container Orchestration Engine to use.')
@utils.arg('--fixed-network',
metavar='<fixed-network>',
help='The private Neutron network name to connect to this bay'
' model.')
@utils.arg('--network-driver',
metavar='<network-driver>',
help='The network driver name for instantiating container'
' networks.')
@utils.arg('--ssh-authorized-key',
metavar='<ssh-authorized-key>',
help='The SSH authorized key to use')
@utils.arg('--dns-nameserver',
metavar='<dns-nameserver>',
default='8.8.8.8',
help='The DNS nameserver to use for this Bay.')
@utils.arg('--flavor-id',
metavar='<flavor-id>',
default='m1.medium',
help='The nova flavor id to use when launching the bay.')
@utils.arg('--master-flavor-id',
metavar='<master-flavor-id>',
help='The nova flavor id to use when launching the master node '
'of the bay.')
@utils.arg('--docker-volume-size',
metavar='<docker-volume-size>',
help='Specify the size of the docker volume to use.')
@utils.arg('--http-proxy',
metavar='<http-proxy>',
help='The http_proxy address to use for nodes in bay.')
@utils.arg('--https-proxy',
metavar='<https-proxy>',
help='The https_proxy address to use for nodes in bay.')
@utils.arg('--no-proxy',
metavar='<no-proxy>',
help='The no_proxy address to use for nodes in bay.')
@utils.arg('--labels', metavar='<KEY1=VALUE1,KEY2=VALUE2...>',
action='append', default=[],
help='Arbitrary labels in the form of key=value pairs '
'to associate with a baymodel. '
'May be used multiple times.')
@utils.arg('--tls-disabled',
action='store_true', default=False,
help='Disable TLS in the Bay.')
@utils.arg('--public',
action='store_true', default=False,
help='Make baymodel public.')
def do_baymodel_create(cs, args):
"""Create a baymodel."""
opts = {}
opts['name'] = args.name
opts['flavor_id'] = args.flavor_id
opts['master_flavor_id'] = args.master_flavor_id
opts['image_id'] = args.image_id
opts['keypair_id'] = args.keypair_id
opts['external_network_id'] = args.external_network_id
opts['fixed_network'] = args.fixed_network
opts['network_driver'] = args.network_driver
opts['dns_nameserver'] = args.dns_nameserver
opts['docker_volume_size'] = args.docker_volume_size
opts['ssh_authorized_key'] = args.ssh_authorized_key
opts['coe'] = args.coe
opts['http_proxy'] = args.http_proxy
opts['https_proxy'] = args.https_proxy
opts['no_proxy'] = args.no_proxy
opts['labels'] = magnum_utils.format_labels(args.labels)
opts['tls_disabled'] = args.tls_disabled
opts['public'] = args.public
baymodel = cs.baymodels.create(**opts)
_show_baymodel(baymodel)
@utils.arg('baymodels',
metavar='<baymodels>',
nargs='+',
help='ID or name of the (baymodel)s to delete.')
def do_baymodel_delete(cs, args):
"""Delete specified baymodel."""
for baymodel in args.baymodels:
try:
cs.baymodels.delete(baymodel)
except Exception as e:
print("Delete for baymodel %(baymodel)s failed: %(e)s" %
{'baymodel': baymodel, 'e': e})
@utils.arg('baymodel',
metavar='<baymodel>',
help='ID of the baymodel to show.')
def do_baymodel_show(cs, args):
"""Show details about the given baymodel."""
baymodel = cs.baymodels.get(args.baymodel)
_show_baymodel(baymodel)
def do_baymodel_list(cs, args):
"""Print a list of bay models."""
nodes = cs.baymodels.list()
columns = ('uuid', 'name')
utils.print_list(nodes, columns,
{'versions': _print_list_field('versions')})
@utils.arg('--bay',
required=True,
metavar='<bay>',
help='ID or name of the bay.')
def do_ca_show(cs, args):
"""Show details about the CA certificate for a bay."""
bay = cs.bays.get(args.bay)
if bay.status not in ['CREATE_COMPLETE', 'UPDATE_COMPLETE']:
print('Bay status for %s is: %s. We can not create a %s there'
' until the status is CREATE_COMPLETE or UPDATE_COMPLETE.' %
(bay.uuid, bay.status, 'certificate'))
return
opts = {
'bay_uuid': bay.uuid
}
cert = cs.certificates.get(**opts)
_show_cert(cert)
@utils.arg('--csr',
metavar='<csr>',
help='File path of the csr file to send to Magnum to get signed.')
@utils.arg('--bay',
required=True,
metavar='<bay>',
help='ID or name of the bay.')
def do_ca_sign(cs, args):
"""Generate the CA certificate for a bay."""
bay = cs.bays.get(args.bay)
if bay.status not in ['CREATE_COMPLETE', 'UPDATE_COMPLETE']:
print('Bay status for %s is: %s. We can not create a %s there'
' until the status is CREATE_COMPLETE or UPDATE_COMPLETE.' %
(bay.uuid, bay.status, 'certificate'))
return
opts = {
'bay_uuid': bay.uuid
}
if args.csr is None or not os.path.isfile(args.csr):
print('A CSR must be provided.')
return
with open(args.csr, 'r') as f:
opts['csr'] = f.read()
cert = cs.certificates.create(**opts)
_show_cert(cert)
@utils.arg('baymodel', metavar='<baymodel>', help="UUID or name of baymodel")
@utils.arg(
'op',
metavar='<op>',
choices=['add', 'replace', 'remove'],
help="Operations: 'add', 'replace' or 'remove'")
@utils.arg(
'attributes',
metavar='<path=value>',
nargs='+',
action='append',
default=[],
help="Attributes to add/replace or remove "
"(only PATH is necessary on remove)")
def do_baymodel_update(cs, args):
"""Updates one or more baymodel attributes."""
patch = magnum_utils.args_array_to_patch(args.op, args.attributes[0])
p = patch[0]
if p['path'] == '/manifest' and os.path.isfile(p['value']):
with open(p['value'], 'r') as f:
p['value'] = f.read()
baymodel = cs.baymodels.update(args.baymodel, patch)
_show_baymodel(baymodel)
def do_node_list(cs, args):
"""Print a list of configured nodes."""
nodes = cs.nodes.list()
columns = ('uuid', 'type', 'image_id')
utils.print_list(nodes, columns,
{'versions': _print_list_field('versions')})
@utils.arg('--type',
metavar='<type>',
help='Type of node to create (virt or bare).')
@utils.arg('--image-id',
metavar='<image-id>',
help='The name or UUID of the base image to use for the node.')
def do_node_create(cs, args):
"""Create a node."""
opts = {}
opts['type'] = args.type
opts['image_id'] = args.image_id
node = cs.nodes.create(**opts)
_show_node(node)
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_pod_list(cs, args):
"""Print a list of registered pods."""
pods = cs.pods.list(args.bay)
columns = ('uuid', 'name')
utils.print_list(pods, columns,
{'versions': _print_list_field('versions')})
@utils.arg('--manifest-url',
metavar='<manifest-url>',
help='Name/URL of the pod file to use for creating PODs.')
@utils.arg('--manifest',
metavar='<manifest>',
help='File path of the pod file to use for creating PODs.')
@utils.arg('--bay',
required=True,
metavar='<bay>',
help='ID or name of the bay.')
def do_pod_create(cs, args):
"""Create a pod."""
bay = cs.bays.get(args.bay)
if bay.status not in ['CREATE_COMPLETE', 'UPDATE_COMPLETE']:
print('Bay status for %s is: %s. We can not create a %s there'
' until the status is CREATE_COMPLETE or UPDATE_COMPLETE.' %
(bay.uuid, bay.status, "pod"))
return
opts = {}
opts['manifest_url'] = args.manifest_url
opts['bay_uuid'] = bay.uuid
if args.manifest is not None and os.path.isfile(args.manifest):
with open(args.manifest, 'r') as f:
opts['manifest'] = f.read()
node = cs.pods.create(**opts)
_show_pod(node)
pass
@utils.arg('pod', metavar='<pod-id>', help="UUID or name of pod")
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
@utils.arg(
'op',
metavar='<op>',
choices=['add', 'replace', 'remove'],
help="Operations: 'add', 'replace' or 'remove'")
@utils.arg(
'attributes',
metavar='<path=value>',
nargs='+',
action='append',
default=[],
help="Attributes to add/replace or remove "
"(only PATH is necessary on remove)")
def do_pod_update(cs, args):
"""Update information about the given pod."""
patch = magnum_utils.args_array_to_patch(args.op, args.attributes[0])
p = patch[0]
if p['path'] == '/manifest' and os.path.isfile(p['value']):
with open(p['value'], 'r') as f:
p['value'] = f.read()
pod = cs.pods.update(args.pod, args.bay, patch)
_show_pod(pod)
@utils.arg('pods',
metavar='<pods>',
nargs='+',
help='ID or name of the (pod)s to delete.')
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_pod_delete(cs, args):
"""Delete specified pod."""
for pod in args.pods:
try:
cs.pods.delete(pod, args.bay)
except Exception as e:
print("Delete for pod %(pod)s failed: %(e)s" %
{'pod': pod, 'e': e})
pass
@utils.arg('pod',
metavar='<pod>',
help='ID or name of the pod to show.')
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_pod_show(cs, args):
"""Show details about the given pod."""
pod = cs.pods.get(args.pod, args.bay)
_show_pod(pod)
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_rc_list(cs, args):
"""Print a list of registered replication controllers."""
rcs = cs.rcs.list(args.bay)
columns = ('uuid', 'name')
utils.print_list(rcs, columns,
{'versions': _print_list_field('versions')})
@utils.arg('--manifest-url',
metavar='<manifest-url>',
help='Name/URL of the replication controller file to use for '
'creating replication controllers.')
@utils.arg('--manifest',
metavar='<manifest>',
help='File path of the replication controller file to use for '
'creating replication controllers.')
@utils.arg('--bay',
required=True,
metavar='<bay>',
help='ID or name of the bay.')
def do_rc_create(cs, args):
"""Create a replication controller."""
bay = cs.bays.get(args.bay)
if bay.status not in ['CREATE_COMPLETE', 'UPDATE_COMPLETE']:
print('Bay status for %s is: %s. We can not create a '
'replication controller in bay until the status '
'is CREATE_COMPLETE or UPDATE_COMPLETE.' %
(args.bay, bay.status))
return
opts = {}
opts['manifest_url'] = args.manifest_url
opts['bay_uuid'] = bay.uuid
if args.manifest is not None and os.path.isfile(args.manifest):
with open(args.manifest, 'r') as f:
opts['manifest'] = f.read()
rc = cs.rcs.create(**opts)
_show_rc(rc)
@utils.arg('rc', metavar='<rc>', help="UUID or name of replication controller")
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
@utils.arg(
'op',
metavar='<op>',
choices=['add', 'replace', 'remove'],
help="Operations: 'add', 'replace' or 'remove'")
@utils.arg(
'attributes',
metavar='<path=value>',
nargs='+',
action='append',
default=[],
help="Attributes to add/replace or remove "
"(only PATH is necessary on remove)")
def do_rc_update(cs, args):
"""Update information about the given replication controller."""
patch = magnum_utils.args_array_to_patch(args.op, args.attributes[0])
p = patch[0]
if p['path'] == '/manifest' and os.path.isfile(p['value']):
with open(p['value'], 'r') as f:
p['value'] = f.read()
rc = cs.rcs.update(args.rc, args.bay, patch)
_show_rc(rc)
@utils.arg('rcs',
metavar='<rcs>',
nargs='+',
help='ID or name of the replication (controller)s to delete.')
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_rc_delete(cs, args):
"""Delete specified replication controller."""
for rc in args.rcs:
try:
cs.rcs.delete(rc, args.bay)
except Exception as e:
print("Delete for rc %(rc)s failed: %(e)s" %
{'rc': rc, 'e': e})
@utils.arg('rc',
metavar='<rc>',
help='ID or name of the replication controller to show.')
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_rc_show(cs, args):
"""Show details about the given replication controller."""
rc = cs.rcs.get(args.rc, args.bay)
_show_rc(rc)
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_coe_service_list(cs, args):
"""Print a list of coe services."""
services = cs.services.list(args.bay)
columns = ('uuid', 'name', 'bay_uuid')
utils.print_list(services, columns,
{'versions': _print_list_field('versions')})
def do_service_list(cs, args):
"""Print a list of magnum services."""
mservices = cs.mservices.list()
columns = ('id', 'host', 'binary', 'state')
utils.print_list(mservices, columns,
{'versions': _print_list_field('versions')})
@utils.arg('--manifest-url',
metavar='<manifest-url>',
help='Name/URL of the serivce file to use for creating services.')
@utils.arg('--manifest',
metavar='<manifest>',
help='File path of the service file to use for creating services.')
@utils.arg('--bay',
required=True,
metavar='<bay>',
help='Id or name of the bay.')
def do_coe_service_create(cs, args):
"""Create a coe service."""
bay = cs.bays.get(args.bay)
if bay.status not in ['CREATE_COMPLETE', 'UPDATE_COMPLETE']:
print('Bay status for %s is: %s. We can not create a service in bay '
'until the status is CREATE_COMPLETE or UPDATE_COMPLETE.' %
(args.bay, bay.status))
return
opts = {}
opts['manifest_url'] = args.manifest_url
opts['bay_uuid'] = bay.uuid
if args.manifest is not None and os.path.isfile(args.manifest):
with open(args.manifest, 'r') as f:
opts['manifest'] = f.read()
service = cs.services.create(**opts)
_show_coe_service(service)
@utils.arg('service', metavar='<service>', help="UUID or name of service")
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
@utils.arg(
'op',
metavar='<op>',
choices=['add', 'replace', 'remove'],
help="Operations: 'add', 'replace' or 'remove'")
@utils.arg(
'attributes',
metavar='<path=value>',
nargs='+',
action='append',
default=[],
help="Attributes to add/replace or remove "
"(only PATH is necessary on remove)")
def do_coe_service_update(cs, args):
"""Update information about the given coe service."""
patch = magnum_utils.args_array_to_patch(args.op, args.attributes[0])
p = patch[0]
if p['path'] == '/manifest' and os.path.isfile(p['value']):
with open(p['value'], 'r') as f:
p['value'] = f.read()
service = cs.services.update(args.service, args.bay, patch)
_show_coe_service(service)
@utils.arg('services',
metavar='<services>',
nargs='+',
help='ID or name of the (service)s to delete.')
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_coe_service_delete(cs, args):
"""Delete specified coe service(s)."""
for service in args.services:
try:
cs.services.delete(service, args.bay)
except Exception as e:
print("Delete for service %(service)s failed: %(e)s" %
{'service': service, 'e': e})
@utils.arg('service',
metavar='<service>',
help='ID or name of the service to show.')
@utils.arg('bay', metavar='<bay>', help="UUID or Name of Bay")
def do_coe_service_show(cs, args):
"""Show details about the given coe service."""
service = cs.services.get(args.service, args.bay)
_show_coe_service(service)
#
# Containers
@utils.arg('--name',
metavar='<name>',
help='name of the container')
@utils.arg('--image',
required=True,
metavar='<image>',
help='name or ID of the image')
@utils.arg('--bay',
required=True,
metavar='<bay>',
help='ID or name of the bay.')
@utils.arg('--command',
metavar='<command>',
help='Send command to the container')
@utils.arg('--memory',
metavar='<memory>',
help='The container memory size (format: <number><optional unit>, '
'where unit = b, k, m or g)')
def do_container_create(cs, args):
"""Create a container."""
bay = cs.bays.get(args.bay)
if bay.status not in ['CREATE_COMPLETE', 'UPDATE_COMPLETE']:
print('Bay status for %s is: %s. We can not create a %s there'
' until the status is CREATE_COMPLETE or UPDATE_COMPLETE.' %
(bay.uuid, bay.status, "pod"))
return
opts = {}
opts['name'] = args.name
opts['image'] = args.image
opts['bay_uuid'] = bay.uuid
opts['command'] = args.command
opts['memory'] = args.memory
_show_container(cs.containers.create(**opts))
def do_container_list(cs, args):
"""Print a list of available containers."""
containers = cs.containers.list()
columns = ('uuid', 'name', 'status')
utils.print_list(containers, columns,
{'versions': _print_list_field('versions')})
@utils.arg('containers',
metavar='<container>',
nargs='+',
help='ID or name of the (container)s to delete.')
def do_container_delete(cs, args):
"""Delete specified containers."""
for container in args.containers:
try:
cs.containers.delete(container)
except Exception as e:
print("Delete for container %(container)s failed: %(e)s" %
{'container': container, 'e': e})
@utils.arg('container',
metavar='<container>',
help='ID or name of the container to show.')
@utils.arg('--json',
action='store_true',
default=False,
help='Print JSON representation of the container.')
def do_container_show(cs, args):
"""Show details of a container."""
container = cs.containers.get(args.container)
if args.json:
print(json.dumps(container._info))
else:
_show_container(container)
@utils.arg('containers',
metavar='<container>',
nargs='+',
help='ID or name of the (container)s to start.')
def do_container_reboot(cs, args):
"""Reboot specified containers."""
for container in args.containers:
try:
cs.containers.reboot(container)
except Exception as e:
print("Reboot for container %(container)s failed: %(e)s" %
{'container': container, 'e': e})
@utils.arg('containers',
metavar='<container>',
nargs='+',
help='ID or name of the (container)s to start.')
def do_container_stop(cs, args):
"""Stop specified containers."""
for container in args.containers:
try:
cs.containers.stop(container)
except Exception as e:
print("Stop for container %(container)s failed: %(e)s" %
{'container': container, 'e': e})
@utils.arg('containers',
metavar='<container>',
nargs='+',
help='ID of the (container)s to start.')
def do_container_start(cs, args):
"""Start specified containers."""
for container in args.containers:
try:
cs.containers.start(container)
except Exception as e:
print("Start for container %(container)s failed: %(e)s" %
{'container': container, 'e': e})
@utils.arg('containers',
metavar='<container>',
nargs='+',
help='ID or name of the (container)s to start.')
def do_container_pause(cs, args):
"""Pause specified containers."""
for container in args.containers:
try:
cs.containers.pause(container)
except Exception as e:
print("Pause for container %(container)s failed: %(e)s" %
{'container': container, 'e': e})
@utils.arg('containers',
metavar='<container>',
nargs='+',
help='ID or name of the (container)s to start.')
def do_container_unpause(cs, args):
"""Unpause specified containers."""
for container in args.containers:
try:
cs.containers.unpause(container)
except Exception as e:
print("Unpause for container %(container)s failed: %(e)s" %
{'container': container, 'e': e})
@utils.arg('container',
metavar='<container>',
help='ID or name of the container to start.')
def do_container_logs(cs, args):
"""Get logs of a container."""
logs = cs.containers.logs(args.container)
print(logs)
@utils.arg('container',
metavar='<container>',
help='ID or name of the container to start.')
@utils.arg('--command',
required=True,
metavar='<command>',
help='The command to execute')
def do_container_exec(cs, args):
"""Execute command in a container."""
output = cs.containers.execute(args.container, args.command)
print(output)
from magnumclient.v1 import baymodels_shell
from magnumclient.v1 import bays_shell
from magnumclient.v1 import certificates_shell
from magnumclient.v1 import containers_shell
from magnumclient.v1 import mservices_shell
from magnumclient.v1 import nodes_shell
from magnumclient.v1 import pods_shell
from magnumclient.v1 import replicationcontrollers_shell
from magnumclient.v1 import services_shell
COMMAND_MODULES = [
baymodels_shell,
bays_shell,
certificates_shell,
containers_shell,
mservices_shell,
nodes_shell,
pods_shell,
replicationcontrollers_shell,
services_shell,
]