starlingx
/
distcloud-client
Code Issues Proposed changes

Merge remote-tracking branch 'origin/master' into f/centos8 

Signed-off-by: Charles Short <charles.short@windriver.com>
Change-Id: I4aed28a2844577bf25f6a36297981f88f57c74bf
This commit is contained in:
Charles Short 2021-06-04 08:34:27 -04:00
parent 30f883c67c 8d2b5478c6
commit 0d0781278f
43 changed files with 1051 additions and 3161 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
.zuul.yaml
View File

@ -3,17 +3,20 @@
templates:
- publish-stx-docs
- stx-release-notes-jobs
- stx-bandit-jobs
check:
jobs:
- openstack-tox-linters
- stx-distcloud-client-tox-pep8
- stx-distcloud-client-tox-py27
- stx-distcloud-client-tox-py36
- stx-distcloud-client-tox-pylint
gate:
jobs:
- openstack-tox-linters
- stx-distcloud-client-tox-pep8
- stx-distcloud-client-tox-py27
- stx-distcloud-client-tox-py36
- stx-distcloud-client-tox-pylint
post:
jobs:
@ -28,6 +31,15 @@
tox_envlist: py27
tox_extra_args: -c distributedcloud-client/tox.ini
- job:
name: stx-distcloud-client-tox-py36
parent: tox
description: Run py36 for distcloud-client
nodeset: ubuntu-bionic
vars:
tox_envlist: py36
tox_extra_args: -c distributedcloud-client/tox.ini
- job:
name: stx-distcloud-client-tox-pylint
parent: tox

2
distributedcloud-client/centos/build_srpm.data
View File

@ -1,3 +1,3 @@
SRC_DIR="."
TIS_PATCH_VER=0
TIS_PATCH_VER=PKG_GITREVCOUNT

18
distributedcloud-client/dcmanagerclient/api/base.py
View File

@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
# Copyright (c) 2017 Wind River Systems, Inc.
# Copyright (c) 2017-2021 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
@ -41,8 +41,11 @@ class ResourceManager(object):
resource = []
for json_object in json_objects:
for resource_data in json_object:
resource.append(self.resource_class(self, resource_data,
json_object[resource_data]))
resource.append(
self.resource_class( # pylint: disable=not-callable
self,
resource_data,
json_object[resource_data]))
return resource
def _list(self, url, response_key=None):
@ -77,9 +80,12 @@ class ResourceManager(object):
for json_object in json_objects:
data = json_object.get('usage').keys()
for values in data:
resource.append(self.resource_class(self, values,
json_object['limits'][values],
json_object['usage'][values]))
resource.append(
self.resource_class( # pylint: disable=not-callable
self,
values,
json_object['limits'][values],
json_object['usage'][values]))
return resource
def _delete(self, url):

14
distributedcloud-client/dcmanagerclient/api/httpclient.py
View File

@ -61,8 +61,10 @@ class HTTPClient(object):
LOG.warning('Client is set to not verify even though '
'cacert is provided.')
self.ssl_options['verify'] = not insecure
self.ssl_options['cert'] = cacert
if insecure:
self.ssl_options['verify'] = False
else:
self.ssl_options['verify'] = True if not cacert else cacert
@log_request
def get(self, url, headers=None):
@ -97,9 +99,11 @@ class HTTPClient(object):
def _get_request_options(self, method, headers):
headers = self._update_headers(headers)
if method in ['post', 'put', 'patch']:
content_type = headers.get('content-type', 'application/json')
headers['content-type'] = content_type
CONTENT_TYPE = 'content-type'
if method in ['post', 'put', 'patch'] and CONTENT_TYPE not in headers:
content_type = headers.get(CONTENT_TYPE, 'application/json')
headers[CONTENT_TYPE] = content_type
options = copy.deepcopy(self.ssl_options)
options['headers'] = headers

4
distributedcloud-client/dcmanagerclient/api/v1/client.py
View File

@ -14,7 +14,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
# Copyright (c) 2017-2020 Wind River Systems, Inc.
# Copyright (c) 2017-2021 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
@ -29,6 +29,7 @@ import osprofiler.profiler
from dcmanagerclient.api import httpclient
from dcmanagerclient.api.v1 import alarm_manager as am
from dcmanagerclient.api.v1 import fw_update_manager as fum
from dcmanagerclient.api.v1 import kube_upgrade_manager as kupm
from dcmanagerclient.api.v1 import strategy_step_manager as ssm
from dcmanagerclient.api.v1 import subcloud_deploy_manager as sdm
from dcmanagerclient.api.v1 import subcloud_group_manager as gm
@ -102,6 +103,7 @@ class Client(object):
self.http_client)
self.alarm_manager = am.alarm_manager(self.http_client)
self.fw_update_manager = fum.fw_update_manager(self.http_client)
self.kube_upgrade_manager = kupm.kube_upgrade_manager(self.http_client)
self.sw_patch_manager = spm.sw_patch_manager(self.http_client)
self.sw_update_options_manager = \
suom.sw_update_options_manager(self.http_client)

32
distributedcloud-client/dcmanagerclient/api/v1/kube_upgrade_manager.py Normal file
View File

@ -0,0 +1,32 @@
# Copyright (c) 2017 Ericsson AB.
# 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.
#
# Copyright (c) 2020-2021 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
# of an applicable Wind River license agreement.
#
from dcmanagerclient.api.v1.sw_update_manager import sw_update_manager
SW_UPDATE_TYPE_KUBERNETES = "kubernetes"
class kube_upgrade_manager(sw_update_manager):
def __init__(self, http_client):
super(kube_upgrade_manager, self).__init__(
http_client,
update_type=SW_UPDATE_TYPE_KUBERNETES)

2
distributedcloud-client/dcmanagerclient/api/v1/subcloud_deploy_manager.py
View File

@ -56,7 +56,7 @@ class subcloud_deploy_manager(base.ResourceManager):
for k, v in data.items():
fields.update({k: (v, open(v, 'rb'),)})
enc = MultipartEncoder(fields=fields)
headers = {'Content-Type': enc.content_type}
headers = {'content-type': enc.content_type}
resp = self.http_client.post(url, enc, headers=headers)
if resp.status_code != 200:
self._raise_api_exception(resp)

106
distributedcloud-client/dcmanagerclient/api/v1/subcloud_manager.py
View File

@ -13,15 +13,13 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
# Copyright (c) 2017-2019 Wind River Systems, Inc.
# Copyright (c) 2017-2021 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
# of an applicable Wind River license agreement.
#
import json
from requests_toolbelt import MultipartEncoder
from dcmanagerclient.api import base
@ -37,7 +35,9 @@ class Subcloud(base.Resource):
management_subnet, management_start_ip, management_end_ip,
management_gateway_ip, systemcontroller_gateway_ip,
created_at, updated_at, group_id, sync_status="unknown",
endpoint_sync_status={}):
endpoint_sync_status=None):
if endpoint_sync_status is None:
endpoint_sync_status = {}
self.manager = manager
self.subcloud_id = subcloud_id
self.name = name
@ -90,7 +90,7 @@ class subcloud_manager(base.ResourceManager):
fields.update({k: (v, open(v, 'rb'),)})
fields.update(data)
enc = MultipartEncoder(fields=fields)
headers = {'Content-Type': enc.content_type}
headers = {'content-type': enc.content_type}
resp = self.http_client.post(url, enc, headers=headers)
if resp.status_code != 200:
self._raise_api_exception(resp)
@ -99,33 +99,62 @@ class subcloud_manager(base.ResourceManager):
resource.append(self.json_to_resource(json_object))
return resource
def subcloud_update(self, url, data):
data = json.dumps(data)
resp = self.http_client.patch(url, data)
def subcloud_update(self, url, body, data):
fields = dict()
if body is not None:
for k, v in body.items():
fields.update({k: (v, open(v, 'rb'),)})
fields.update(data)
enc = MultipartEncoder(fields=fields)
headers = {'content-type': enc.content_type}
resp = self.http_client.patch(url, enc, headers=headers)
if resp.status_code != 200:
self._raise_api_exception(resp)
json_object = get_json(resp)
resource = list()
resource.append(
self.resource_class(
self,
subcloud_id=json_object['id'],
name=json_object['name'],
description=json_object['description'],
location=json_object['location'],
software_version=json_object['software-version'],
management_state=json_object['management-state'],
availability_status=json_object['availability-status'],
deploy_status=json_object['deploy-status'],
management_subnet=json_object['management-subnet'],
management_start_ip=json_object['management-start-ip'],
management_end_ip=json_object['management-end-ip'],
management_gateway_ip=json_object['management-gateway-ip'],
systemcontroller_gateway_ip=json_object[
'systemcontroller-gateway-ip'],
created_at=json_object['created-at'],
updated_at=json_object['updated-at'],
group_id=json_object['group_id']))
resource.append(self.json_to_resource(json_object))
return resource
def subcloud_reconfigure(self, url, body, data):
fields = dict()
for k, v in body.items():
fields.update({k: (v, open(v, 'rb'),)})
fields.update(data)
enc = MultipartEncoder(fields=fields)
headers = {'content-type': enc.content_type}
resp = self.http_client.patch(url, enc, headers=headers)
if resp.status_code != 200:
self._raise_api_exception(resp)
json_object = get_json(resp)
resource = list()
resource.append(self.json_to_resource(json_object))
return resource
def subcloud_reinstall(self, url):
fields = dict()
enc = MultipartEncoder(fields=fields)
headers = {'content-type': enc.content_type}
resp = self.http_client.patch(url, enc, headers=headers)
if resp.status_code != 200:
self._raise_api_exception(resp)
json_object = get_json(resp)
resource = list()
resource.append(self.json_to_resource(json_object))
return resource
def subcloud_restore(self, url, body, data):
fields = dict()
for k, v in body.items():
fields.update({k: (v, open(v, 'rb'),)})
fields.update(data)
enc = MultipartEncoder(fields=fields)
headers = {'content-type': enc.content_type}
resp = self.http_client.patch(url, enc, headers=headers)
if resp.status_code != 200:
self._raise_api_exception(resp)
json_object = get_json(resp)
resource = list()
resource.append(self.json_to_resource(json_object))
return resource
def subcloud_list(self, url):
@ -214,6 +243,23 @@ class subcloud_manager(base.ResourceManager):
return self._delete(url)
def update_subcloud(self, subcloud_ref, **kwargs):
data = kwargs
files = kwargs.get('files')
data = kwargs.get('data')
url = '/subclouds/%s' % subcloud_ref
return self.subcloud_update(url, data)
return self.subcloud_update(url, files, data)
def reconfigure_subcloud(self, subcloud_ref, **kwargs):
files = kwargs.get('files')
data = kwargs.get('data')
url = '/subclouds/%s/reconfigure' % subcloud_ref
return self.subcloud_reconfigure(url, files, data)
def reinstall_subcloud(self, subcloud_ref):
url = '/subclouds/%s/reinstall' % subcloud_ref
return self.subcloud_reinstall(url)
def restore_subcloud(self, subcloud_ref, **kwargs):
files = kwargs.get('files')
data = kwargs.get('data')
url = '/subclouds/%s/restore' % subcloud_ref
return self.subcloud_restore(url, files, data)

9
distributedcloud-client/dcmanagerclient/api/v1/sw_update_manager.py
View File

@ -67,11 +67,14 @@ class sw_update_manager(base.ResourceManager):
# create_url is typically /<foo>/
self.create_url = '/{}/'.format(url)
# get_url is typically /<foo>
self.get_url = '/{}'.format(url)
self.get_url = '/{url}?type={update_type}'.format(
url=url, update_type=self.update_type)
# delete_url is typically /<foo> (same as get)
self.delete_url = '/{}'.format(url)
self.delete_url = '/{url}?type={update_type}'.format(
url=url, update_type=self.update_type)
# actions_url is typically /<foo>/actions
self.actions_url = '/{}/actions'.format(url)
self.actions_url = '/{url}/actions?type={update_type}'.format(
url=url, update_type=self.update_type)
def create_sw_update_strategy(self, **kwargs):
data = kwargs

10
distributedcloud-client/dcmanagerclient/commands/v1/alarm_manager.py
View File

@ -34,11 +34,11 @@ def format(alarms=None):
if alarms:
data = (
alarms.name if alarms.name >= 0 else '-',
alarms.critical if alarms.critical >= 0 else '-',
alarms.major if alarms.major >= 0 else '-',
alarms.minor if alarms.minor >= 0 else '-',
alarms.warnings if alarms.warnings >= 0 else '-',
alarms.name if alarms.name else '-',
alarms.critical if int(alarms.critical) >= 0 else '-',
alarms.major if int(alarms.major) >= 0 else '-',
alarms.minor if int(alarms.minor) >= 0 else '-',
alarms.warnings if int(alarms.warnings) >= 0 else '-',
alarms.status
)

59
distributedcloud-client/dcmanagerclient/commands/v1/kube_upgrade_manager.py Normal file
View File

@ -0,0 +1,59 @@
# Copyright (c) 2017 Ericsson AB.
#
# 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.
#
# Copyright (c) 2020-2021 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
# of an applicable Wind River license agreement.
#
from dcmanagerclient.commands.v1 import sw_update_manager
class KubeUpgradeManagerMixin(object):
"""This Mixin provides the update manager used for kubernetes upgrades."""
def get_sw_update_manager(self):
dcmanager_client = self.app.client_manager.kube_upgrade_manager
return dcmanager_client.kube_upgrade_manager
class CreateKubeUpgradeStrategy(KubeUpgradeManagerMixin,
sw_update_manager.CreateSwUpdateStrategy):
"""Create a kubernetes upgrade strategy."""
pass
class ShowKubeUpgradeStrategy(KubeUpgradeManagerMixin,
sw_update_manager.ShowSwUpdateStrategy):
"""Show the details of a kubernetes upgrade strategy for a subcloud."""
pass
class DeleteKubeUpgradeStrategy(KubeUpgradeManagerMixin,
sw_update_manager.DeleteSwUpdateStrategy):
"""Delete kubernetes upgrade strategy from the database."""
pass
class ApplyKubeUpgradeStrategy(KubeUpgradeManagerMixin,
sw_update_manager.ApplySwUpdateStrategy):
"""Apply a kubernetes upgrade strategy."""
pass
class AbortKubeUpgradeStrategy(KubeUpgradeManagerMixin,
sw_update_manager.AbortSwUpdateStrategy):
"""Abort a kubernetes upgrade strategy."""
pass

321
distributedcloud-client/dcmanagerclient/commands/v1/subcloud_manager.py
View File

@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
# Copyright (c) 2017-2020 Wind River Systems, Inc.
# Copyright (c) 2017-2021 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
@ -22,6 +22,7 @@
import base64
import getpass
import os
import six
from osc_lib.command import command
@ -95,8 +96,7 @@ def detail_format(subcloud=None):
subcloud.updated_at,
)
for listitem, sync_status in enumerate(subcloud.endpoint_sync_status
):
for _listitem, sync_status in enumerate(subcloud.endpoint_sync_status):
added_field = (sync_status['endpoint_type'] +
"_sync_status",)
added_value = (sync_status['sync_status'],)
@ -112,6 +112,23 @@ def detail_format(subcloud=None):
return columns, data
def prompt_for_password(password_type='sysadmin'):
while True:
password = getpass.getpass(
"Enter the " + password_type + " password for the subcloud: ")
if len(password) < 1:
print("Password cannot be empty")
continue
confirm = getpass.getpass(
"Re-enter " + password_type + " password to confirm: ")
if password != confirm:
print("Passwords did not match")
continue
break
return password
class AddSubcloud(base.DCManagerShowOne):
"""Add a new subcloud."""
@ -159,7 +176,8 @@ class AddSubcloud(base.DCManagerShowOne):
'--bmc-password',
required=False,
help='bmc password of the subcloud to be configured, '
'if not provided you will be prompted.'
'if not provided you will be prompted. This parameter is only'
' valid if the --install-values are specified.'
)
parser.add_argument(
@ -167,6 +185,13 @@ class AddSubcloud(base.DCManagerShowOne):
required=False,
help='Name or ID of subcloud group.'
)
parser.add_argument(
'--migrate',
required=False,
action='store_true',
help='Migrate a subcloud from another distributed cloud.'
)
return parser
def _get_resources(self, parsed_args):
@ -203,46 +228,25 @@ class AddSubcloud(base.DCManagerShowOne):
data['sysadmin_password'] = base64.b64encode(
parsed_args.sysadmin_password.encode("utf-8"))
else:
while True:
password = getpass.getpass(
"Enter the sysadmin password for the subcloud: ")
if len(password) < 1:
print("Password cannot be empty")
continue
confirm = getpass.getpass(
"Re-enter sysadmin password to confirm: ")
if password != confirm:
print("Passwords did not match")
continue
data["sysadmin_password"] = base64.b64encode(
password.encode("utf-8"))
break
password = prompt_for_password()
data["sysadmin_password"] = base64.b64encode(
password.encode("utf-8"))
if parsed_args.install_values is not None:
if parsed_args.bmc_password is not None:
data['bmc_password'] = base64.b64encode(
parsed_args.bmc_password.encode("utf-8"))
else:
while True:
password = getpass.getpass(
"Enter the bmc password for the subcloud: ")
if len(password) < 1:
print("Password cannot be empty")
continue
confirm = getpass.getpass(
"Re-enter bmc password to confirm: ")
if password != confirm:
print("Passwords did not match")
continue
data["bmc_password"] = base64.b64encode(
password.encode("utf-8"))
break
password = prompt_for_password('bmc')
data["bmc_password"] = base64.b64encode(
password.encode("utf-8"))
if parsed_args.group is not None:
data['group_id'] = parsed_args.group
if parsed_args.migrate:
data['migrate'] = 'true'
return dcmanager_client.subcloud_manager.add_subcloud(files=files,
data=data)
@ -340,7 +344,7 @@ class UnmanageSubcloud(base.DCManagerShowOne):
kwargs['management-state'] = 'unmanaged'
try:
return dcmanager_client.subcloud_manager.update_subcloud(
subcloud_ref, **kwargs)
subcloud_ref, files=None, data=kwargs)
except Exception as e:
print(e)
error_msg = "Unable to unmanage subcloud %s" % (subcloud_ref)
@ -360,6 +364,14 @@ class ManageSubcloud(base.DCManagerShowOne):
'subcloud',
help='Name or ID of the subcloud to manage.'
)
parser.add_argument(
'--force',
required=False,
action='store_true',
help='Disregard subcloud availability status, intended for \
some upgrade recovery scenarios.'
)
return parser
def _get_resources(self, parsed_args):
@ -367,9 +379,12 @@ class ManageSubcloud(base.DCManagerShowOne):
dcmanager_client = self.app.client_manager.subcloud_manager
kwargs = dict()
kwargs['management-state'] = 'managed'
if parsed_args.force:
kwargs['force'] = 'true'
try:
return dcmanager_client.subcloud_manager.update_subcloud(
subcloud_ref, **kwargs)
subcloud_ref, files=None, data=kwargs)
except Exception as e:
print(e)
error_msg = "Unable to manage subcloud %s" % (subcloud_ref)
@ -408,26 +423,248 @@ class UpdateSubcloud(base.DCManagerShowOne):
help='Name or ID of subcloud group.'
)
parser.add_argument(
'--install-values',
required=False,
help='YAML file containing subcloud variables required for remote '
'install playbook.'
)
parser.add_argument(
'--bmc-password',
required=False,
help='bmc password of the subcloud to be configured, if not '
'provided you will be prompted. This parameter is only'
' valid if the --install-values are specified.'
)
return parser
def _get_resources(self, parsed_args):
subcloud_ref = parsed_args.subcloud
dcmanager_client = self.app.client_manager.subcloud_manager
kwargs = dict()
files = dict()
data = dict()
if parsed_args.description:
kwargs['description'] = parsed_args.description
data['description'] = parsed_args.description
if parsed_args.location:
kwargs['location'] = parsed_args.location
data['location'] = parsed_args.location
if parsed_args.group:
kwargs['group_id'] = parsed_args.group
if len(kwargs) == 0:
data['group_id'] = parsed_args.group
if parsed_args.install_values:
if not os.path.isfile(parsed_args.install_values):
error_msg = "install-values does not exist: %s" % \
parsed_args.install_values
raise exceptions.DCManagerClientException(error_msg)
files['install_values'] = parsed_args.install_values
if parsed_args.bmc_password is not None:
data['bmc_password'] = base64.b64encode(
parsed_args.bmc_password.encode("utf-8"))
else:
while True:
password = getpass.getpass(
"Enter the bmc password for the subcloud: ")
if len(password) < 1:
print("Password cannot be empty")
continue
confirm = getpass.getpass(
"Re-enter bmc password to confirm: ")
if password != confirm:
print("Passwords did not match")
continue
data["bmc_password"] = base64.b64encode(
password.encode("utf-8"))
break
if len(data) == 0:
error_msg = "Nothing to update"
raise exceptions.DCManagerClientException(error_msg)
try:
return dcmanager_client.subcloud_manager.update_subcloud(
subcloud_ref, **kwargs)
subcloud_ref, files=files, data=data)
except Exception as e:
print(e)
error_msg = "Unable to update subcloud %s" % (subcloud_ref)
raise exceptions.DCManagerClientException(error_msg)
class ReconfigSubcloud(base.DCManagerShowOne):
"""Reconfigure a subcloud."""
def _get_format_function(self):
return detail_format
def get_parser(self, prog_name):
parser = super(ReconfigSubcloud, self).get_parser(prog_name)
parser.add_argument(
'subcloud',
help='Name or ID of the subcloud to update.'
)
parser.add_argument(
'--deploy-config',
required=True,
help='YAML file containing subcloud variables to be passed to the '
'deploy playbook.'
)
parser.add_argument(
'--sysadmin-password',
required=False,
help='sysadmin password of the subcloud to be configured, '
'if not provided you will be prompted.'
)
return parser
def _get_resources(self, parsed_args):
subcloud_ref = parsed_args.subcloud
dcmanager_client = self.app.client_manager.subcloud_manager
files = dict()
data = dict()
# Get the deploy config yaml file
if parsed_args.deploy_config is not None:
if not os.path.isfile(parsed_args.deploy_config):
error_msg = "deploy-config file does not exist: %s" % \
parsed_args.deploy_config
raise exceptions.DCManagerClientException(error_msg)
files['deploy_config'] = parsed_args.deploy_config
# Prompt the user for the subcloud's password if it isn't provided
if parsed_args.sysadmin_password is not None:
data['sysadmin_password'] = base64.b64encode(
parsed_args.sysadmin_password.encode("utf-8"))
else:
password = prompt_for_password()
data["sysadmin_password"] = base64.b64encode(
password.encode("utf-8"))
try:
return dcmanager_client.subcloud_manager.reconfigure_subcloud(
subcloud_ref=subcloud_ref, files=files, data=data)
except Exception:
error_msg = "Unable to reconfigure subcloud %s" % (subcloud_ref)
raise exceptions.DCManagerClientException(error_msg)
class ReinstallSubcloud(base.DCManagerShowOne):
"""Reinstall a subcloud."""
def _get_format_function(self):
return detail_format
def get_parser(self, prog_name):
parser = super(ReinstallSubcloud, self).get_parser(prog_name)
parser.add_argument(
'subcloud',
help='Name or ID of the subcloud to reinstall.'
)
return parser
def _get_resources(self, parsed_args):
subcloud_ref = parsed_args.subcloud
dcmanager_client = self.app.client_manager.subcloud_manager
# Require user to type reinstall to confirm
print("WARNING: This will reinstall the subcloud. "
"All applications and data on the subcloud will be lost.")
confirm = six.moves.input(
"Please type \"reinstall\" to confirm:").strip().lower()
if confirm == 'reinstall':
try:
return dcmanager_client.subcloud_manager.reinstall_subcloud(
subcloud_ref=subcloud_ref)
except Exception:
error_msg = "Unable to reinstall subcloud %s" % (subcloud_ref)
raise exceptions.DCManagerClientException(error_msg)
else:
msg = "Subcloud %s will not be reinstalled" % (subcloud_ref)
raise exceptions.DCManagerClientException(msg)
class RestoreSubcloud(base.DCManagerShowOne):
"""Restore a subcloud."""
def _get_format_function(self):
return detail_format
def get_parser(self, prog_name):
parser = super(RestoreSubcloud, self).get_parser(prog_name)
parser.add_argument(
'--restore-values',
required=True,
help='YAML file containing subcloud restore settings. '
'Can be either a local file path or a URL.'
)
parser.add_argument(
'--sysadmin-password',
required=False,
help='sysadmin password of the subcloud to be restored, '
'if not provided you will be prompted.'
)
parser.add_argument(
'--with-install',
required=False,
action='store_true',
help='option to reinstall the subcloud as part of restore, '
'suitable only for subclouds that can be installed remotely.'
)
parser.add_argument(
'subcloud',
help='Name or ID of the subcloud to update.'
)
return parser
def _get_resources(self, parsed_args):
subcloud_ref = parsed_args.subcloud
dcmanager_client = self.app.client_manager.subcloud_manager
files = dict()
data = dict()
if parsed_args.with_install:
data['with_install'] = 'true'
# Get the restore values yaml file
if not os.path.isfile(parsed_args.restore_values):
error_msg = "restore-values does not exist: %s" % \
parsed_args.restore_values
raise exceptions.DCManagerClientException(error_msg)
files['restore_values'] = parsed_args.restore_values
# Prompt the user for the subcloud's password if it isn't provided
if parsed_args.sysadmin_password is not None:
data['sysadmin_password'] = base64.b64encode(
parsed_args.sysadmin_password.encode("utf-8"))
else:
password = prompt_for_password()
data["sysadmin_password"] = base64.b64encode(
password.encode("utf-8"))
subcloud_list = \
dcmanager_client.subcloud_manager.subcloud_detail(subcloud_ref)
if subcloud_list:
if subcloud_list[0].management_state != 'unmanaged':
error_msg = "Subcloud can not be restored while it is " +\
"still in managed state. Please unmanage " +\
"the subcloud and try again."
raise exceptions.DCManagerClientException(error_msg)
else:
error_msg = subcloud_ref + " not found."
raise exceptions.DCManagerClientException(error_msg)
try:
return dcmanager_client.subcloud_manager.restore_subcloud(
subcloud_ref, files=files, data=data)
except Exception as e:
print(e)
error_msg = "Unable to restore subcloud %s" % (subcloud_ref)
raise exceptions.DCManagerClientException(error_msg)

38
distributedcloud-client/dcmanagerclient/commands/v1/sw_update_manager.py
View File

@ -146,12 +146,28 @@ class CreateSwUpdateStrategy(base.DCManagerShowOne):
help='Do not update any additional subclouds after a failure.'
)
parser.add_argument(
'--force',
required=False,
action='store_true',
help='Disregard subcloud availability status, intended for \
some upgrade recovery scenarios. Subcloud name must be \
specified.'
)
parser.add_argument(
'--group',
required=False,
help='Name or ID of subcloud group to update.'
)
parser.add_argument(
'cloud_name',
nargs='?',
default=None,
help='Name of a single cloud to update.'
)
return parser
def _get_resources(self, parsed_args):
@ -163,8 +179,30 @@ class CreateSwUpdateStrategy(base.DCManagerShowOne):
parsed_args.max_parallel_subclouds
if parsed_args.stop_on_failure:
kwargs['stop-on-failure'] = 'true'
if parsed_args.force:
kwargs['force'] = 'true'
if parsed_args.cloud_name is not None:
kwargs['cloud_name'] = parsed_args.cloud_name
if parsed_args.group is not None:
kwargs['subcloud_group'] = parsed_args.group
if parsed_args.force and not parsed_args.cloud_name:
error_msg = 'The --force option can only be applied to a single ' \
'subcloud. Please specify the subcloud name.'
raise exceptions.DCManagerClientException(error_msg)
if parsed_args.cloud_name and parsed_args.group:
error_msg = 'The cloud_name and group options are mutually ' \
'exclusive.'
raise exceptions.DCManagerClientException(error_msg)
if parsed_args.group and (parsed_args.subcloud_apply_type or
parsed_args.max_parallel_subclouds):
error_msg = 'The --subcloud-apply-type and ' \
'--max-parallel-subclouds options are not ' \
'supported when --group option is applied.'
raise exceptions.DCManagerClientException(error_msg)
return self.get_sw_update_manager().create_sw_update_strategy(**kwargs)

0
distributedcloud-client/dcmanagerclient/openstack/__init__.py
View File

24
distributedcloud-client/dcmanagerclient/openstack/common/__init__.py
View File

@ -1,24 +0,0 @@
#
# 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.
#
# Copyright (c) 2017 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
# of an applicable Wind River license agreement.
#
import six
six.add_move(six.MovedModule('mox', 'mox', 'mox3.mox'))

0
distributedcloud-client/dcmanagerclient/openstack/common/apiclient/__init__.py
View File

231
distributedcloud-client/dcmanagerclient/openstack/common/apiclient/auth.py
View File

@ -1,231 +0,0 @@
# Copyright 2013 OpenStack Foundation
# Copyright 2013 Spanish National Research Council.
# 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.
#
# Copyright (c) 2017 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
# of an applicable Wind River license agreement.
#
# E0202: An attribute inherited from %s hide this method
# pylint: disable=E0202
import abc
import argparse
import os
import six
from stevedore import extension
from dcmanagerclient.openstack.common.apiclient import exceptions
_discovered_plugins = {}
def discover_auth_systems():
"""Discover the available auth-systems.
This won't take into account the old style auth-systems.
"""
global _discovered_plugins
_discovered_plugins = {}
def add_plugin(ext):
_discovered_plugins[ext.name] = ext.plugin
ep_namespace = "dcmanagerclient.openstack.common.apiclient.auth"
mgr = extension.ExtensionManager(ep_namespace)
mgr.map(add_plugin)
def load_auth_system_opts(parser):
"""Load options needed by the available auth-systems into a parser.
This function will try to populate the parser with options from the
available plugins.
"""
group = parser.add_argument_group("Common auth options")
BaseAuthPlugin.add_common_opts(group)
for name, auth_plugin in six.iteritems(_discovered_plugins):
group = parser.add_argument_group(
"Auth-system '%s' options" % name,
conflict_handler="resolve")
auth_plugin.add_opts(group)
def load_plugin(auth_system):
try:
plugin_class = _discovered_plugins[auth_system]
except KeyError:
raise exceptions.AuthSystemNotFound(auth_system)
return plugin_class(auth_system=auth_system)
def load_plugin_from_args(args):
"""Load required plugin and populate it with options.
Try to guess auth system if it is not specified. Systems are tried in
alphabetical order.
:type args: argparse.Namespace
:raises: AuthPluginOptionsMissing
"""
auth_system = args.os_auth_system
if auth_system:
plugin = load_plugin(auth_system)
plugin.parse_opts(args)
plugin.sufficient_options()
return plugin
for plugin_auth_system in sorted(six.iterkeys(_discovered_plugins)):
plugin_class = _discovered_plugins[plugin_auth_system]
plugin = plugin_class()
plugin.parse_opts(args)
try:
plugin.sufficient_options()
except exceptions.AuthPluginOptionsMissing:
continue
return plugin
raise exceptions.AuthPluginOptionsMissing(["auth_system"])
@six.add_metaclass(abc.ABCMeta)
class BaseAuthPlugin(object):
"""Base class for authentication plugins.
An authentication plugin needs to override at least the authenticate
method to be a valid plugin.
"""
auth_system = None
opt_names = []
common_opt_names = [
"auth_system",
"username",
"password",
"tenant_name",
"token",
"auth_url",
]
def __init__(self, auth_system=None, **kwargs):
self.auth_system = auth_system or self.auth_system
self.opts = dict((name, kwargs.get(name))
for name in self.opt_names)
@staticmethod
def _parser_add_opt(parser, opt):
"""Add an option to parser in two variants.
:param opt: option name (with underscores)
"""
dashed_opt = opt.replace("_", "-")
env_var = "OS_%s" % opt.upper()
arg_default = os.environ.get(env_var, "")
arg_help = "Defaults to env[%s]." % env_var
parser.add_argument(
"--os-%s" % dashed_opt,
metavar="<%s>" % dashed_opt,
default=arg_default,
help=arg_help)
parser.add_argument(
"--os_%s" % opt,
metavar="<%s>" % dashed_opt,
help=argparse.SUPPRESS)
@classmethod
def add_opts(cls, parser):
"""Populate the parser with the options for this plugin.
"""
for opt in cls.opt_names:
# use `BaseAuthPlugin.common_opt_names` since it is never
# changed in child classes
if opt not in BaseAuthPlugin.common_opt_names:
cls._parser_add_opt(parser, opt)
@classmethod
def add_common_opts(cls, parser):
"""Add options that are common for several plugins.
"""
for opt in cls.common_opt_names:
cls._parser_add_opt(parser, opt)
@staticmethod
def get_opt(opt_name, args):
"""Return option name and value.
:param opt_name: name of the option, e.g., "username"
:param args: parsed arguments
"""
return (opt_name, getattr(args, "os_%s" % opt_name, None))
def parse_opts(self, args):
"""Parse the actual auth-system options if any.
This method is expected to populate the attribute `self.opts` with a
dict containing the options and values needed to make authentication.
"""
self.opts.update(dict(self.get_opt(opt_name, args)
for opt_name in self.opt_names))
def authenticate(self, http_client):
"""Authenticate using plugin defined method.
The method usually analyses `self.opts` and performs
a request to authentication server.
:param http_client: client object that needs authentication
:type http_client: HTTPClient
:raises: AuthorizationFailure
"""
self.sufficient_options()
self._do_authenticate(http_client)
@abc.abstractmethod
def _do_authenticate(self, http_client):
"""Protected method for authentication.
"""
def sufficient_options(self):
"""Check if all required options are present.
:raises: AuthPluginOptionsMissing
"""
missing = [opt
for opt in self.opt_names
if not self.opts.get(opt)]
if missing:
raise exceptions.AuthPluginOptionsMissing(missing)
@abc.abstractmethod
def token_and_endpoint(self, endpoint_type, service_type):
"""Return token and endpoint.
:param service_type: Service type of the endpoint
:type service_type: string
:param endpoint_type: Type of endpoint.
Possible values: public or publicURL,
internal or internalURL,
admin or adminURL
:type endpoint_type: string
:returns: tuple of token and endpoint strings
:raises: EndpointException
"""

515
distributedcloud-client/dcmanagerclient/openstack/common/apiclient/base.py
View File

@ -1,515 +0,0 @@
# Copyright 2010 Jacob Kaplan-Moss
# Copyright 2011 OpenStack Foundation
# Copyright 2012 Grid Dynamics
# Copyright 2013 OpenStack Foundation
# 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.
#
# Copyright (c) 2017 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
# of an applicable Wind River license agreement.
#
"""
Base utilities to build API operation managers and objects on top of.
"""
# E1102: %s is not callable
# pylint: disable=E1102
import abc
import copy
import six
from six.moves.urllib import parse
from dcmanagerclient.openstack.common.apiclient import exceptions
from dcmanagerclient.openstack.common.gettextutils import _
from dcmanagerclient.openstack.common import strutils
def getid(obj):
"""Return id if argument is a Resource.
Abstracts the common pattern of allowing both an object or an object's ID
(UUID) as a parameter when dealing with relationships.
"""
try:
if obj.uuid:
return obj.uuid
except AttributeError:
pass
try:
return obj.id
except AttributeError:
return obj
# TODO(aababilov): call run_hooks() in HookableMixin's child classes
class HookableMixin(object):
"""Mixin so classes can register and run hooks."""
_hooks_map = {}
@classmethod
def add_hook(cls, hook_type, hook_func):
"""Add a new hook of specified type.
:param cls: class that registers hooks
:param hook_type: hook type, e.g., '__pre_parse_args__'
:param hook_func: hook function
"""
if hook_type not in cls._hooks_map:
cls._hooks_map[hook_type] = []
cls._hooks_map[hook_type].append(hook_func)
@classmethod
def run_hooks(cls, hook_type, *args, **kwargs):
"""Run all hooks of specified type.
:param cls: class that registers hooks
:param hook_type: hook type, e.g., '__pre_parse_args__'
:param args: args to be passed to every hook function
:param kwargs: kwargs to be passed to every hook function
"""
hook_funcs = cls._hooks_map.get(hook_type) or []
for hook_func in hook_funcs:
hook_func(*args, **kwargs)
class BaseManager(HookableMixin):
"""Basic manager type providing common operations.
Managers interact with a particular type of API (servers, flavors, images,
etc.) and provide CRUD operations for them.
"""
resource_class = None
def __init__(self, client):
"""Initializes BaseManager with `client`.
:param client: instance of BaseClient descendant for HTTP requests
"""
super(BaseManager, self).__init__()
self.client = client
def _list(self, url, response_key, obj_class=None, json=None):
"""List the collection.
:param url: a partial URL, e.g., '/servers'
:param response_key: the key to be looked up in response dictionary,
e.g., 'servers'
:param obj_class: class for constructing the returned objects
(self.resource_class will be used by default)
:param json: data that will be encoded as JSON and passed in POST
request (GET will be sent by default)
"""
if json:
body = self.client.post(url, json=json).json()
else:
body = self.client.get(url).json()
if obj_class is None:
obj_class = self.resource_class
data = body[response_key]
# NOTE(ja): keystone returns values as list as {'values': [ ... ]}
# unlike other services which just return the list...
try:
data = data['values']
except (KeyError, TypeError):
pass
return [obj_class(self, res, loaded=True) for res in data if res]
def _get(self, url, response_key):
"""Get an object from collection.
:param url: a partial URL, e.g., '/servers'
:param response_key: the key to be looked up in response dictionary,
e.g., 'server'
"""
body = self.client.get(url).json()
return self.resource_class(self, body[response_key], loaded=True)
def _head(self, url):
"""Retrieve request headers for an object.
:param url: a partial URL, e.g., '/servers'
"""
resp = self.client.head(url)
return resp.status_code == 204
def _post(self, url, json, response_key, return_raw=False):
"""Create an object.
:param url: a partial URL, e.g., '/servers'
:param json: data that will be encoded as JSON and passed in POST
request (GET will be sent by default)
:param response_key: the key to be looked up in response dictionary,
e.g., 'servers'
:param return_raw: flag to force returning raw JSON instead of
Python object of self.resource_class
"""
body = self.client.post(url, json=json).json()
if return_raw:
return body[response_key]
return self.resource_class(self, body[response_key])
def _put(self, url, json=None, response_key=None):
"""Update an object with PUT method.
:param url: a partial URL, e.g., '/servers'
:param json: data that will be encoded as JSON and passed in POST
request (GET will be sent by default)
:param response_key: the key to be looked up in response dictionary,
e.g., 'servers'
"""
resp = self.client.put(url, json=json)
# PUT requests may not return a body
if resp.content:
body = resp.json()
if response_key is not None:
return self.resource_class(self, body[response_key])
else:
return self.resource_class(self, body)
def _patch(self, url, json=None, response_key=None):
"""Update an object with PATCH method.
:param url: a partial URL, e.g., '/servers'
:param json: data that will be encoded as JSON and passed in POST
request (GET will be sent by default)
:param response_key: the key to be looked up in response dictionary,
e.g., 'servers'
"""
body = self.client.patch(url, json=json).json()
if response_key is not None:
return self.resource_class(self, body[response_key])
else:
return self.resource_class(self, body)
def _delete(self, url):
"""Delete an object.
:param url: a partial URL, e.g., '/servers/my-server'
"""
return self.client.delete(url)
@six.add_metaclass(abc.ABCMeta)
class ManagerWithFind(BaseManager):
"""Manager with additional `find()`/`findall()` methods."""
@abc.abstractmethod
def list(self):
pass
def find(self, **kwargs):
"""Find a single item with attributes matching ``**kwargs``.
This isn't very efficient: it loads the entire list then filters on
the Python side.
"""
matches = self.findall(**kwargs)
num_matches = len(matches)
if num_matches == 0:
msg = _("No %(name)s matching %(args)s.") % {
'name': self.resource_class.__name__,
'args': kwargs
}
raise exceptions.NotFound(msg)
elif num_matches > 1:
raise exceptions.NoUniqueMatch()
else:
return matches[0]
def findall(self, **kwargs):
"""Find all items with attributes matching ``**kwargs``.
This isn't very efficient: it loads the entire list then filters on
the Python side.
"""
found = []
searches = kwargs.items()
for obj in self.list():
try:
if all(getattr(obj, attr) == value
for (attr, value) in searches):
found.append(obj)
except AttributeError:
continue
return found
class CrudManager(BaseManager):
"""Base manager class for manipulating entities.
Children of this class are expected to define a `collection_key` and `key`.
- `collection_key`: Usually a plural noun by convention (e.g. `entities`);
used to refer collections in both URL's (e.g. `/v3/entities`) and JSON
objects containing a list of member resources (e.g. `{'entities': [{},
{}, {}]}`).
- `key`: Usually a singular noun by convention (e.g. `entity`); used to
refer to an individual member of the collection.
"""
collection_key = None
key = None
def build_url(self, base_url=None, **kwargs):
"""Builds a resource URL for the given kwargs.
Given an example collection where `collection_key = 'entities'` and
`key = 'entity'`, the following URL's could be generated.
By default, the URL will represent a collection of entities, e.g.::
/entities
If kwargs contains an `entity_id`, then the URL will represent a
specific member, e.g.::
/entities/{entity_id}
:param base_url: if provided, the generated URL will be appended to it
"""
url = base_url if base_url is not None else ''
url += '/%s' % self.collection_key
# do we have a specific entity?
entity_id = kwargs.get('%s_id' % self.key)
if entity_id is not None:
url += '/%s' % entity_id
return url
def _filter_kwargs(self, kwargs):
"""Drop null values and handle ids."""
for key, ref in six.iteritems(kwargs.copy()):
if ref is None:
kwargs.pop(key)
else:
if isinstance(ref, Resource):
kwargs.pop(key)
kwargs['%s_id' % key] = getid(ref)
return kwargs
def create(self, **kwargs):
kwargs = self._filter_kwargs(kwargs)
return self._post(
self.build_url(**kwargs),
{self.key: kwargs},
self.key)
def get(self, **kwargs):
kwargs = self._filter_kwargs(kwargs)
return self._get(
self.build_url(**kwargs),
self.key)
def head(self, **kwargs):
kwargs = self._filter_kwargs(kwargs)
return self._head(self.build_url(**kwargs))
def list(self, base_url=None, **kwargs):
"""List the collection.
:param base_url: if provided, the generated URL will be appended to it
"""
kwargs = self._filter_kwargs(kwargs)
return self._list(
'%(base_url)s%(query)s' % {
'base_url': self.build_url(base_url=base_url, **kwargs),
'query': '?%s' % parse.urlencode(kwargs) if kwargs else '',
},
self.collection_key)
def put(self, base_url=None, **kwargs):
"""Update an element.
:param base_url: if provided, the generated URL will be appended to it
"""
kwargs = self._filter_kwargs(kwargs)
return self._put(self.build_url(base_url=base_url, **kwargs))
def update(self, **kwargs):
kwargs = self._filter_kwargs(kwargs)
params = kwargs.copy()
params.pop('%s_id' % self.key)
return self._patch(
self.build_url(**kwargs),
{self.key: params},
self.key)
def delete(self, **kwargs):
kwargs = self._filter_kwargs(kwargs)
return self._delete(
self.build_url(**kwargs))
def find(self, base_url=None, **kwargs):
"""Find a single item with attributes matching ``**kwargs``.
:param base_url: if provided, the generated URL will be appended to it
"""
kwargs = self._filter_kwargs(kwargs)
rl = self._list(
'%(base_url)s%(query)s' % {
'base_url': self.build_url(base_url=base_url, **kwargs),
'query': '?%s' % parse.urlencode(kwargs) if kwargs else '',
},
self.collection_key)
num = len(rl)
if num == 0:
msg = _("No %(name)s matching %(args)s.") % {
'name': self.resource_class.__name__,
'args': kwargs
}
raise exceptions.NotFound(404, msg)
elif num > 1:
raise exceptions.NoUniqueMatch
else:
return rl[0]
class Extension(HookableMixin):
"""Extension descriptor."""
SUPPORTED_HOOKS = ('__pre_parse_args__', '__post_parse_args__')
manager_class = None
def __init__(self, name, module):
super(Extension, self).__init__()
self.name = name
self.module = module
self._parse_extension_module()
def _parse_extension_module(self):
self.manager_class = None
for attr_name, attr_value in self.module.__dict__.items():
if attr_name in self.SUPPORTED_HOOKS:
self.add_hook(attr_name, attr_value)
else:
try:
if issubclass(attr_value, BaseManager):
self.manager_class = attr_value
except TypeError:
pass
def __repr__(self):
return "<Extension '%s'>" % self.name
class Resource(object):
"""Base class for OpenStack resources (tenant, user, etc.).
This is pretty much just a bag for attributes.
"""
HUMAN_ID = False
NAME_ATTR = 'name'
def __init__(self, manager, info, loaded=False):
"""Populate and bind to a manager.
:param manager: BaseManager object
:param info: dictionary representing resource attributes
:param loaded: prevent lazy-loading if set to True
"""
self.manager = manager
self._info = info
self._add_details(info)
self._loaded = loaded
def __repr__(self):
reprkeys = sorted(k
for k in self.__dict__.keys()
if k[0] != '_' and k != 'manager')
info = ", ".join("%s=%s" % (k, getattr(self, k)) for k in reprkeys)
return "<%s %s>" % (self.__class__.__name__, info)
@property
def human_id(self):
"""Human-readable ID which can be used for bash completion.
"""
if self.NAME_ATTR in self.__dict__ and self.HUMAN_ID:
return strutils.to_slug(getattr(self, self.NAME_ATTR))
return None
def _add_details(self, info):
for (k, v) in six.iteritems(info):
try:
setattr(self, k, v)
self._info[k] = v
except AttributeError:
# In this case we already defined the attribute on the class
pass
def __getattr__(self, k):
if k not in self.__dict__:
# NOTE(bcwaldon): disallow lazy-loading if already loaded once
if not self.is_loaded():
self.get()
return self.__getattr__(k)
raise AttributeError(k)
else:
return self.__dict__[k]
def get(self):
"""Support for lazy loading details.
Some clients, such as novaclient have the option to lazy load the
details, details which can be loaded with this function.
"""
# set_loaded() first ... so if we have to bail, we know we tried.
self.set_loaded(True)
if not hasattr(self.manager, 'get'):
return
new = self.manager.get(self.id)
if new:
self._add_details(new._info)
def __eq__(self, other):
if not isinstance(other, Resource):
return NotImplemented
# two resources of different types are not equal
if not isinstance(other, self.__class__):
return False
if hasattr(self, 'id') and hasattr(other, 'id'):
return self.id == other.id
return self._info == other._info
def is_loaded(self):
return self._loaded
def set_loaded(self, val):
self._loaded = val
def to_dict(self):
return copy.deepcopy(self._info)

370
distributedcloud-client/dcmanagerclient/openstack/common/apiclient/client.py
View File

@ -1,370 +0,0 @@
# Copyright 2010 Jacob Kaplan-Moss
# Copyright 2011 OpenStack Foundation
# Copyright 2011 Piston Cloud Computing, Inc.
# Copyright 2013 Alessio Ababilov
# Copyright 2013 Grid Dynamics
# Copyright 2013 OpenStack Foundation
# 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.
#
# Copyright (c) 2017 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
# of an applicable Wind River license agreement.
#
"""
OpenStack Client interface. Handles the REST calls and responses.
"""
# E0202: An attribute inherited from %s hide this method
# pylint: disable=E0202
import logging
import time
try:
import simplejson as json
except ImportError:
import json
import requests
from dcmanagerclient.openstack.common.apiclient import exceptions
from dcmanagerclient.openstack.common.gettextutils import _
from dcmanagerclient.openstack.common import importutils
_logger = logging.getLogger(__name__)
class HTTPClient(object):
"""This client handles sending HTTP requests to OpenStack servers.
Features:
- share authentication information between several clients to different
services (e.g., for compute and image clients);
- reissue authentication request for expired tokens;
- encode/decode JSON bodies;
- raise exceptions on HTTP errors;
- pluggable authentication;
- store authentication information in a keyring;
- store time spent for requests;
- register clients for particular services, so one can use
`http_client.identity` or `http_client.compute`;
- log requests and responses in a format that is easy to copy-and-paste
into terminal and send the same request with curl.
"""
user_agent = "dcmanagerclient.openstack.common.apiclient"
def __init__(self,
auth_plugin,
region_name=None,
endpoint_type="publicURL",
original_ip=None,
verify=True,
cert=None,
timeout=None,
timings=False,
keyring_saver=None,
debug=False,
user_agent=None,
http=None):
self.auth_plugin = auth_plugin
self.endpoint_type = endpoint_type
self.region_name = region_name
self.original_ip = original_ip
self.timeout = timeout
self.verify = verify
self.cert = cert
self.keyring_saver = keyring_saver
self.debug = debug
self.user_agent = user_agent or self.user_agent
self.times = [] # [("item", starttime, endtime), ...]
self.timings = timings
# requests within the same session can reuse TCP connections from pool
self.http = http or requests.Session()
self.cached_token = None
def _http_log_req(self, method, url, kwargs):
if not self.debug:
return
string_parts = [
"curl -i",
"-X '%s'" % method,
"'%s'" % url,
]
for element in kwargs['headers']:
header = "-H '%s: %s'" % (element, kwargs['headers'][element])
string_parts.append(header)
_logger.debug("REQ: %s" % " ".join(string_parts))
if 'data' in kwargs:
_logger.debug("REQ BODY: %s\n" % (kwargs['data']))
def _http_log_resp(self, resp):
if not self.debug:
return
_logger.debug(
"RESP: [%s] %s\n",
resp.status_code,
resp.headers)
if resp._content_consumed:
_logger.debug(
"RESP BODY: %s\n",
resp.text)
def serialize(self, kwargs):
if kwargs.get('json') is not None:
kwargs['headers']['Content-Type'] = 'application/json'
kwargs['data'] = json.dumps(kwargs['json'])
try:
del kwargs['json']
except KeyError:
pass
def get_timings(self):
return self.times
def reset_timings(self):
self.times = []
def request(self, method, url, **kwargs):
"""Send an http request with the specified characteristics.
Wrapper around `requests.Session.request` to handle tasks such as
setting headers, JSON encoding/decoding, and error handling.
:param method: method of HTTP request
:param url: URL of HTTP request
:param kwargs: any other parameter that can be passed to
requests.Session.request (such as `headers`) or `json`
that will be encoded as JSON and used as `data` argument
"""
kwargs.setdefault("headers", kwargs.get("headers", {}))
kwargs["headers"]["User-Agent"] = self.user_agent
if self.original_ip:
kwargs["headers"]["Forwarded"] = "for=%s;by=%s" % (
self.original_ip, self.user_agent)
if self.timeout is not None:
kwargs.setdefault("timeout", self.timeout)
kwargs.setdefault("verify", self.verify)
if self.cert is not None:
kwargs.setdefault("cert", self.cert)
self.serialize(kwargs)
self._http_log_req(method, url, kwargs)
if self.timings:
start_time = time.time()
resp = self.http.request(method, url, **kwargs)
if self.timings:
self.times.append(("%s %s" % (method, url),
start_time, time.time()))
self._http_log_resp(resp)
if resp.status_code >= 400:
_logger.debug(
"Request returned failure status: %s",
resp.status_code)
raise exceptions.from_response(resp, method, url)
return resp
@staticmethod
def concat_url(endpoint, url):
"""Concatenate endpoint and final URL.
E.g., "http://keystone/v2.0/" and "/tokens" are concatenated to
"http://keystone/v2.0/tokens".
:param endpoint: the base URL
:param url: the final URL
"""
return "%s/%s" % (endpoint.rstrip("/"), url.strip("/"))
def client_request(self, client, method, url, **kwargs):
"""Send an http request using `client`'s endpoint and specified `url`.
If request was rejected as unauthorized (possibly because the token is
expired), issue one authorization attempt and send the request once
again.
:param client: instance of BaseClient descendant
:param method: method of HTTP request
:param url: URL of HTTP request
:param kwargs: any other parameter that can be passed to
`HTTPClient.request`
"""
filter_args = {
"endpoint_type": client.endpoint_type or self.endpoint_type,
"service_type": client.service_type,
}
token, endpoint = (self.cached_token, client.cached_endpoint)
just_authenticated = False
if not (token and endpoint):
try:
token, endpoint = self.auth_plugin.token_and_endpoint(
**filter_args)
except exceptions.EndpointException:
pass
if not (token and endpoint):
self.authenticate()
just_authenticated = True
token, endpoint = self.auth_plugin.token_and_endpoint(
**filter_args)
if not (token and endpoint):
raise exceptions.AuthorizationFailure(
_("Cannot find endpoint or token for request"))
old_token_endpoint = (token, endpoint)
kwargs.setdefault("headers", {})["X-Auth-Token"] = token
self.cached_token = token
client.cached_endpoint = endpoint
# Perform the request once. If we get Unauthorized, then it
# might be because the auth token expired, so try to
# re-authenticate and try again. If it still fails, bail.
try:
return self.request(
method, self.concat_url(endpoint, url), **kwargs)
except exceptions.Unauthorized as unauth_ex:
if just_authenticated:
raise
self.cached_token = None
client.cached_endpoint = None
self.authenticate()
try:
token, endpoint = self.auth_plugin.token_and_endpoint(
**filter_args)
except exceptions.EndpointException:
raise unauth_ex
if (not (token and endpoint) or
old_token_endpoint == (token, endpoint)):
raise unauth_ex
self.cached_token = token
client.cached_endpoint = endpoint
kwargs["headers"]["X-Auth-Token"] = token
return self.request(
method, self.concat_url(endpoint, url), **kwargs)
def add_client(self, base_client_instance):
"""Add a new instance of :class:`BaseClient` descendant.
`self` will store a reference to `base_client_instance`.
Example:
>>> def test_clients():
... from keystoneclient.auth import keystone
... from openstack.common.apiclient import client
... auth = keystone.KeystoneAuthPlugin(
... username="user", password="pass", tenant_name="tenant",
... auth_url="http://auth:5000/v2.0")
... openstack_client = client.HTTPClient(auth)
... # create nova client
... from novaclient.v1_1 import client
... client.Client(openstack_client)
... # create keystone client
... from keystoneclient.v2_0 import client
... client.Client(openstack_client)
... # use them
... openstack_client.identity.tenants.list()
... openstack_client.compute.servers.list()
"""
service_type = base_client_instance.service_type
if service_type and not hasattr(self, service_type):
setattr(self, service_type, base_client_instance)
def authenticate(self):
self.auth_plugin.authenticate(self)
# Store the authentication results in the keyring for later requests
if self.keyring_saver:
self.keyring_saver.save(self)
class BaseClient(object):
"""Top-level object to access the OpenStack API.
This client uses :class:`HTTPClient` to send requests. :class:`HTTPClient`
will handle a bunch of issues such as authentication.
"""
service_type = None
endpoint_type = None # "publicURL" will be used
cached_endpoint = None
def __init__(self, http_client, extensions=None):
self.http_client = http_client
http_client.add_client(self)
# Add in any extensions...
if extensions:
for extension in extensions:
if extension.manager_class:
setattr(self, extension.name,
extension.manager_class(self))
def client_request(self, method, url, **kwargs):
return self.http_client.client_request(
self, method, url, **kwargs)
def head(self, url, **kwargs):
return self.client_request("HEAD", url, **kwargs)
def get(self, url, **kwargs):
return self.client_request("GET", url, **kwargs)
def post(self, url, **kwargs):
return self.client_request("POST", url, **kwargs)
def put(self, url, **kwargs):
return self.client_request("PUT", url, **kwargs)
def delete(self, url, **kwargs):
return self.client_request("DELETE", url, **kwargs)
def patch(self, url, **kwargs):
return self.client_request("PATCH", url, **kwargs)
@staticmethod
def get_class(api_name, version, version_map):
"""Returns the client class for the requested API version
:param api_name: the name of the API, e.g. 'compute', 'image', etc
:param version: the requested API version
:param version_map: a dict of client classes keyed by version
:rtype: a client class for the requested API version
"""
try:
client_path = version_map[str(version)]
except (KeyError, ValueError):
msg = _("Invalid %(api_name)s client version '%(version)s'. "
"Must be one of: %(version_map)s") % \
{'api_name': api_name,
'version': version,
'version_map': ', '.join(version_map.keys())
}
raise exceptions.UnsupportedVersion(msg)
return importutils.import_class(client_path)

473
distributedcloud-client/dcmanagerclient/openstack/common/apiclient/exceptions.py
View File

@ -1,473 +0,0 @@
# Copyright 2010 Jacob Kaplan-Moss
# Copyright 2011 Nebula, Inc.
# Copyright 2013 Alessio Ababilov
# Copyright 2013 OpenStack Foundation
# 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.
#
# Copyright (c) 2017 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
# of an applicable Wind River license agreement.
#
"""
Exception definitions.
"""
import inspect
import sys
import six
from dcmanagerclient.openstack.common.gettextutils import _
class ClientException(Exception):
"""The base exception class for all exceptions this library raises.
"""
pass
class MissingArgs(ClientException):
"""Supplied arguments are not sufficient for calling a function."""
def __init__(self, missing):
self.missing = missing
msg = _("Missing arguments: %s") % ", ".join(missing)
super(MissingArgs, self).__init__(msg)
class ValidationError(ClientException):
"""Error in validation on API client side."""
pass
class UnsupportedVersion(ClientException):
"""User is trying to use an unsupported version of the API."""
pass
class CommandError(ClientException):
"""Error in CLI tool."""
pass
class AuthorizationFailure(ClientException):
"""Cannot authorize API client."""
pass
class ConnectionRefused(ClientException):
"""Cannot connect to API service."""
pass
class AuthPluginOptionsMissing(AuthorizationFailure):
"""Auth plugin misses some options."""
def __init__(self, opt_names):
super(AuthPluginOptionsMissing, self).__init__(
_("Authentication failed. Missing options: %s") %
", ".join(opt_names))
self.opt_names = opt_names
class AuthSystemNotFound(AuthorizationFailure):
"""User has specified a AuthSystem that is not installed."""
def __init__(self, auth_system):
super(AuthSystemNotFound, self).__init__(
_("AuthSystemNotFound: %s") % repr(auth_system))
self.auth_system = auth_system
class NoUniqueMatch(ClientException):
"""Multiple entities found instead of one."""
pass
class EndpointException(ClientException):
"""Something is rotten in Service Catalog."""
pass
class EndpointNotFound(EndpointException):
"""Could not find requested endpoint in Service Catalog."""
pass
class AmbiguousEndpoints(EndpointException):
"""Found more than one matching endpoint in Service Catalog."""
def __init__(self, endpoints=None):
super(AmbiguousEndpoints, self).__init__(
_("AmbiguousEndpoints: %s") % repr(endpoints))
self.endpoints = endpoints
class HttpError(ClientException):
"""The base exception class for all HTTP exceptions."""
http_status = 0
message = _("HTTP Error")
def __init__(self, message=None, details=None,
response=None, request_id=None,
url=None, method=None, http_status=None):
self.http_status = http_status or self.http_status
self.message = message or self.message
self.details = details
self.request_id = request_id
self.response = response
self.url = url
self.method = method
formatted_string = "%s (HTTP %s)" % (self.message, self.http_status)
if request_id:
formatted_string += " (Request-ID: %s)" % request_id
super(HttpError, self).__init__(formatted_string)
class HTTPRedirection(HttpError):
"""HTTP Redirection."""
message = _("HTTP Redirection")
class HTTPClientError(HttpError):
"""Client-side HTTP error.
Exception for cases in which the client seems to have erred.
"""
message = _("HTTP Client Error")
class HttpServerError(HttpError):
"""Server-side HTTP error.
Exception for cases in which the server is aware that it has
erred or is incapable of performing the request.
"""
message = _("HTTP Server Error")
class MultipleChoices(HTTPRedirection):
"""HTTP 300 - Multiple Choices.
Indicates multiple options for the resource that the client may follow.
"""
http_status = 300
message = _("Multiple Choices")
class BadRequest(HTTPClientError):
"""HTTP 400 - Bad Request.
The request cannot be fulfilled due to bad syntax.
"""
http_status = 400
message = _("Bad Request")
class Unauthorized(HTTPClientError):
"""HTTP 401 - Unauthorized.
Similar to 403 Forbidden, but specifically for use when authentication
is required and has failed or has not yet been provided.
"""
http_status = 401
message = _("Unauthorized")
class PaymentRequired(HTTPClientError):
"""HTTP 402 - Payment Required.
Reserved for future use.
"""
http_status = 402
message = _("Payment Required")
class Forbidden(HTTPClientError):
"""HTTP 403 - Forbidden.
The request was a valid request, but the server is refusing to respond
to it.
"""
http_status = 403
message = _("Forbidden")
class NotFound(HTTPClientError):
"""HTTP 404 - Not Found.
The requested resource could not be found but may be available again
in the future.
"""
http_status = 404
message = _("Not Found")
class MethodNotAllowed(HTTPClientError):
"""HTTP 405 - Method Not Allowed.
A request was made of a resource using a request method not supported
by that resource.
"""
http_status = 405
message = _("Method Not Allowed")
class NotAcceptable(HTTPClientError):
"""HTTP 406 - Not Acceptable.
The requested resource is only capable of generating content not
acceptable according to the Accept headers sent in the request.
"""
http_status = 406
message = _("Not Acceptable")
class ProxyAuthenticationRequired(HTTPClientError):
"""HTTP 407 - Proxy Authentication Required.
The client must first authenticate itself with the proxy.
"""
http_status = 407
message = _("Proxy Authentication Required")
class RequestTimeout(HTTPClientError):
"""HTTP 408 - Request Timeout.
The server timed out waiting for the request.
"""
http_status = 408
message = _("Request Timeout")
class Conflict(HTTPClientError):
"""HTTP 409 - Conflict.
Indicates that the request could not be processed because of conflict
in the request, such as an edit conflict.
"""
http_status = 409
message = _("Conflict")
class Gone(HTTPClientError):
"""HTTP 410 - Gone.
Indicates that the resource requested is no longer available and will
not be available again.
"""
http_status = 410
message = _("Gone")
class LengthRequired(HTTPClientError):
"""HTTP 411 - Length Required.
The request did not specify the length of its content, which is
required by the requested resource.
"""
http_status = 411
message = _("Length Required")
class PreconditionFailed(HTTPClientError):
"""HTTP 412 - Precondition Failed.
The server does not meet one of the preconditions that the requester
put on the request.
"""
http_status = 412
message = _("Precondition Failed")
class RequestEntityTooLarge(HTTPClientError):
"""HTTP 413 - Request Entity Too Large.
The request is larger than the server is willing or able to process.
"""
http_status = 413
message = _("Request Entity Too Large")
def __init__(self, *args, **kwargs):
try:
self.retry_after = int(kwargs.pop('retry_after'))
except (KeyError, ValueError):
self.retry_after = 0
super(RequestEntityTooLarge, self).__init__(*args, **kwargs)
class RequestUriTooLong(HTTPClientError):
"""HTTP 414 - Request-URI Too Long.
The URI provided was too long for the server to process.
"""
http_status = 414
message = _("Request-URI Too Long")
class UnsupportedMediaType(HTTPClientError):
"""HTTP 415 - Unsupported Media Type.
The request entity has a media type which the server or resource does
not support.
"""
http_status = 415
message = _("Unsupported Media Type")
class RequestedRangeNotSatisfiable(HTTPClientError):
"""HTTP 416 - Requested Range Not Satisfiable.
The client has asked for a portion of the file, but the server cannot
supply that portion.
"""
http_status = 416
message = _("Requested Range Not Satisfiable")
class ExpectationFailed(HTTPClientError):
"""HTTP 417 - Expectation Failed.
The server cannot meet the requirements of the Expect request-header field.
"""
http_status = 417
message = _("Expectation Failed")
class UnprocessableEntity(HTTPClientError):
"""HTTP 422 - Unprocessable Entity.
The request was well-formed but was unable to be followed due to semantic
errors.
"""
http_status = 422
message = _("Unprocessable Entity")
class InternalServerError(HttpServerError):
"""HTTP 500 - Internal Server Error.
A generic error message, given when no more specific message is suitable.
"""
http_status = 500
message = _("Internal Server Error")
# NotImplemented is a python keyword.
class HttpNotImplemented(HttpServerError):
"""HTTP 501 - Not Implemented.
The server either does not recognize the request method, or it lacks
the ability to fulfill the request.
"""
http_status = 501
message = _("Not Implemented")
class BadGateway(HttpServerError):
"""HTTP 502 - Bad Gateway.
The server was acting as a gateway or proxy and received an invalid
response from the upstream server.
"""
http_status = 502
message = _("Bad Gateway")
class ServiceUnavailable(HttpServerError):
"""HTTP 503 - Service Unavailable.
The server is currently unavailable.
"""
http_status = 503
message = _("Service Unavailable")
class GatewayTimeout(HttpServerError):
"""HTTP 504 - Gateway Timeout.
The server was acting as a gateway or proxy and did not receive a timely
response from the upstream server.
"""
http_status = 504
message = _("Gateway Timeout")
class HttpVersionNotSupported(HttpServerError):
"""HTTP 505 - HttpVersion Not Supported.
The server does not support the HTTP protocol version used in the request.
"""
http_status = 505
message = _("HTTP Version Not Supported")
# _code_map contains all the classes that have http_status attribute.
_code_map = dict(
(getattr(obj, 'http_status', None), obj)
for name, obj in six.iteritems(vars(sys.modules[__name__]))
if inspect.isclass(obj) and getattr(obj, 'http_status', False)
)
def from_response(response, method, url):
"""Returns an instance of :class:`HttpError` or subclass based on response.
:param response: instance of `requests.Response` class
:param method: HTTP method used for request
:param url: URL used for request
"""
req_id = response.headers.get("x-openstack-request-id")
# NOTE(hdd) true for older versions of nova and cinder
if not req_id:
req_id = response.headers.get("x-compute-request-id")
kwargs = {
"http_status": response.status_code,
"response": response,
"method": method,
"url": url,
"request_id": req_id,
}
if "retry-after" in response.headers:
kwargs["retry_after"] = response.headers["retry-after"]
content_type = response.headers.get("Content-Type", "")
if content_type.startswith("application/json"):
try:
body = response.json()
except ValueError:
pass
else:
if isinstance(body, dict):
error = list(body.values())[0]
kwargs["message"] = error.get("message")
kwargs["details"] = error.get("details")
elif content_type.startswith("text/"):
kwargs["details"] = response.text
try:
cls = _code_map[response.status_code]
except KeyError:
if 500 <= response.status_code < 600:
cls = HttpServerError
elif 400 <= response.status_code < 500:
cls = HTTPClientError
else:
cls = HttpError
return cls(**kwargs)

184
distributedcloud-client/dcmanagerclient/openstack/common/apiclient/fake_client.py
View File

@ -1,184 +0,0 @@
# Copyright 2013 OpenStack Foundation
# 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.
#
# Copyright (c) 2017 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
# of an applicable Wind River license agreement.
#
"""
A fake server that "responds" to API methods with pre-canned responses.
All of these responses come from the spec, so if for some reason the spec's
wrong the tests might raise AssertionError. I've indicated in comments the
places where actual behavior differs from the spec.
"""
# W0102: Dangerous default value %s as argument
# pylint: disable=W0102
import json
import requests
import six
from six.moves.urllib import parse
from dcmanagerclient.openstack.common.apiclient import client
def assert_has_keys(dct, required=None, optional=None):
required = required or []
optional = optional or []
for k in required:
try:
assert k in dct
except AssertionError:
extra_keys = set(dct.keys()).difference(set(required + optional))
raise AssertionError("found unexpected keys: %s" %
list(extra_keys))
class TestResponse(requests.Response):
"""Wrap requests.Response and provide a convenient initialization.
"""
def __init__(self, data):
super(TestResponse, self).__init__()
self._content_consumed = True
if isinstance(data, dict):
self.status_code = data.get('status_code', 200)
# Fake the text attribute to streamline Response creation
text = data.get('text', "")
if isinstance(text, (dict, list)):
self._content = json.dumps(text)
default_headers = {
"Content-Type": "application/json",
}
else:
self._content = text
default_headers = {}
if six.PY3 and isinstance(self._content, six.string_types):
self._content = self._content.encode('utf-8', 'strict')
self.headers = data.get('headers') or default_headers
else:
self.status_code = data
def __eq__(self, other):
return (self.status_code == other.status_code and
self.headers == other.headers and
self._content == other._content)
class FakeHTTPClient(client.HTTPClient):
def __init__(self, *args, **kwargs):
self.callstack = []
self.fixtures = kwargs.pop("fixtures", None) or {}
if not args and not ("auth_plugin" in kwargs):
args = (None,)
super(FakeHTTPClient, self).__init__(*args, **kwargs)
def assert_called(self, method, url, body=None, pos=-1):
"""Assert than an API method was just called.
"""
expected = (method, url)
called = self.callstack[pos][0:2]
assert self.callstack, \
"Expected %s %s but no calls were made." % expected
assert expected == called, 'Expected %s %s; got %s %s' % \
(expected + called)
if body is not None:
if self.callstack[pos][3] != body:
raise AssertionError('%r != %r' %
(self.callstack[pos][3], body))
def assert_called_anytime(self, method, url, body=None):
"""Assert than an API method was called anytime in the test.
"""
expected = (method, url)
assert self.callstack, \
"Expected %s %s but no calls were made." % expected
found = False
entry = None
for entry in self.callstack:
if expected == entry[0:2]:
found = True
break
assert found, 'Expected %s %s; got %s' % \
(method, url, self.callstack)
if body is not None:
assert entry[3] == body, "%s != %s" % (entry[3], body)
self.callstack = []
def clear_callstack(self):
self.callstack = []
def authenticate(self):
pass
def client_request(self, client, method, url, **kwargs):
# Check that certain things are called correctly
if method in ["GET", "DELETE"]:
assert "json" not in kwargs
# Note the call
self.callstack.append(
(method,
url,
kwargs.get("headers") or {},
kwargs.get("json") or kwargs.get("data")))
try:
fixture = self.fixtures[url][method]
except KeyError:
pass
else:
return TestResponse({"headers": fixture[0],
"text": fixture[1]})
# Call the method
args = parse.parse_qsl(parse.urlparse(url)[4])
kwargs.update(args)
munged_url = url.rsplit('?', 1)[0]
munged_url = munged_url.strip('/').replace('/', '_').replace('.', '_')
munged_url = munged_url.replace('-', '_')
callback = "%s_%s" % (method.lower(), munged_url)
if not hasattr(self, callback):
raise AssertionError('Called unknown API method: %s %s, '
'expected fakes method name: %s' %
(method, url, callback))
resp = getattr(self, callback)(**kwargs)
if len(resp) == 3:
status, headers, body = resp
else:
status, body = resp
headers = {}
return TestResponse({
"status_code": status,
"text": body,
"headers": headers,
})

319
distributedcloud-client/dcmanagerclient/openstack/common/cliutils.py
View File

@ -1,319 +0,0 @@
# Copyright 2012 Red Hat, Inc.
#
# 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.
#
# Copyright (c) 2017 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
# of an applicable Wind River license agreement.
#
# W0603: Using the global statement
# W0621: Redefining name %s from outer scope
# pylint: disable=W0603,W0621
from __future__ import print_function
import getpass
import inspect
import os
import sys
import textwrap
import prettytable
import six
from six import moves
from dcmanagerclient.openstack.common.apiclient import exceptions
from dcmanagerclient.openstack.common.gettextutils import _
from dcmanagerclient.openstack.common import strutils
from dcmanagerclient.openstack.common import uuidutils
def validate_args(fn, *args, **kwargs):
"""Check that the supplied args are sufficient for calling a function.
>>> validate_args(lambda a: None)
Traceback (most recent call last):
...
MissingArgs: Missing argument(s): a
>>> validate_args(lambda a, b, c, d: None, 0, c=1)
Traceback (most recent call last):
...
MissingArgs: Missing argument(s): b, d
:param fn: the function to check
:param arg: the positional arguments supplied
:param kwargs: the keyword arguments supplied
"""
argspec = inspect.getargspec(fn)
num_defaults = len(argspec.defaults or [])
required_args = argspec.args[:len(argspec.args) - num_defaults]
def isbound(method):
return getattr(method, '__self__', None) is not None
if isbound(fn):
required_args.pop(0)
missing = [arg for arg in required_args if arg not in kwargs]
missing = missing[len(args):]
if missing:
raise exceptions.MissingArgs(missing)
def arg(*args, **kwargs):
"""Decorator for CLI args.
Example:
>>> @arg("name", help="Name of the new entity")
... def entity_create(args):
... pass
"""
def _decorator(func):
add_arg(func, *args, **kwargs)
return func
return _decorator
def env(*args, **kwargs):
"""Returns the first environment variable set.
If all are empty, defaults to '' or keyword arg `default`.
"""
for arg in args:
value = os.environ.get(arg)
if value:
return value
return kwargs.get('default', '')
def add_arg(func, *args, **kwargs):
"""Bind CLI arguments to a shell.py `do_foo` function."""
if not hasattr(func, 'arguments'):
func.arguments = []
# NOTE(sirp): avoid dups that can occur when the module is shared across
# tests.
if (args, kwargs) not in func.arguments:
# Because of the semantics of decorator composition if we just append
# to the options list positional options will appear to be backwards.
func.arguments.insert(0, (args, kwargs))
def unauthenticated(func):
"""Adds 'unauthenticated' attribute to decorated function.
Usage:
>>> @unauthenticated
... def mymethod(f):
... pass
"""
func.unauthenticated = True
return func
def isunauthenticated(func):
"""Checks if the function does not require authentication.
Mark such functions with the `@unauthenticated` decorator.
:returns: bool
"""
return getattr(func, 'unauthenticated', False)
def print_list(objs, fields, formatters=None, sortby_index=0,
mixed_case_fields=None):
"""Print a list or objects as a table, one row per object.
:param objs: iterable of :class:`Resource`
:param fields: attributes that correspond to columns, in order
:param formatters: `dict` of callables for field formatting
:param sortby_index: index of the field for sorting table rows
:param mixed_case_fields: fields corresponding to object attributes that
have mixed case names (e.g., 'serverId')
"""
formatters = formatters or {}
mixed_case_fields = mixed_case_fields or []
if sortby_index is None:
kwargs = {}
else:
kwargs = {'sortby': fields[sortby_index]}
pt = prettytable.PrettyTable(fields, caching=False)
pt.align = 'l'
for o in objs:
row = []
for field in fields:
if field in formatters:
row.append(formatters[field](o))
else:
if field in mixed_case_fields:
field_name = field.replace(' ', '_')
else:
field_name = field.lower().replace(' ', '_')
data = getattr(o, field_name, '')
row.append(data)
pt.add_row(row)
print(strutils.safe_encode(pt.get_string(**kwargs)))
def print_dict(dct, dict_property="Property", wrap=0):
"""Print a `dict` as a table of two columns.
:param dct: `dict` to print
:param dict_property: name of the first column
:param wrap: wrapping for the second column
"""
pt = prettytable.PrettyTable([dict_property, 'Value'], caching=False)
pt.align = 'l'
for k, v in six.iteritems(dct):
# convert dict to str to check length
if isinstance(v, dict):
v = six.text_type(v)
if wrap > 0:
v = textwrap.fill(six.text_type(v), wrap)
# if value has a newline, add in multiple rows
# e.g. fault with stacktrace
if v and isinstance(v, six.string_types) and r'\n' in v:
lines = v.strip().split(r'\n')
col1 = k
for line in lines:
pt.add_row([col1, line])
col1 = ''
else:
pt.add_row([k, v])
print(strutils.safe_encode(pt.get_string()))
def get_password(max_password_prompts=3):
"""Read password from TTY."""
verify = strutils.bool_from_string(env("OS_VERIFY_PASSWORD"))
pw = None
if hasattr(sys.stdin, "isatty") and sys.stdin.isatty():
# Check for Ctrl-D
try:
for __ in moves.range(max_password_prompts):
pw1 = getpass.getpass("OS Password: ")
if verify:
pw2 = getpass.getpass("Please verify: ")
else:
pw2 = pw1
if pw1 == pw2 and pw1:
pw = pw1
break
except EOFError:
pass
return pw
def find_resource(manager, name_or_id, **find_args):
"""Look for resource in a given manager.
Used as a helper for the _find_* methods.
Example:
def _find_hypervisor(cs, hypervisor):
#Get a hypervisor by name or ID.
return cliutils.find_resource(cs.hypervisors, hypervisor)
"""
# first try to get entity as integer id
try:
return manager.get(int(name_or_id))
except (TypeError, ValueError, exceptions.NotFound):
pass
# now try to get entity as uuid
try:
if six.PY2:
tmp_id = strutils.safe_encode(name_or_id)
else:
tmp_id = strutils.safe_decode(name_or_id)
if uuidutils.is_uuid_like(tmp_id):
return manager.get(tmp_id)
except (TypeError, ValueError, exceptions.NotFound):
pass
# for str id which is not uuid
if getattr(manager, 'is_alphanum_id_allowed', False):
try:
return manager.get(name_or_id)
except exceptions.NotFound:
pass
try:
try:
return manager.find(human_id=name_or_id, **find_args)
except exceptions.NotFound:
pass
# finally try to find entity by name
try:
resource = getattr(manager, 'resource_class', None)
name_attr = resource.NAME_ATTR if resource else 'name'
kwargs = {name_attr: name_or_id}
kwargs.update(find_args)
return manager.find(**kwargs)
except exceptions.NotFound:
msg = _("No %(name)s with a name or "
"ID of '%(name_or_id)s' exists.") % \
{
"name": manager.resource_class.__name__.lower(),
"name_or_id": name_or_id
}
raise exceptions.CommandError(msg)
except exceptions.NoUniqueMatch:
msg = _("Multiple %(name)s matches found for "
"'%(name_or_id)s', use an ID to be more specific.") % \
{
"name": manager.resource_class.__name__.lower(),
"name_or_id": name_or_id
}
raise exceptions.CommandError(msg)
def service_type(stype):
"""Adds 'service_type' attribute to decorated function.
Usage:
@service_type('volume')
def mymethod(f):
...
"""
def inner(f):
f.service_type = stype
return f
return inner
def get_service_type(f):
"""Retrieves service type from function."""
return getattr(f, 'service_type', None)
def pretty_choice_list(l):
return ', '.join("'%s'" % i for i in l)
def exit(msg=''):
if msg:
print (msg, file=sys.stderr)
sys.exit(1)

506
distributedcloud-client/dcmanagerclient/openstack/common/gettextutils.py
View File

@ -1,506 +0,0 @@
# Copyright 2012 Red Hat, Inc.
# Copyright 2013 IBM Corp.
# 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.
#
# Copyright (c) 2017 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
# of an applicable Wind River license agreement.
#
"""
gettext for openstack-common modules.
Usual usage in an openstack.common module:
from dcmanagerclient.openstack.common.gettextutils import _
"""
import copy
import functools
import gettext
import locale
from logging import handlers
import os
from babel import localedata
import six
_AVAILABLE_LANGUAGES = {}
# FIXME(dhellmann): Remove this when moving to oslo.i18n.
USE_LAZY = False
class TranslatorFactory(object):
"""Create translator functions
"""
def __init__(self, domain, lazy=False, localedir=None):
"""Establish a set of translation functions for the domain.
:param domain: Name of translation domain,
specifying a message catalog.
:type domain: str
:param lazy: Delays translation until a message is emitted.
Defaults to False.
:type lazy: Boolean
:param localedir: Directory with translation catalogs.
:type localedir: str
"""
self.domain = domain
self.lazy = lazy
if localedir is None:
localedir = os.environ.get(domain.upper() + '_LOCALEDIR')
self.localedir = localedir
def _make_translation_func(self, domain=None):
"""Return a new translation function ready for use.
Takes into account whether or not lazy translation is being
done.
The domain can be specified to override the default from the
factory, but the localedir from the factory is always used
because we assume the log-level translation catalogs are
installed in the same directory as the main application
catalog.
"""
if domain is None:
domain = self.domain
if self.lazy:
return functools.partial(Message, domain=domain)
t = gettext.translation(
domain,
localedir=self.localedir,
fallback=True,
)
if six.PY3:
return t.gettext
return t.ugettext
@property
def primary(self):
"The default translation function."
return self._make_translation_func()
def _make_log_translation_func(self, level):
return self._make_translation_func(self.domain + '-log-' + level)
@property
def log_info(self):
"Translate info-level log messages."
return self._make_log_translation_func('info')
@property
def log_warning(self):
"Translate warning-level log messages."
return self._make_log_translation_func('warning')
@property
def log_error(self):
"Translate error-level log messages."
return self._make_log_translation_func('error')
@property
def log_critical(self):
"Translate critical-level log messages."
return self._make_log_translation_func('critical')
# NOTE(dhellmann): When this module moves out of the incubator into
# oslo.i18n, these global variables can be moved to an integration
# module within each application.
# Create the global translation functions.
_translators = TranslatorFactory('dcmanagerclient')
# The primary translation function using the well-known name "_"
_ = _translators.primary
# Translators for log levels.
#
# The abbreviated names are meant to reflect the usual use of a short
# name like '_'. The "L" is for "log" and the other letter comes from
# the level.
_LI = _translators.log_info
_LW = _translators.log_warning
_LE = _translators.log_error
_LC = _translators.log_critical
# NOTE(dhellmann): End of globals that will move to the application's
# integration module.
def enable_lazy():
"""Convenience function for configuring _() to use lazy gettext
Call this at the start of execution to enable the gettextutils._
function to use lazy gettext functionality. This is useful if
your project is importing _ directly instead of using the
gettextutils.install() way of importing the _ function.
"""
# FIXME(dhellmann): This function will be removed in oslo.i18n,
# because the TranslatorFactory makes it superfluous.
global _, _LI, _LW, _LE, _LC, USE_LAZY
tf = TranslatorFactory('dcmanagerclient', lazy=True)
_ = tf.primary
_LI = tf.log_info
_LW = tf.log_warning
_LE = tf.log_error
_LC = tf.log_critical
USE_LAZY = True
def install(domain, lazy=False):
"""Install a _() function using the given translation domain.
Given a translation domain, install a _() function using gettext's
install() function.
The main difference from gettext.install() is that we allow
overriding the default localedir (e.g. /usr/share/locale) using
a translation-domain-specific environment variable (e.g.
NOVA_LOCALEDIR).
:param domain: the translation domain
:param lazy: indicates whether or not to install the lazy _() function.
The lazy _() introduces a way to do deferred translation
of messages by installing a _ that builds Message objects,
instead of strings, which can then be lazily translated into
any available locale.
"""
if lazy:
from six import moves
tf = TranslatorFactory(domain, lazy=True)
moves.builtins.__dict__['_'] = tf.primary
else:
localedir = '%s_LOCALEDIR' % domain.upper()
if six.PY3:
gettext.install(domain,
localedir=os.environ.get(localedir))
else:
gettext.install(domain,
localedir=os.environ.get(localedir),
unicode=True)
class Message(six.text_type):
"""A Message object is a unicode object that can be translated.
Translation of Message is done explicitly using the translate() method.
For all non-translation intents and purposes, a Message is simply unicode,
and can be treated as such.
"""
def __new__(cls, msgid, msgtext=None, params=None,
domain='dcmanagerclient', *args):
"""Create a new Message object.
In order for translation to work gettext requires a message ID, this
msgid will be used as the base unicode text. It is also possible
for the msgid and the base unicode text to be different by passing
the msgtext parameter.
"""
# If the base msgtext is not given, we use the default translation
# of the msgid (which is in English) just in case the system locale is
# not English, so that the base text will be in that locale by default.
if not msgtext:
msgtext = Message._translate_msgid(msgid, domain)
# We want to initialize the parent unicode with the actual object that
# would have been plain unicode if 'Message' was not enabled.
msg = super(Message, cls).__new__(cls, msgtext)
msg.msgid = msgid
msg.domain = domain
msg.params = params
return msg
def translate(self, desired_locale=None):
"""Translate this message to the desired locale.
:param desired_locale: The desired locale to translate the message to,
if no locale is provided the message will be
translated to the system's default locale.
:returns: the translated message in unicode
"""
translated_message = Message._translate_msgid(self.msgid,
self.domain,
desired_locale)
if self.params is None:
# No need for more translation
return translated_message
# This Message object may have been formatted with one or more
# Message objects as substitution arguments, given either as a single
# argument, part of a tuple, or as one or more values in a dictionary.
# When translating this Message we need to translate those Messages too
translated_params = _translate_args(self.params, desired_locale)
translated_message = translated_message % translated_params
return translated_message
@staticmethod
def _translate_msgid(msgid, domain, desired_locale=None):
if not desired_locale:
system_locale = locale.getdefaultlocale()
# If the system locale is not available to the runtime use English
if not system_locale[0]:
desired_locale = 'en_US'
else:
desired_locale = system_locale[0]
locale_dir = os.environ.get(domain.upper() + '_LOCALEDIR')
lang = gettext.translation(domain,
localedir=locale_dir,
languages=[desired_locale],
fallback=True)
if six.PY3:
translator = lang.gettext
else:
translator = lang.ugettext
translated_message = translator(msgid)
return translated_message
def __mod__(self, other):
# When we mod a Message we want the actual operation to be performed
# by the parent class (i.e. unicode()), the only thing we do here is
# save the original msgid and the parameters in case of a translation
params = self._sanitize_mod_params(other)
unicode_mod = super(Message, self).__mod__(params)
modded = Message(self.msgid,
msgtext=unicode_mod,
params=params,
domain=self.domain)
return modded
def _sanitize_mod_params(self, other):
"""Sanitize the object being modded with this Message.
- Add support for modding 'None' so translation supports it
- Trim the modded object, which can be a large dictionary, to only
those keys that would actually be used in a translation
- Snapshot the object being modded, in case the message is
translated, it will be used as it was when the Message was created
"""
if other is None:
params = (other,)
elif isinstance(other, dict):
# Merge the dictionaries
# Copy each item in case one does not support deep copy.
params = {}
if isinstance(self.params, dict):
for key, val in self.params.items():
params[key] = self._copy_param(val)
for key, val in other.items():
params[key] = self._copy_param(val)
else:
params = self._copy_param(other)
return params
def _copy_param(self, param):
try:
return copy.deepcopy(param)
except Exception:
# Fallback to casting to unicode this will handle the
# python code-like objects that can't be deep-copied
return six.text_type(param)
def __add__(self, other):
msg = _('Message objects do not support addition.')
raise TypeError(msg)
def __radd__(self, other):
return self.__add__(other)
if six.PY2:
def __str__(self):
# NOTE(luisg): Logging in python 2.6 tries to str() log records,
# and it expects specifically a UnicodeError in order to proceed.
msg = _('Message objects do not support str() because they may '
'contain non-ascii characters. '
'Please use unicode() or translate() instead.')
raise UnicodeError(msg)
def get_available_languages(domain):
"""Lists the available languages for the given translation domain.
:param domain: the domain to get languages for
"""
if domain in _AVAILABLE_LANGUAGES:
return copy.copy(_AVAILABLE_LANGUAGES[domain])
localedir = '%s_LOCALEDIR' % domain.upper()
find = lambda x: gettext.find(domain,
localedir=os.environ.get(localedir),
languages=[x])
# NOTE(mrodden): en_US should always be available (and first in case
# order matters) since our in-line message strings are en_US
language_list = ['en_US']
# NOTE(luisg): Babel <1.0 used a function called list(), which was
# renamed to locale_identifiers() in >=1.0, the requirements master list
# requires >=0.9.6, uncapped, so defensively work with both. We can remove
# this check when the master list updates to >=1.0, and update all projects
list_identifiers = (getattr(localedata, 'list', None) or
getattr(localedata, 'locale_identifiers'))
locale_identifiers = list_identifiers()
for i in locale_identifiers:
if find(i) is not None:
language_list.append(i)
# NOTE(luisg): Babel>=1.0,<1.3 has a bug where some OpenStack supported
# locales (e.g. 'zh_CN', and 'zh_TW') aren't supported even though they
# are perfectly legitimate locales:
# https://github.com/mitsuhiko/babel/issues/37
# In Babel 1.3 they fixed the bug and they support these locales, but
# they are still not explicitly "listed" by locale_identifiers().
# That is why we add the locales here explicitly if necessary so that
# they are listed as supported.
aliases = {'zh': 'zh_CN',
'zh_Hant_HK': 'zh_HK',
'zh_Hant': 'zh_TW',
'fil': 'tl_PH'}
for (loc, alias) in six.iteritems(aliases):
if loc in language_list and alias not in language_list:
language_list.append(alias)
_AVAILABLE_LANGUAGES[domain] = language_list
return copy.copy(language_list)
def translate(obj, desired_locale=None):
"""Gets the translated unicode representation of the given object.
If the object is not translatable it is returned as-is.
If the locale is None the object is translated to the system locale.
:param obj: the object to translate
:param desired_locale: the locale to translate the message to, if None the
default system locale will be used
:returns: the translated object in unicode, or the original object if
it could not be translated
"""
message = obj
if not isinstance(message, Message):
# If the object to translate is not already translatable,
# let's first get its unicode representation
message = six.text_type(obj)
if isinstance(message, Message):
# Even after unicoding() we still need to check if we are
# running with translatable unicode before translating
return message.translate(desired_locale)
return obj
def _translate_args(args, desired_locale=None):
"""Translates all the translatable elements of the given arguments object.
This method is used for translating the translatable values in method
arguments which include values of tuples or dictionaries.
If the object is not a tuple or a dictionary the object itself is
translated if it is translatable.
If the locale is None the object is translated to the system locale.
:param args: the args to translate
:param desired_locale: the locale to translate the args to, if None the
default system locale will be used
:returns: a new args object with the translated contents of the original
"""
if isinstance(args, tuple):
return tuple(translate(v, desired_locale) for v in args)
if isinstance(args, dict):
translated_dict = {}
for (k, v) in six.iteritems(args):
translated_v = translate(v, desired_locale)
translated_dict[k] = translated_v
return translated_dict
return translate(args, desired_locale)
class TranslationHandler(handlers.MemoryHandler):
"""Handler that translates records before logging them.
The TranslationHandler takes a locale and a target logging.Handler object
to forward LogRecord objects to after translating them. This handler
depends on Message objects being logged, instead of regular strings.
The handler can be configured declaratively in the logging.conf as follows:
[handlers]
keys = translatedlog, translator
[handler_translatedlog]
class = handlers.WatchedFileHandler
args = ('/var/log/api-localized.log',)
formatter = context
[handler_translator]
class = openstack.common.log.TranslationHandler
target = translatedlog
args = ('zh_CN',)
If the specified locale is not available in the system, the handler will
log in the default locale.
"""
def __init__(self, locale=None, target=None):
"""Initialize a TranslationHandler
:param locale: locale to use for translating messages
:param target: logging.Handler object to forward
LogRecord objects to after translation
"""
# NOTE(luisg): In order to allow this handler to be a wrapper for
# other handlers, such as a FileHandler, and still be able to
# configure it using logging.conf, this handler has to extend
# MemoryHandler because only the MemoryHandlers' logging.conf
# parsing is implemented such that it accepts a target handler.
handlers.MemoryHandler.__init__(self, capacity=0, target=target)
self.locale = locale
def setFormatter(self, fmt):
self.target.setFormatter(fmt)
def emit(self, record):
# We save the message from the original record to restore it
# after translation, so other handlers are not affected by this
original_msg = record.msg
original_args = record.args
try:
self._translate_and_log_record(record)
finally:
record.msg = original_msg
record.args = original_args
def _translate_and_log_record(self, record):
record.msg = translate(record.msg, self.locale)
# In addition to translating the message, we also need to translate
# arguments that were passed to the log method that were not part
# of the main message e.g., log.info(_('Some message %s'), this_one))
record.args = _translate_args(record.args, self.locale)
self.target.emit(record)

80
distributedcloud-client/dcmanagerclient/openstack/common/importutils.py
View File

@ -1,80 +0,0 @@
# Copyright 2011 OpenStack Foundation.
# 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.
#
# Copyright (c) 2017 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
# of an applicable Wind River license agreement.
#
"""
Import related utilities and helper functions.
"""
import sys
import traceback
def import_class(import_str):
"""Returns a class from a string including module and class."""
mod_str, _sep, class_str = import_str.rpartition('.')
__import__(mod_str)
try:
return getattr(sys.modules[mod_str], class_str)
except AttributeError:
raise ImportError('Class %s cannot be found (%s)' %
(class_str,
traceback.format_exception(*sys.exc_info())))
def import_object(import_str, *args, **kwargs):
"""Import a class and return an instance of it."""
return import_class(import_str)(*args, **kwargs)
def import_object_ns(name_space, import_str, *args, **kwargs):
"""Tries to import object from default namespace.
Imports a class and return an instance of it, first by trying
to find the class in a default namespace, then failing back to
a full path if not found in the default namespace.
"""
import_value = "%s.%s" % (name_space, import_str)
try:
return import_class(import_value)(*args, **kwargs)
except ImportError:
return import_class(import_str)(*args, **kwargs)
def import_module(import_str):
"""Import a module."""
__import__(import_str)
return sys.modules[import_str]
def import_versioned_module(version, submodule=None):
module = 'dcmanagerclient.v%s' % version
if submodule:
module = '.'.join((module, submodule))
return import_module(module)
def try_import(import_str, default=None):
"""Try to import a module and if it fails return default."""
try:
return import_module(import_str)
except ImportError:
return default

247
distributedcloud-client/dcmanagerclient/openstack/common/strutils.py
View File

@ -1,247 +0,0 @@
# Copyright 2011 OpenStack Foundation.
# 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.
#
# Copyright (c) 2017 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
# of an applicable Wind River license agreement.
#
"""
System-level utilities and helper functions.
"""
import math
import re
import sys
import unicodedata
import six
from dcmanagerclient.openstack.common.gettextutils import _
UNIT_PREFIX_EXPONENT = {
'k': 1,
'K': 1,
'Ki': 1,
'M': 2,
'Mi': 2,
'G': 3,
'Gi': 3,
'T': 4,
'Ti': 4,
}
UNIT_SYSTEM_INFO = {
'IEC': (1024, re.compile(r'(^[-+]?\d*\.?\d+)([KMGT]i?)?(b|bit|B)$')),
'SI': (1000, re.compile(r'(^[-+]?\d*\.?\d+)([kMGT])?(b|bit|B)$')),
}
TRUE_STRINGS = ('1', 't', 'true', 'on', 'y', 'yes')
FALSE_STRINGS = ('0', 'f', 'false', 'off', 'n', 'no')
SLUGIFY_STRIP_RE = re.compile(r"[^\w\s-]")
SLUGIFY_HYPHENATE_RE = re.compile(r"[-\s]+")
def int_from_bool_as_string(subject):
"""Interpret a string as a boolean and return either 1 or 0.
Any string value in:
('True', 'true', 'On', 'on', '1')
is interpreted as a boolean True.
Useful for JSON-decoded stuff and config file parsing
"""
return bool_from_string(subject) and 1 or 0
def bool_from_string(subject, strict=False, default=False):
"""Interpret a string as a boolean.
A case-insensitive match is performed such that strings matching 't',
'true', 'on', 'y', 'yes', or '1' are considered True and, when
`strict=False`, anything else returns the value specified by 'default'.
Useful for JSON-decoded stuff and config file parsing.
If `strict=True`, unrecognized values, including None, will raise a
ValueError which is useful when parsing values passed in from an API call.
Strings yielding False are 'f', 'false', 'off', 'n', 'no', or '0'.
"""
if not isinstance(subject, six.string_types):
subject = six.text_type(subject)
lowered = subject.strip().lower()
if lowered in TRUE_STRINGS:
return True
elif lowered in FALSE_STRINGS:
return False
elif strict:
acceptable = ', '.join(
"'%s'" % s for s in sorted(TRUE_STRINGS + FALSE_STRINGS))
msg = _("Unrecognized value '%(val)s', acceptable values are:"
" %(acceptable)s") % {'val': subject,
'acceptable': acceptable}
raise ValueError(msg)
else:
return default
def safe_decode(text, incoming=None, errors='strict'):
"""Decodes incoming text/bytes string using `incoming`
if they're not already unicode.
:param incoming: Text's current encoding
:param errors: Errors handling policy. See here for valid
values http://docs.python.org/2/library/codecs.html
:returns: text or a unicode `incoming` encoded
representation of it.
:raises TypeError: If text is not an instance of str
"""
if not isinstance(text, (six.string_types, six.binary_type)):
raise TypeError("%s can't be decoded" % type(text))
if isinstance(text, six.text_type):
return text
if not incoming:
incoming = (sys.stdin.encoding or
sys.getdefaultencoding())
try:
return text.decode(incoming, errors)
except UnicodeDecodeError:
# Note(flaper87) If we get here, it means that
# sys.stdin.encoding / sys.getdefaultencoding
# didn't return a suitable encoding to decode
# text. This happens mostly when global LANG
# var is not set correctly and there's no
# default encoding. In this case, most likely
# python will use ASCII or ANSI encoders as
# default encodings but they won't be capable
# of decoding non-ASCII characters.
#
# Also, UTF-8 is being used since it's an ASCII
# extension.
return text.decode('utf-8', errors)
def safe_encode(text, incoming=None,
encoding='utf-8', errors='strict'):
"""Encodes incoming text/bytes string using `encoding`.
If incoming is not specified, text is expected to be encoded with
current python's default encoding. (`sys.getdefaultencoding`)
:param incoming: Text's current encoding
:param encoding: Expected encoding for text (Default UTF-8)
:param errors: Errors handling policy. See here for valid
values http://docs.python.org/2/library/codecs.html
:returns: text or a bytestring `encoding` encoded
representation of it.
:raises TypeError: If text is not an instance of str
"""
if not isinstance(text, (six.string_types, six.binary_type)):
raise TypeError("%s can't be encoded" % type(text))
if not incoming:
incoming = (sys.stdin.encoding or
sys.getdefaultencoding())
if isinstance(text, six.text_type):
return text.encode(encoding, errors)
elif text and encoding != incoming:
# Decode text before encoding it with `encoding`
text = safe_decode(text, incoming, errors)
return text.encode(encoding, errors)
else:
return text
def string_to_bytes(text, unit_system='IEC', return_int=False):
"""Converts a string into an float representation of bytes.
The units supported for IEC ::
Kb(it), Kib(it), Mb(it), Mib(it), Gb(it), Gib(it), Tb(it), Tib(it)
KB, KiB, MB, MiB, GB, GiB, TB, TiB
The units supported for SI ::
kb(it), Mb(it), Gb(it), Tb(it)
kB, MB, GB, TB
Note that the SI unit system does not support capital letter 'K'
:param text: String input for bytes size conversion.
:param unit_system: Unit system for byte size conversion.
:param return_int: If True, returns integer representation of text
in bytes. (default: decimal)
:returns: Numerical representation of text in bytes.
:raises ValueError: If text has an invalid value.
"""
try:
base, reg_ex = UNIT_SYSTEM_INFO[unit_system]
except KeyError:
msg = _('Invalid unit system: "%s"') % unit_system
raise ValueError(msg)
match = reg_ex.match(text)
if match:
magnitude = float(match.group(1))
unit_prefix = match.group(2)
if match.group(3) in ['b', 'bit']:
magnitude /= 8
else:
msg = _('Invalid string format: %s') % text
raise ValueError(msg)
if not unit_prefix:
res = magnitude
else:
res = magnitude * pow(base, UNIT_PREFIX_EXPONENT[unit_prefix])
if return_int:
return int(math.ceil(res))
return res
def to_slug(value, incoming=None, errors="strict"):
"""Normalize string.
Convert to lowercase, remove non-word characters, and convert spaces
to hyphens.
Inspired by Django's `slugify` filter.
:param value: Text to slugify
:param incoming: Text's current encoding
:param errors: Errors handling policy. See here for valid
values http://docs.python.org/2/library/codecs.html
:returns: slugified unicode representation of `value`
:raises TypeError: If text is not an instance of str
"""
value = safe_decode(value, incoming, errors)
# NOTE(aababilov): no need to use safe_(encode|decode) here:
# encodings are always "ascii", error handling is always "ignore"
# and types are always known (first: unicode; second: str)
value = unicodedata.normalize("NFKD", value).encode(
"ascii", "ignore").decode("ascii")
value = SLUGIFY_STRIP_RE.sub("", value).strip().lower()
return SLUGIFY_HYPHENATE_RE.sub("-", value)

44
distributedcloud-client/dcmanagerclient/openstack/common/uuidutils.py
View File

@ -1,44 +0,0 @@
# Copyright (c) 2012 Intel 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.
#
# Copyright (c) 2017 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
# of an applicable Wind River license agreement.
#
"""
UUID related utilities and helper functions.
"""
import uuid
def generate_uuid():
return str(uuid.uuid4())
def is_uuid_like(val):
"""Returns validation of a value as a UUID.
For our purposes, a UUID is a canonical form string:
aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa
"""
try:
return str(uuid.UUID(val)) == val
except (TypeError, ValueError, AttributeError):
return False

84
distributedcloud-client/dcmanagerclient/shell.py
View File

@ -24,12 +24,12 @@ Command-line interface to the DC Manager APIs
"""
import logging
import os
import sys
from dcmanagerclient import __version__ as dcmanager_version
from dcmanagerclient.api import client
from dcmanagerclient import exceptions
from dcmanagerclient.openstack.common import cliutils as c
from cliff import app
from cliff import commandmanager
@ -37,7 +37,8 @@ from osc_lib.command import command
import argparse
from dcmanagerclient.commands.v1 import alarm_manager as am
# from dcmanagerclient.commands.v1 import fw_update_manager as fum
from dcmanagerclient.commands.v1 import fw_update_manager as fum
from dcmanagerclient.commands.v1 import kube_upgrade_manager as kupm
from dcmanagerclient.commands.v1 import subcloud_deploy_manager as sdm
from dcmanagerclient.commands.v1 import subcloud_group_manager as gm
from dcmanagerclient.commands.v1 import subcloud_manager as sm
@ -49,6 +50,18 @@ from dcmanagerclient.commands.v1 import sw_upgrade_manager as supm
LOG = logging.getLogger(__name__)
def env(*args, **kwargs):
"""Returns the first environment variable set.
If all are empty, defaults to '' or keyword arg `default`.
"""
for arg in args:
value = os.environ.get(arg)
if value:
return value
return kwargs.get('default', '')
class OpenStackHelpFormatter(argparse.HelpFormatter):
def __init__(self, prog, indent_increment=2, max_help_position=32,
width=None):
@ -207,7 +220,7 @@ class DCManagerShell(app.App):
'--dcmanager-url',
action='store',
dest='dcmanager_url',
default=c.env('DCMANAGER_URL'),
default=env('DCMANAGER_URL'),
help='DC Manager API host (Env: DCMANAGER_URL)'
)
@ -215,7 +228,7 @@ class DCManagerShell(app.App):
'--dcmanager-api-version',
action='store',
dest='dcmanager_version',
default=c.env('DCMANAGER_API_VERSION', default='v1.0'),
default=env('DCMANAGER_API_VERSION', default='v1.0'),
help='DC Manager API version (default = v1.0) (Env: '
'DCMANAGER_API_VERSION)'
)
@ -224,8 +237,8 @@ class DCManagerShell(app.App):
'--dcmanager-service-type',
action='store',
dest='service_type',
default=c.env('DCMANAGER_SERVICE_TYPE',
default='dcmanager'),
default=env('DCMANAGER_SERVICE_TYPE',
default='dcmanager'),
help='DC Manager service-type (should be the same name as in '
'keystone-endpoint) (default = dcmanager) (Env: '
'DCMANAGER_SERVICE_TYPE)'
@ -235,8 +248,8 @@ class DCManagerShell(app.App):
'--os-endpoint-type',
action='store',
dest='endpoint_type',
default=c.env('OS_ENDPOINT_TYPE',
default='internalURL'),
default=env('OS_ENDPOINT_TYPE',
default='internalURL'),
help='DC Manager endpoint-type (should be the same name as in '
'keystone-endpoint) (default = OS_ENDPOINT_TYPE)'
)
@ -245,7 +258,7 @@ class DCManagerShell(app.App):
'--os-username',
action='store',
dest='username',
default=c.env('OS_USERNAME', default='admin'),
default=env('OS_USERNAME', default='admin'),
help='Authentication username (Env: OS_USERNAME)'
)
@ -253,7 +266,7 @@ class DCManagerShell(app.App):
'--os-password',
action='store',
dest='password',
default=c.env('OS_PASSWORD'),
default=env('OS_PASSWORD'),
help='Authentication password (Env: OS_PASSWORD)'
)
@ -261,7 +274,7 @@ class DCManagerShell(app.App):
'--os-tenant-id',
action='store',
dest='tenant_id',
default=c.env('OS_TENANT_ID', 'OS_PROJECT_ID'),
default=env('OS_TENANT_ID', 'OS_PROJECT_ID'),
help='Authentication tenant identifier (Env: OS_TENANT_ID)'
)
@ -269,7 +282,7 @@ class DCManagerShell(app.App):
'--os-project-id',
action='store',
dest='project_id',
default=c.env('OS_TENANT_ID', 'OS_PROJECT_ID'),
default=env('OS_TENANT_ID', 'OS_PROJECT_ID'),
help='Authentication project identifier (Env: OS_TENANT_ID'
' or OS_PROJECT_ID), will use tenant_id if both tenant_id'
' and project_id are set'
@ -279,7 +292,7 @@ class DCManagerShell(app.App):
'--os-tenant-name',
action='store',
dest='tenant_name',
default=c.env('OS_TENANT_NAME', 'OS_PROJECT_NAME'),
default=env('OS_TENANT_NAME', 'OS_PROJECT_NAME'),
help='Authentication tenant name (Env: OS_TENANT_NAME)'
)
@ -287,7 +300,7 @@ class DCManagerShell(app.App):
'--os-project-name',
action='store',
dest='project_name',
default=c.env('OS_TENANT_NAME', 'OS_PROJECT_NAME'),
default=env('OS_TENANT_NAME', 'OS_PROJECT_NAME'),
help='Authentication project name (Env: OS_TENANT_NAME'
' or OS_PROJECT_NAME), will use tenant_name if both'
' tenant_name and project_name are set'
@ -297,7 +310,7 @@ class DCManagerShell(app.App):
'--os-auth-token',
action='store',
dest='token',
default=c.env('OS_AUTH_TOKEN'),
default=env('OS_AUTH_TOKEN'),
help='Authentication token (Env: OS_AUTH_TOKEN)'
)
@ -305,7 +318,7 @@ class DCManagerShell(app.App):
'--os-project-domain-name',
action='store',
dest='project_domain_name',
default=c.env('OS_PROJECT_DOMAIN_NAME'),
default=env('OS_PROJECT_DOMAIN_NAME'),
help='Authentication project domain name or ID'
' (Env: OS_PROJECT_DOMAIN_NAME)'
)
@ -314,7 +327,7 @@ class DCManagerShell(app.App):
'--os-project-domain-id',
action='store',
dest='project_domain_id',
default=c.env('OS_PROJECT_DOMAIN_ID'),
default=env('OS_PROJECT_DOMAIN_ID'),
help='Authentication project domain ID'
' (Env: OS_PROJECT_DOMAIN_ID)'
)
@ -323,7 +336,7 @@ class DCManagerShell(app.App):
'--os-user-domain-name',
action='store',
dest='user_domain_name',
default=c.env('OS_USER_DOMAIN_NAME'),
default=env('OS_USER_DOMAIN_NAME'),
help='Authentication user domain name'
' (Env: OS_USER_DOMAIN_NAME)'
)
@ -332,7 +345,7 @@ class DCManagerShell(app.App):
'--os-user-domain-id',
action='store',
dest='user_domain_id',
default=c.env('OS_USER_DOMAIN_ID'),
default=env('OS_USER_DOMAIN_ID'),
help='Authentication user domain name'
' (Env: OS_USER_DOMAIN_ID)'
)
@ -341,7 +354,7 @@ class DCManagerShell(app.App):
'--os-auth-url',
action='store',
dest='auth_url',
default=c.env('OS_AUTH_URL'),
default=env('OS_AUTH_URL'),
help='Authentication URL (Env: OS_AUTH_URL)'
)
@ -349,7 +362,7 @@ class DCManagerShell(app.App):
'--os-cacert',
action='store',
dest='cacert',
default=c.env('OS_CACERT'),
default=env('OS_CACERT'),
help='Authentication CA Certificate (Env: OS_CACERT)'
)
@ -357,7 +370,7 @@ class DCManagerShell(app.App):
'--insecure',
action='store_true',
dest='insecure',
default=c.env('DCMANAGERCLIENT_INSECURE', default=False),
default=env('DCMANAGERCLIENT_INSECURE', default=False),
help='Disables SSL/TLS certificate verification '
'(Env: DCMANAGERCLIENT_INSECURE)'
)
@ -394,7 +407,7 @@ class DCManagerShell(app.App):
self.options.auth_url = None
if self.options.auth_url and not self.options.token \
and not skip_auth:
and not skip_auth:
if not self.options.tenant_name:
raise exceptions.CommandError(
("You must provide a tenant_name "
@ -441,7 +454,7 @@ class DCManagerShell(app.App):
"--os-auth-url or env[OS_AUTH_URL] or "
"specify an auth_system which defines a"
" default url with --os-auth-system or env[OS_AUTH_SYSTEM]")
)
)
# Adding client_manager variable to make dcmanager client work with
# unified OpenStack client.
@ -456,7 +469,8 @@ class DCManagerShell(app.App):
sw_patch_manager=self.client,
strategy_step_manager=self.client,
sw_update_options_manager=self.client,
sw_upgrade_manager=self.client)
sw_upgrade_manager=self.client,
kube_upgrade_manager=self.client)
)
self.client_manager = ClientManager()
@ -468,7 +482,7 @@ class DCManagerShell(app.App):
exclude_cmds = ['help', 'complete']
cmds = self.command_manager.commands.copy()
for k, v in cmds.items():
for k, _v in cmds.items():
if k not in exclude_cmds:
self.command_manager.commands.pop(k)
@ -489,6 +503,9 @@ class DCManagerShell(app.App):
'subcloud unmanage': sm.UnmanageSubcloud,
'subcloud manage': sm.ManageSubcloud,
'subcloud update': sm.UpdateSubcloud,
'subcloud reconfig': sm.ReconfigSubcloud,
'subcloud reinstall': sm.ReinstallSubcloud,
'subcloud restore': sm.RestoreSubcloud,
'subcloud-group add': gm.AddSubcloudGroup,
'subcloud-group delete': gm.DeleteSubcloudGroup,
'subcloud-group list': gm.ListSubcloudGroup,
@ -498,11 +515,16 @@ class DCManagerShell(app.App):
'subcloud-deploy upload': sdm.SubcloudDeployUpload,
'subcloud-deploy show': sdm.SubcloudDeployShow,
'alarm summary': am.ListAlarmSummary,
# 'fw-update-strategy create': fum.CreateFwUpdateStrategy,
# 'fw-update-strategy delete': fum.DeleteFwUpdateStrategy,
# 'fw-update-strategy apply': fum.ApplyFwUpdateStrategy,
# 'fw-update-strategy abort': fum.AbortFwUpdateStrategy,
# 'fw-update-strategy show': fum.ShowFwUpdateStrategy,
'fw-update-strategy create': fum.CreateFwUpdateStrategy,
'fw-update-strategy delete': fum.DeleteFwUpdateStrategy,
'fw-update-strategy apply': fum.ApplyFwUpdateStrategy,
'fw-update-strategy abort': fum.AbortFwUpdateStrategy,
'fw-update-strategy show': fum.ShowFwUpdateStrategy,
'kube-upgrade-strategy create': kupm.CreateKubeUpgradeStrategy,
'kube-upgrade-strategy delete': kupm.DeleteKubeUpgradeStrategy,
'kube-upgrade-strategy apply': kupm.ApplyKubeUpgradeStrategy,
'kube-upgrade-strategy abort': kupm.AbortKubeUpgradeStrategy,
'kube-upgrade-strategy show': kupm.ShowKubeUpgradeStrategy,
'patch-strategy create': spm.CreatePatchUpdateStrategy,
'patch-strategy delete': spm.DeletePatchUpdateStrategy,
'patch-strategy apply': spm.ApplyPatchUpdateStrategy,

17
distributedcloud-client/dcmanagerclient/tests/base.py
View File

@ -13,7 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
# Copyright (c) 2017 Wind River Systems, Inc.
# Copyright (c) 2017-2021 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
@ -21,9 +21,8 @@
#
import json
import mock
import unittest2
import testtools
class FakeResponse(object):
@ -39,9 +38,12 @@ class FakeResponse(object):
return json.loads(self.content)
class BaseClientTest(unittest2.TestCase):
class BaseClientTest(testtools.TestCase):
_client = None
def setUp(self):
super(BaseClientTest, self).setUp()
def mock_http_get(self, content, status_code=200):
if isinstance(content, dict):
content = json.dumps(content)
@ -76,12 +78,15 @@ class BaseClientTest(unittest2.TestCase):
return self._client.http_client.delete
class BaseCommandTest(unittest2.TestCase):
class BaseCommandTest(testtools.TestCase):
def setUp(self):
super(BaseCommandTest, self).setUp()
self.app = mock.Mock()
self.client = self.app.client_manager.subcloud_manager
def call(self, command, app_args=[], prog_name=''):
def call(self, command, app_args=None, prog_name=''):
if app_args is None:
app_args = []
cmd = command(self.app, app_args)
parsed_args = cmd.get_parser(prog_name).parse_args(app_args)

4
distributedcloud-client/dcmanagerclient/tests/test_help_and_bash_completion.py
View File

@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
# Copyright (c) 2017 Wind River Systems, Inc.
# Copyright (c) 2017-2021 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
@ -38,7 +38,7 @@ class TestCLIHelp(base.BaseShellTests):
required = [
'.*?^usage: ',
'.*?^\s+help\s+print detailed help for another command'
]
]
kb_help, stderr = self.shell('help')
for r in required:
self.assertThat((kb_help + stderr),

151
distributedcloud-client/dcmanagerclient/tests/v1/mixins.py Normal file
View File

@ -0,0 +1,151 @@
#
# Copyright (c) 2020-2021 Wind River Systems, Inc.
#
# SPDX-License-Identifier: Apache-2.0
#
from dcmanagerclient.tests.v1 import utils
class UpdateStrategyMixin(object):
"""Mixin for testing the different types of dcmanager update strategies.
Used by concrete testsuites of strategy types such as patch, upgrade, etc..
Subclasses must:
- mix with BaseCommandTest
- provide: self.sw_update_manager
- provide: self.create_command
- provide: self.show_command
- provide: self.delete_command
- provide: self.apply_command
- provide: self.abort_command
"""
def setUp(self):
super(UpdateStrategyMixin, self).setUp()
def test_create_strategy(self):
"""Test that if no strategy exists, one can be created."""
# prepare mixin attributes
manager_to_test = self.sw_update_manager
expected_strategy_type = manager_to_test.update_type
# mock the result of the API call
strategy = utils.make_strategy(strategy_type=expected_strategy_type)
# mock that there is no pre-existing strategy
manager_to_test.create_sw_update_strategy.return_value = strategy
# invoke the backend method for the CLI.
# Returns a tuple of field descriptions, and a second tuple of values
fields, results = self.call(self.create_command)
# results is a tuple of expected length 7
self.assertEqual(len(results), 7)
# result tuple values are
# - strategy_type
# - subcloud_apply_type
# - max_parallel_subclouds
# - stop_on_failure
# - state
# - created_at
# - updated_at
self.assertEqual(results[0], expected_strategy_type)
def test_get_strategy(self):
# prepare mocked results
manager_to_test = self.sw_update_manager
expected_strategy_type = manager_to_test.update_type
expected_apply_type = 'parallel'
strategy = utils.make_strategy(strategy_type=expected_strategy_type,
subcloud_apply_type=expected_apply_type)
manager_to_test.update_sw_strategy_detail.return_value = strategy
# invoke the backend method for the CLI.
# Returns a tuple of field descriptions, and a second tuple of values
fields, results = self.call(self.show_command)
# results is a tuple of expected length 7
self.assertEqual(len(results), 7)
# result tuple values are
# - strategy_type
# - subcloud_apply_type
# - max_parallel_subclouds
# - stop_on_failure
# - state
# - created_at
# - updated_at
self.assertEqual(results[0], expected_strategy_type)
self.assertEqual(results[1], expected_apply_type)
def test_apply_strategy(self):
# prepare mocked results
manager_to_test = self.sw_update_manager
expected_strategy_type = manager_to_test.update_type
expected_apply_type = 'parallel'
strategy = utils.make_strategy(strategy_type=expected_strategy_type,
subcloud_apply_type=expected_apply_type)
manager_to_test.apply_sw_update_strategy.return_value = strategy
# invoke the backend method for the CLI.
# Returns a tuple of field descriptions, and a second tuple of values
fields, results = self.call(self.apply_command)
# results is a tuple of expected length 7
self.assertEqual(len(results), 7)
# result tuple values are
# - strategy_type
# - subcloud_apply_type
# - max_parallel_subclouds
# - stop_on_failure
# - state
# - created_at
# - updated_at
self.assertEqual(results[0], expected_strategy_type)
self.assertEqual(results[1], expected_apply_type)
def test_abort_strategy(self):
# prepare mocked results
manager_to_test = self.sw_update_manager
expected_strategy_type = manager_to_test.update_type
expected_apply_type = 'parallel'
strategy = utils.make_strategy(strategy_type=expected_strategy_type,
subcloud_apply_type=expected_apply_type)
manager_to_test.abort_sw_update_strategy.return_value = strategy
# invoke the backend method for the CLI.
# Returns a tuple of field descriptions, and a second tuple of values
fields, results = self.call(self.abort_command)
# results is a tuple of expected length 7
self.assertEqual(len(results), 7)
# result tuple values are
# - strategy_type
# - subcloud_apply_type
# - max_parallel_subclouds
# - stop_on_failure
# - state
# - created_at
# - updated_at
self.assertEqual(results[0], expected_strategy_type)
self.assertEqual(results[1], expected_apply_type)
def test_delete_strategy(self):
# prepare mocked results
manager_to_test = self.sw_update_manager
expected_strategy_type = manager_to_test.update_type
expected_apply_type = 'parallel'
strategy = utils.make_strategy(strategy_type=expected_strategy_type,
subcloud_apply_type=expected_apply_type)
manager_to_test.delete_sw_update_strategy.return_value = strategy
# invoke the backend method for the CLI.
# Returns a tuple of field descriptions, and a second tuple of values
fields, results = self.call(self.delete_command)
# results is a tuple of expected length 7
self.assertEqual(len(results), 7)
# result tuple values are
# - strategy_type
# - subcloud_apply_type
# - max_parallel_subclouds
# - stop_on_failure
# - state
# - created_at
# - updated_at
self.assertEqual(results[0], expected_strategy_type)
self.assertEqual(results[1], expected_apply_type)

22
distributedcloud-client/dcmanagerclient/tests/v1/test_fw_update_strategy.py Normal file
View File

@ -0,0 +1,22 @@
#
# Copyright (c) 2020-2021 Wind River Systems, Inc.
#
# SPDX-License-Identifier: Apache-2.0
#
from dcmanagerclient.commands.v1 import fw_update_manager as cli_cmd
from dcmanagerclient.tests import base
from dcmanagerclient.tests.v1.mixins import UpdateStrategyMixin
class TestFwUpdateStrategy(UpdateStrategyMixin, base.BaseCommandTest):
def setUp(self):
super(TestFwUpdateStrategy, self).setUp()
self.sw_update_manager = \
self.app.client_manager.fw_update_manager.fw_update_manager
self.create_command = cli_cmd.CreateFwUpdateStrategy
self.show_command = cli_cmd.ShowFwUpdateStrategy
self.delete_command = cli_cmd.DeleteFwUpdateStrategy
self.apply_command = cli_cmd.ApplyFwUpdateStrategy
self.abort_command = cli_cmd.AbortFwUpdateStrategy

22
distributedcloud-client/dcmanagerclient/tests/v1/test_kube_upgrade_strategy.py Normal file
View File

@ -0,0 +1,22 @@
#
# Copyright (c) 2020-2021 Wind River Systems, Inc.
#
# SPDX-License-Identifier: Apache-2.0
#
from dcmanagerclient.commands.v1 import kube_upgrade_manager as cli_cmd
from dcmanagerclient.tests import base
from dcmanagerclient.tests.v1.mixins import UpdateStrategyMixin
class TestKubeUpgradeStrategy(UpdateStrategyMixin, base.BaseCommandTest):
def setUp(self):
super(TestKubeUpgradeStrategy, self).setUp()
self.sw_update_manager = \
self.app.client_manager.kube_upgrade_manager.kube_upgrade_manager
self.create_command = cli_cmd.CreateKubeUpgradeStrategy
self.show_command = cli_cmd.ShowKubeUpgradeStrategy
self.delete_command = cli_cmd.DeleteKubeUpgradeStrategy
self.apply_command = cli_cmd.ApplyKubeUpgradeStrategy
self.abort_command = cli_cmd.AbortKubeUpgradeStrategy

160
distributedcloud-client/dcmanagerclient/tests/v1/test_subcloud_manager.py
View File

@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
# Copyright (c) 2017-2020 Wind River Systems, Inc.
# Copyright (c) 2017-2021 Wind River Systems, Inc.
#
# The right to copy, distribute, modify, or otherwise make use
# of this software may be licensed only pursuant to the terms
@ -29,6 +29,7 @@ from oslo_utils import timeutils
from dcmanagerclient.api.v1 import subcloud_manager as sm
from dcmanagerclient.commands.v1 import subcloud_manager as subcloud_cmd
from dcmanagerclient.exceptions import DCManagerClientException
from dcmanagerclient.tests import base
BOOTSTRAP_ADDRESS = '10.10.10.12'
@ -43,6 +44,8 @@ SOFTWARE_VERSION = '12.34'
MANAGEMENT_STATE = 'unmanaged'
AVAILABILITY_STATUS = 'offline'
DEPLOY_STATUS = 'not-deployed'
DEPLOY_STATE_PRE_DEPLOY = 'pre-deploy'
DEPLOY_STATE_PRE_RESTORE = 'pre-restore'
MANAGEMENT_SUBNET = '192.168.101.0/24'
MANAGEMENT_START_IP = '192.168.101.2'
MANAGEMENT_END_IP = '192.168.101.50'
@ -145,9 +148,8 @@ class TestCLISubcloudManagerV1(base.BaseCommandTest):
def test_show_subcloud_with_additional_detail(self):
SUBCLOUD_WITH_ADDITIONAL_DETAIL = copy.copy(SUBCLOUD)
setattr(SUBCLOUD_WITH_ADDITIONAL_DETAIL,
'oam_floating_ip',
SUBCLOUD_DICT['OAM_FLOATING_IP'])
SUBCLOUD_WITH_ADDITIONAL_DETAIL.oam_floating_ip = \
SUBCLOUD_DICT['OAM_FLOATING_IP']
self.client.subcloud_manager.subcloud_additional_details.\
return_value = [SUBCLOUD_WITH_ADDITIONAL_DETAIL]
actual_call = self.call(subcloud_cmd.ShowSubcloud,
@ -197,7 +199,7 @@ class TestCLISubcloudManagerV1(base.BaseCommandTest):
"external_oam_floating_address": EXTERNAL_OAM_FLOATING_ADDRESS,
}
with tempfile.NamedTemporaryFile() as f:
with tempfile.NamedTemporaryFile(mode='w') as f:
yaml.dump(values, f)
file_path = os.path.abspath(f.name)
actual_call = self.call(
@ -213,6 +215,42 @@ class TestCLISubcloudManagerV1(base.BaseCommandTest):
DEFAULT_SUBCLOUD_GROUP_ID,
TIME_NOW, TIME_NOW), actual_call[1])
@mock.patch('getpass.getpass', return_value='testpassword')
def test_add_migrate_subcloud(self, getpass):
self.client.subcloud_manager.add_subcloud.\
return_value = [SUBCLOUD]
values = {
"system_mode": SYSTEM_MODE,
"name": NAME,
"description": DESCRIPTION,
"location": LOCATION,
"management_subnet": MANAGEMENT_SUBNET,
"management_start_address": MANAGEMENT_START_IP,
"management_end_address": MANAGEMENT_END_IP,
"management_gateway_address": MANAGEMENT_GATEWAY_IP,
"external_oam_subnet": EXTERNAL_OAM_SUBNET,
"external_oam_gateway_address": EXTERNAL_OAM_GATEWAY_ADDRESS,
"external_oam_floating_address": EXTERNAL_OAM_FLOATING_ADDRESS,
}
with tempfile.NamedTemporaryFile(mode='w') as f:
yaml.dump(values, f)
file_path = os.path.abspath(f.name)
actual_call = self.call(
subcloud_cmd.AddSubcloud, app_args=[
'--bootstrap-address', BOOTSTRAP_ADDRESS,
'--bootstrap-values', file_path,
'--migrate',
])
self.assertEqual((ID, NAME, DESCRIPTION, LOCATION, SOFTWARE_VERSION,
MANAGEMENT_STATE, AVAILABILITY_STATUS, DEPLOY_STATUS,
MANAGEMENT_SUBNET, MANAGEMENT_START_IP,
MANAGEMENT_END_IP, MANAGEMENT_GATEWAY_IP,
SYSTEMCONTROLLER_GATEWAY_IP,
DEFAULT_SUBCLOUD_GROUP_ID,
TIME_NOW, TIME_NOW), actual_call[1])
def test_unmanage_subcloud(self):
self.client.subcloud_manager.update_subcloud.\
return_value = [SUBCLOUD]
@ -268,3 +306,115 @@ class TestCLISubcloudManagerV1(base.BaseCommandTest):
SYSTEMCONTROLLER_GATEWAY_IP,
DEFAULT_SUBCLOUD_GROUP_ID,
TIME_NOW, TIME_NOW), actual_call[1])
@mock.patch('getpass.getpass', return_value='testpassword')
def test_success_reconfigure_subcloud(self, getpass):
SUBCLOUD_BEING_DEPLOYED = copy.copy(SUBCLOUD)
SUBCLOUD_BEING_DEPLOYED.deploy_status = DEPLOY_STATE_PRE_DEPLOY
self.client.subcloud_manager.reconfigure_subcloud.\
return_value = [SUBCLOUD_BEING_DEPLOYED]
with tempfile.NamedTemporaryFile() as f:
file_path = os.path.abspath(f.name)
actual_call = self.call(
subcloud_cmd.ReconfigSubcloud,
app_args=[ID,
'--deploy-config', file_path])
self.assertEqual((ID, NAME,
DESCRIPTION, LOCATION,
SOFTWARE_VERSION, MANAGEMENT_STATE,
AVAILABILITY_STATUS, DEPLOY_STATE_PRE_DEPLOY,
MANAGEMENT_SUBNET, MANAGEMENT_START_IP,
MANAGEMENT_END_IP, MANAGEMENT_GATEWAY_IP,
SYSTEMCONTROLLER_GATEWAY_IP,
DEFAULT_SUBCLOUD_GROUP_ID,
TIME_NOW, TIME_NOW), actual_call[1])
@mock.patch('getpass.getpass', return_value='testpassword')
def test_reconfigure_file_does_not_exist(self, getpass):
SUBCLOUD_BEING_DEPLOYED = copy.copy(SUBCLOUD)
SUBCLOUD_BEING_DEPLOYED.deploy_status = DEPLOY_STATE_PRE_DEPLOY
self.client.subcloud_manager.reconfigure_subcloud.\
return_value = [SUBCLOUD_BEING_DEPLOYED]
with tempfile.NamedTemporaryFile() as f:
file_path = os.path.abspath(f.name)
e = self.assertRaises(DCManagerClientException,
self.call,
subcloud_cmd.ReconfigSubcloud,
app_args=[ID, '--deploy-config', file_path])
self.assertTrue('deploy-config file does not exist'
in str(e))
@mock.patch('six.moves.input', return_value='reinstall')
def test_reinstall_subcloud(self, mock_input):
self.client.subcloud_manager.reinstall_subcloud.\
return_value = [SUBCLOUD]
actual_call = self.call(
subcloud_cmd.ReinstallSubcloud, app_args=[ID])
self.assertEqual((ID, NAME,
DESCRIPTION, LOCATION,
SOFTWARE_VERSION, MANAGEMENT_STATE,
AVAILABILITY_STATUS, DEPLOY_STATUS,
MANAGEMENT_SUBNET, MANAGEMENT_START_IP,
MANAGEMENT_END_IP, MANAGEMENT_GATEWAY_IP,
SYSTEMCONTROLLER_GATEWAY_IP,
DEFAULT_SUBCLOUD_GROUP_ID,
TIME_NOW, TIME_NOW), actual_call[1])
@mock.patch('getpass.getpass', return_value='testpassword')
def test_restore_subcloud(self, getpass):
self.client.subcloud_manager.subcloud_detail.\
return_value = [SUBCLOUD]
SUBCLOUD_BEING_RESTORED = copy.copy(SUBCLOUD)
setattr(SUBCLOUD_BEING_RESTORED,
'deploy_status',
DEPLOY_STATE_PRE_RESTORE)
self.client.subcloud_manager.restore_subcloud.\
return_value = [SUBCLOUD_BEING_RESTORED]
with tempfile.NamedTemporaryFile() as f:
file_path = os.path.abspath(f.name)
actual_call = self.call(
subcloud_cmd.RestoreSubcloud,
app_args=[ID,
'--restore-values', file_path])
self.assertEqual((ID, NAME,
DESCRIPTION, LOCATION,
SOFTWARE_VERSION, MANAGEMENT_STATE,
AVAILABILITY_STATUS, DEPLOY_STATE_PRE_RESTORE,
MANAGEMENT_SUBNET, MANAGEMENT_START_IP,
MANAGEMENT_END_IP, MANAGEMENT_GATEWAY_IP,
SYSTEMCONTROLLER_GATEWAY_IP,
DEFAULT_SUBCLOUD_GROUP_ID,
TIME_NOW, TIME_NOW), actual_call[1])
@mock.patch('getpass.getpass', return_value='testpassword')
def test_restore_file_does_not_exist(self, getpass):
with tempfile.NamedTemporaryFile() as f:
file_path = os.path.abspath(f.name)
e = self.assertRaises(DCManagerClientException,
self.call,
subcloud_cmd.RestoreSubcloud,
app_args=[ID, '--restore-values', file_path])
self.assertTrue('restore-values does not exist'
in str(e))
@mock.patch('getpass.getpass', return_value='testpassword')
def test_restore_subcloud_does_not_exist(self, getpass):
self.client.subcloud_manager.subcloud_detail.\
return_value = []
with tempfile.NamedTemporaryFile() as f:
file_path = os.path.abspath(f.name)
e = self.assertRaises(DCManagerClientException,
self.call,
subcloud_cmd.RestoreSubcloud,
app_args=[ID, '--restore-values', file_path])
self.assertTrue('does not exist'
in str(e))

22
distributedcloud-client/dcmanagerclient/tests/v1/test_sw_upgrade_strategy.py Normal file
View File

@ -0,0 +1,22 @@
#
# Copyright (c) 2020-2021 Wind River Systems, Inc.
#
# SPDX-License-Identifier: Apache-2.0
#
from dcmanagerclient.commands.v1 import sw_upgrade_manager as cli_cmd
from dcmanagerclient.tests import base
from dcmanagerclient.tests.v1.mixins import UpdateStrategyMixin
class TestSwUpgradeStrategy(UpdateStrategyMixin, base.BaseCommandTest):
def setUp(self):
super(TestSwUpgradeStrategy, self).setUp()
self.sw_update_manager = \
self.app.client_manager.sw_upgrade_manager.sw_upgrade_manager
self.create_command = cli_cmd.CreateSwUpgradeStrategy
self.show_command = cli_cmd.ShowSwUpgradeStrategy
self.delete_command = cli_cmd.DeleteSwUpgradeStrategy
self.apply_command = cli_cmd.ApplySwUpgradeStrategy
self.abort_command = cli_cmd.AbortSwUpgradeStrategy

37
distributedcloud-client/dcmanagerclient/tests/v1/utils.py Normal file
View File

@ -0,0 +1,37 @@
#
# Copyright (c) 2020-2021 Wind River Systems, Inc.
#
# SPDX-License-Identifier: Apache-2.0
#
import mock
from oslo_utils import timeutils
from dcmanagerclient.api.v1.sw_update_manager import SwUpdateStrategy
TIME_NOW = timeutils.utcnow().isoformat()
DEFAULT_APPLY_TYPE = 'serial'
DEFAULT_MAX_PARALLEL = 2
DEFAULT_STATE = 'initial'
DEFAULT_STRATEGY_TYPE = 'patch'
def make_strategy(manager=None,
strategy_type=DEFAULT_STRATEGY_TYPE,
subcloud_apply_type=DEFAULT_APPLY_TYPE,
max_parallel_subclouds=DEFAULT_MAX_PARALLEL,
stop_on_failure=False,
state=DEFAULT_STATE,
created_at=TIME_NOW,
updated_at=None):
if manager is None:
manager = mock.MagicMock()
return SwUpdateStrategy(manager,
strategy_type,
subcloud_apply_type,
max_parallel_subclouds,
stop_on_failure,
state,
created_at,
updated_at)

9
distributedcloud-client/pylint.rc
View File

@ -37,22 +37,23 @@ load-plugins=
# R detect Refactor for a "good practice" metric violation
# C detect Convention for coding standard violation
# W0102: dangerous-default-value
# W0107: unnecessary-pass
# W0201: attribute-defined-outside-init
# W0212: protected-access
# W0231: super-init-not-called
# W0403: relative-import (typically caused by six)
# W0603: global-statement
# W0612: unused-variable
# W0613: unused-argument
# W0621: redefined-outer-name
# W0622: redefined-builtin
# W0703: broad-except
# W1201: logging-not-lazy
# W1113: keyword-arg-before-vararg
# W1201: logging-not-lazy
# W1505: deprecated-method
# E1102: not-callable
disable=fixme,C,R,
W0102,W0201,W0212,W0231,W0403,
W0612,W0613,W0603,W0621,W0622,W0703,W1201,W1113,
W0102,W0107,W0201,W0212,W0231,W0403,
W0612,W0613,W0603,W0621,W0622,W0703,W1112,W1201,W1505,
E1102
[REPORTS]

6
distributedcloud-client/test-requirements.txt
View File

@ -1,14 +1,12 @@
# The order of packages is significant, because pip processes them in the order
# of appearance. Changing the order has an impact on the overall integration
# process, which may cause wedges in the gate later.
hacking!=0.13.0,<0.14,>=0.12.0 # Apache-2.0
astroid <= 2.2.5
hacking>=1.1.0,<=2.0.0 # Apache-2.0
isort<5;python_version>="3.0"
pylint<2.1.0;python_version<"3.0" # GPLv2
pylint<2.4.0;python_version>="3.0" # GPLv2
pylint<2.3.0;python_version>="3.0" # GPLv2
python-openstackclient>=3.3.0 # Apache-2.0
sphinx>=1.5.1 # BSD
unittest2 # BSD
fixtures>=3.0.0 # Apache-2.0/BSD
mock>=2.0 # BSD
nose # LGPL

19
distributedcloud-client/tox.ini
View File

@ -1,6 +1,6 @@
[tox]
minversion = 2.3
envlist = py27,pep8,pylint
envlist = py27,py36,pep8,pylint
skipsdist = True
toxworkdir = /tmp/{env:USER}_dc_client_tox
@ -32,12 +32,19 @@ commands =
find {toxinidir} -not -path '{toxinidir}/.tox/*' -name '*.py[c|o]' -delete
stestr --test-path={[dcclient]client_base_dir}/dcmanagerclient/tests run '{posargs}'
[testenv:py36]
basepython = python3.6
commands =
find {toxinidir} -not -path '{toxinidir}/.tox/*' -name '*.py[c|o]' -delete
stestr --test-path={[dcclient]client_base_dir}/dcmanagerclient/tests run '{posargs}'
[testenv:pep8]
basepython = python3
deps = {[testenv]deps}
commands = flake8 {posargs}
[testenv:pylint]
basepython = python2.7
basepython = python3
sitepackages = False
deps = {[testenv]deps}
commands =
@ -61,6 +68,8 @@ commands =
coverage erase
stestr --test-path={[dcclient]client_base_dir}/dcmanagerclient/tests run '{posargs}'
coverage combine
coverage html -d cover
coverage xml -o cover/coverage.xml
coverage report
[testenv:debug]
@ -69,10 +78,12 @@ commands = oslo_debug_helper {posargs}
[flake8]
# E123, E125 skipped as they are invalid PEP-8.
# W504 line break after binary operator
# W605 invalid escape sequence
show-source = True
ignore = E123,E125,H102
ignore = E123,E125,W504,W605,H102
builtins = _
exclude=.venv,.git,.tox,dist,doc,*lib/python*,*openstack/common*,*egg,build
exclude=.venv,.git,.tox,dist,doc,*lib/python*,*egg,build
[testenv:linters]
basepython = python3

6
doc/requirements.txt
View File

@ -1,5 +1,5 @@
sphinx>=1.6.2
openstackdocstheme>=1.26.0 # Apache-2.0
sphinx>=2.0.0,!=2.1.0 # BSD
openstackdocstheme>=2.2.1 # Apache-2.0
# Release Notes documentation
reno>=0.1.1 # Apache2
reno>=3.1.0 # Apache-2.0

14
doc/source/conf.py
View File

@ -27,11 +27,6 @@ project = u'StarlingX Distributed Cloud Client'
copyright = u'2018, StarlingX'
author = u'StarlingX'
# The short X.Y version
version = u''
# The full version, including alpha/beta/rc tags
release = u'0.1'
# -- General configuration ---------------------------------------------------
@ -58,8 +53,11 @@ source_suffix = '.rst'
# The master toctree document.
master_doc = 'index'
bug_project = '1027'
bug_tag = 'stx.bug'
# openstackdocstheme options
openstackdocs_repo_name = 'starlingx/distcloud-client'
openstackdocs_use_storyboard = True
openstackdocs_auto_name = False
# The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages.
@ -74,7 +72,7 @@ language = None
exclude_patterns = []
# The name of the Pygments (syntax highlighting) style to use.
pygments_style = 'sphinx'
pygments_style = 'native'
# -- Options for HTML output -------------------------------------------------

12
releasenotes/source/conf.py
View File

@ -36,8 +36,10 @@ extensions = [
project = u'StarlingX Distributed Cloud Client'
bug_project = '1027'
bug_tag = 'stx.bug'
# openstackdocstheme options
openstackdocs_repo_name = 'starlingx/distcloud-client'
openstackdocs_use_storyboard = True
openstackdocs_auto_name = False
# Add any paths that contain templates here, relative to this directory.
# templates_path = ['_templates']
@ -85,7 +87,7 @@ exclude_patterns = []
# show_authors = False
# The name of the Pygments (syntax highlighting) style to use.
pygments_style = 'sphinx'
pygments_style = 'native'
# A list of ignored prefixes for module index sorting.
# modindex_common_prefix = []
@ -134,10 +136,6 @@ html_theme = 'starlingxdocs'
# directly to the root of the documentation.
# html_extra_path = []
# If not '', a 'Last updated on:' timestamp is inserted at every page bottom,
# using the given strftime format.
html_last_updated_fmt = '%Y-%m-%d %H:%M'
# If true, SmartyPants will be used to convert quotes and dashes to
# typographically correct entities.
# html_use_smartypants = True

2
test-requirements.txt
View File

@ -1,3 +1,5 @@
PyYAML>=3.1.0
yamllint>=0.5.2
mock>=2.0 # BSD
isort<5;python_version>="3.5"
bandit;python_version>="3.5"

5
tox.ini
View File

@ -83,3 +83,8 @@ commands =
sphinx-build -W -b html -d api-ref/build/doctrees api-ref/source api-ref/build/html
whitelist_externals = rm
[testenv:bandit]
basepython = python3
description = Bandit code scan for *.py files under config folder
deps = -r{toxinidir}/test-requirements.txt
commands = bandit -r {toxinidir}/ -x '**/.tox/**,**/.eggs/**' -lll