-.coverage

-host=review.openstack.org

-test_path=./watcherclient/tests/functional

- - openstack-python35-jobs

- - openstack-python36-jobs

- - watcherclient-tempest-functional:

- voting: false

-* Source: https://git.openstack.org/cgit/openstack/python-watcher

-10

- debhelper (>= 10),

-Standards-Version: 4.3.0

- Defaults to ``env[OS_WATCHER_API_VERSION]``.

- export OS_AUTH_URL=http://auth.example.com:5000/v2.0

- $ export OS_AUTH_URL=http://auth.example.com:5000/v2.0

- $ export OS_AUTH_URL=http://auth.example.com:5000/v2.0

-https://git.openstack.org/cgit/openstack/python-watcherclient/tree/tools/watcher.bash_completion)

- http://lists.openstack.org/cgi-bin/mailman/listinfo/openstack-dev

- https://git.openstack.org/cgit/openstack/python-watcherclient

- https://review.openstack.org/#/q/status:open+project:openstack/python-watcherclient,n,z

- $ git clone https://git.openstack.org/openstack/python-watcherclient

-prettytable==0.7.2

-PrettyTable<0.8,>=0.7.2 # BSD

-author-email = openstack-dev@lists.openstack.org

- Programming Language :: Python :: 3.5

-tempest.test_plugins =

- watcherclient_tests = watcherclient.plugin:WatcherClientTempestPlugin

-hacking!=0.13.0,<0.14,>=0.12.0 # Apache-2.0

-openstackdocstheme>=1.18.1 # Apache-2.0

-sphinx!=1.6.6,!=1.6.7,>=1.6.5 # BSD

-minversion = 1.8

-envlist = py35,py27,pep8

-install_command =

- constraints: pip install -U --force-reinstall -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt} {opts} {packages}

- pip install -U {opts} {packages}

-deps = -r{toxinidir}/test-requirements.txt

- stestr --test-path=./watcherclient/tests/unit run {posargs}

-[testenv:functional]

-basepython = python2.7

-passenv =

- OS_PROJECT_DOMAIN_NAME

- OS_USER_DOMAIN_NAME

- OS_PROJECT_NAME

- OS_USERNAME

- OS_PASSWORD

- OS_AUTH_URL

- OS_IDENTITY_API_VERSION

- OS_IMAGE_API_VERSION

-commands = stestr --test-path=./watcherclient/tests/functional run --concurrency=1 {posargs}

- stestr slowest

-

- os_watcher_api_version=None, max_retries=None,

- :param os_watcher_api_version: watcher API version to use

- 'os_watcher_api_version': os_watcher_api_version,

-SERVICE_TYPE = "infra-optim"

- try:

- return obj.id

- except AttributeError:

- return obj

-# 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.

-

-# 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

-

-from oslo_utils import encodeutils

-from oslo_utils import strutils

-import prettytable

-import six

-from six import moves

-

-from watcherclient._i18n import _

-

-

-class MissingArgs(Exception):

- """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)

-

-

-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 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, field_labels=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')

- :param field_labels: Labels to use in the heading of the table, default to

- fields.

- """

- formatters = formatters or {}

- mixed_case_fields = mixed_case_fields or []

- field_labels = field_labels or fields

- if len(field_labels) != len(fields):

- raise ValueError(_("Field labels list %(labels)s has different number "

- "of elements than fields list %(fields)s"),

- {'labels': field_labels, 'fields': fields})

-

- if sortby_index is None:

- kwargs = {}

- else:

- kwargs = {'sortby': field_labels[sortby_index]}

- pt = prettytable.PrettyTable(field_labels)

- pt.align = 'l'

-

- for o in objs:

- row = []

- for field in fields:

- data = '-'

- if field in formatters:

- data = 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, '')

- if data is None:

- data = '-'

- row.append(data)

- pt.add_row(row)

-

- if six.PY3:

- print(encodeutils.safe_encode(pt.get_string(**kwargs)).decode())

- else:

- print(encodeutils.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'])

- pt.align = 'l'

- for k, v in dct.items():

- # 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)

- elif wrap < 0:

- raise ValueError(_("Wrap argument should be a positive integer"))

- # 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:

- if v is None:

- v = '-'

- pt.add_row([k, v])

-

- if six.PY3:

- print(encodeutils.safe_encode(pt.get_string()).decode())

- else:

- print(encodeutils.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 service_type(stype):

- """Adds 'service_type' attribute to decorated function.

-

- Usage:

-

- .. code-block:: python

-

- @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)

-# NOTE(deva): Record the latest version that this client was tested with.

-# We still have a lot of work to do in the client to implement

-# microversion support in the client properly! See

-# http://specs.openstack.org/openstack/watcher-specs/specs/kilo/api-microversions.html # noqa

-# for full details.

-DEFAULT_VER = '1.0'

-

- # NOTE: servers before commit 32fb6e99 did not return version headers

- # on error, so we need to perform a GET to determine

- # the supported version range

- if not max_ver:

- LOG.debug('No version header in response, requesting from server')

- if self.os_watcher_api_version:

- base_version = ("/v%s" %

- str(self.os_watcher_api_version).split('.')[0])

- else:

- base_version = API_VERSION

- resp = self._make_simple_request(conn, 'GET', base_version)

- min_ver, max_ver = self._parse_version_headers(resp)

- # version and still failing then error now. The server could

- % {'req': self.os_watcher_api_version,

- % {'req': self.os_watcher_api_version,

- min(version.StrictVersion(self.os_watcher_api_version),

- self.os_watcher_api_version = negotiated_ver

- min_ver = accessor_func('X-OpenStack-Watcher-API-Minimum-Version',

- max_ver = accessor_func('X-OpenStack-Watcher-API-Maximum-Version',

- self.os_watcher_api_version = kwargs.get('os_watcher_api_version',

- DEFAULT_VER)

- return urlparse.urljoin(self.endpoint_trimmed, url)

- if self.os_watcher_api_version:

- kwargs['headers'].setdefault('X-OpenStack-Watcher-API-Version',

- self.os_watcher_api_version)

- kwargs['headers']['X-OpenStack-Watcher-API-Version'] = (

- negotiated_ver)

- http_client.RESET_CONTENT)

- or content_type is None):

- os_watcher_api_version,

- self.os_watcher_api_version = os_watcher_api_version

- if getattr(self, 'os_watcher_api_version', None):

- kwargs['headers'].setdefault('X-OpenStack-Watcher-API-Version',

- self.os_watcher_api_version)

- kwargs['headers']['X-OpenStack-Watcher-API-Version'] = (

- negotiated_ver)

- os_watcher_api_version=DEFAULT_VER,

- return SessionClient(session=session,

- os_watcher_api_version=os_watcher_api_version,

- api_version_select_state=api_version_select_state,

- max_retries=max_retries,

- retry_interval=retry_interval,

- endpoint=endpoint,

- **kwargs)

- return HTTPClient(endpoint=endpoint,

- token=token,

- auth_ref=auth_ref,

- os_watcher_api_version=os_watcher_api_version,

- api_version_select_state=api_version_select_state,

- max_retries=max_retries,

- retry_interval=retry_interval,

- timeout=timeout,

- ca_file=ca_file,

- cert_file=cert_file,

- key_file=key_file,

- insecure=insecure)

-# -*- encoding: utf-8 -*-

-# Copyright (c) 2018 Servionica

-#

-# Licensed under the Apache License, Version 2.0 (the "License");

-# you may not use this file except in compliance with the License.

-# You may obtain a copy of the License at

-#

-# http://www.apache.org/licenses/LICENSE-2.0

-#

-# Unless required by applicable law or agreed to in writing, software

-# distributed under the License is distributed on an "AS IS" BASIS,

-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or

-# implied.

-# See the License for the specific language governing permissions and

-# limitations under the License.

-

-from oslo_config import cfg

-

-

-service_option = cfg.BoolOpt("watcher_client",

- default=True,

- help="Whether or not watcher_client is expected"

- "to be available")

-DEFAULT_API_VERSION = '1'

- instance._api_version[API_NAME],

- os_watcher_api_version=instance._api_version[API_NAME],

-# Licensed under the Apache License, Version 2.0 (the "License"); you may

-# not use this file except in compliance with the License. You may obtain

-# a copy of the License at

-#

-# http://www.apache.org/licenses/LICENSE-2.0

-#

-# Unless required by applicable law or agreed to in writing, software

-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT

-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the

-# License for the specific language governing permissions and limitations

-# under the License.

-

-

-import os

-

-from tempest.test_discover import plugins

-

-from watcherclient import config as watcher_config

-

-

-class WatcherClientTempestPlugin(plugins.TempestPlugin):

- def load_tests(self):

- base_path = os.path.split(os.path.dirname(

- os.path.abspath(__file__)))[0]

- test_dir = "watcherclient/tests/functional/v1"

- full_test_dir = os.path.join(base_path, test_dir)

- return full_test_dir, base_path

-

- def register_opts(self, conf):

- conf.register_opt(watcher_config.service_option,

- group='service_available')

-

- def get_opt_lists(self):

- return [('service_available', [watcher_config.service_option])]

- parser.add_argument('--os-watcher-api-version',

- metavar='<os-watcher-api-version>',

- default=utils.env('OS_WATCHER_API_VERSION',

- default='1'),

- help='Defaults to env[OS_WATCHER_API_VERSION].')

-..

- Except where otherwise noted, this document is licensed under Creative

- Commons Attribution 3.0 License. You can view the license at:

-

- https://creativecommons.org/licenses/by/3.0/

-

-=======

-Testing

-=======

-

-.. _functional_tests:

-

-Functional tests

-================

-

-The following procedure gets you started with Tempest testing but you can also

-refer to the `Tempest documentation`_ for more details.

-

-.. _Tempest documentation: https://docs.openstack.org/tempest/latest/

-

-

-Tempest installation

---------------------

-

-You need to install virtualenv, create a virtual environment and activate it::

-

- $ pip install virtualenv

- $ virtualenv watcher-env

- $ . watcher-env/bin/activate

-

-Then, to install Tempest you can issue the following commands::

-

- $ git clone https://github.com/openstack/tempest/

- $ pip install tempest/

-

-There should be set environment variables using the OpenStack RC file. If

-you don't have RC file yet, create ``admin-openrc`` file and fill it using

-the following example::

-

- export OS_PROJECT_DOMAIN_NAME=default

- export OS_USER_DOMAIN_NAME=default

- export OS_PROJECT_NAME=admin

- export OS_USERNAME=admin

- export OS_PASSWORD=admin

- export OS_AUTH_URL=http://controller:35357/v3

- export OS_IDENTITY_API_VERSION=3

- export OS_IMAGE_API_VERSION=2

-

-Then, save file and execute ``source admin-openrc`` to set environment

-variables.

-

-To run functional tests you need to go to python-watcherclient folder, install

-all requirements and execute ``tempest run`` command::

-

- $ pip install -r requirements.txt -r test-requirements.txt

- $ pip install .

- $ tempest run --regex watcherclient.tests.functional

-

-You can run specified test(s) by using regular expression::

-

- $ tempest run --regex watcherclient.tests.functional.v1.test_action.ActionTests.test_action_list

-# Licensed under the Apache License, Version 2.0 (the "License"); you may

-# not use this file except in compliance with the License. You may obtain

-# a copy of the License at

-#

-# http://www.apache.org/licenses/LICENSE-2.0

-#

-# Unless required by applicable law or agreed to in writing, software

-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT

-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the

-# License for the specific language governing permissions and limitations

-# under the License.

-

-import os

-

-import re

-import shlex

-import subprocess

-import testtools

-

-import six

-from tempest import clients

-from tempest.common import credentials_factory as creds_factory

-from tempest.lib.cli import output_parser

-from tempest.lib import exceptions

-

-

-def credentials():

- # You can get credentials either from tempest.conf file or

- # from OS environment.

- tempest_creds = clients.get_auth_provider(

- creds_factory.get_configured_admin_credentials())

- creds = tempest_creds.credentials

- creds_dict = {

- '--os-username': os.environ.get('OS_USERNAME', creds.username),

- '--os-password': os.environ.get('OS_PASSWORD', creds.password),

- '--os-project-name': os.environ.get('OS_PROJECT_NAME',

- creds.project_name),

- '--os-auth-url': os.environ.get('OS_AUTH_URL',

- tempest_creds.auth_url),

- '--os-project-domain-name': os.environ.get('OS_PROJECT_DOMAIN_ID',

- creds.project_domain_name),

- '--os-user-domain-name': os.environ.get('OS_USER_DOMAIN_ID',

- creds.user_domain_name),

- }

- return [x for sub in creds_dict.items() for x in sub]

-

-

-def execute(cmd, fail_ok=False, merge_stderr=False):

- """Executes specified command for the given action."""

- cmdlist = shlex.split(cmd)

- cmdlist.extend(credentials())

- stdout = subprocess.PIPE

- stderr = subprocess.STDOUT if merge_stderr else subprocess.PIPE

- proc = subprocess.Popen(cmdlist, stdout=stdout, stderr=stderr)

- result, result_err = proc.communicate()

- result = result.decode('utf-8')

- if not fail_ok and proc.returncode != 0:

- raise exceptions.CommandFailed(proc.returncode, cmd, result,

- result_err)

- return result

-

-

-class TestCase(testtools.TestCase):

-

- delimiter_line = re.compile('^\+\-[\+\-]+\-\+$')

-

- @classmethod

- def watcher(cls, cmd, fail_ok=False):

- """Executes watcherclient command for the given action."""

- return execute('watcher {0}'.format(cmd), fail_ok=fail_ok)

-

- @classmethod

- def get_opts(cls, fields, format='value'):

- return ' -f {0} {1}'.format(format,

- ' '.join(['-c ' + it for it in fields]))

-

- @classmethod

- def assertOutput(cls, expected, actual):

- if expected != actual:

- raise Exception('{0} != {1}'.format(expected, actual))

-

- @classmethod

- def assertInOutput(cls, expected, actual):

- if expected not in actual:

- raise Exception('{0} not in {1}'.format(expected, actual))

-

- def assert_table_structure(self, items, field_names):

- """Verify that all items have keys listed in field_names."""

- for item in items:

- for field in field_names:

- self.assertIn(field, item)

-

- def assert_show_fields(self, items, field_names):

- """Verify that all items have keys listed in field_names."""

- for item in items:

- for key in six.iterkeys(item):

- self.assertIn(key, field_names)

-

- def assert_show_structure(self, items, field_names):

- """Verify that all field_names listed in keys of all items."""

- if isinstance(items, list):

- o = {}

- for d in items:

- o.update(d)

- else:

- o = items

- item_keys = o.keys()

- for field in field_names:

- self.assertIn(field, item_keys)

-

- @staticmethod

- def parse_show_as_object(raw_output):

- """Return a dict with values parsed from cli output."""

- items = TestCase.parse_show(raw_output)

- o = {}

- for item in items:

- o.update(item)

- return o

-

- @staticmethod

- def parse_show(raw_output):

- """Return list of dicts with item values parsed from cli output."""

-

- items = []

- table_ = output_parser.table(raw_output)

- for row in table_['values']:

- item = {}

- item[row[0]] = row[1]

- items.append(item)

- return items

-

- def parse_listing(self, raw_output):

- """Return list of dicts with basic item parsed from cli output."""

- return output_parser.listing(raw_output)

-

- def has_actionplan_succeeded(self, ap_uuid):

- return self.parse_show_as_object(

- self.watcher('actionplan show %s' % ap_uuid)

- )['State'] == 'SUCCEEDED'

-

- @classmethod

- def has_audit_created(cls, audit_uuid):

- return cls.parse_show_as_object(

- cls.watcher('audit show %s' % audit_uuid))['State'] == 'SUCCEEDED'

-# Copyright (c) 2016 Servionica

-#

-# Licensed under the Apache License, Version 2.0 (the "License");

-# you may not use this file except in compliance with the License.

-# You may obtain a copy of the License at

-#

-# http://www.apache.org/licenses/LICENSE-2.0

-#

-# Unless required by applicable law or agreed to in writing, software

-# distributed under the License is distributed on an "AS IS" BASIS,

-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or

-# implied.

-# See the License for the specific language governing permissions and

-# limitations under the License.

-

-from oslo_utils import uuidutils

-

-import functools

-

-from tempest.lib.common.utils import test_utils

-

-from watcherclient.tests.functional.v1 import base

-

-

-class ActionTests(base.TestCase):

- """Functional tests for action."""

-

- dummy_name = 'dummy'

- list_fields = ['UUID', 'Parents', 'State', 'Action Plan', 'Action']

- detailed_list_fields = list_fields + ['Created At', 'Updated At',

- 'Deleted At', 'Parameters']

- audit_template_name = 'a' + uuidutils.generate_uuid()

- audit_uuid = None

-

- @classmethod

- def setUpClass(cls):

- template_raw_output = cls.watcher(

- 'audittemplate create %s dummy -s dummy' % cls.audit_template_name)

- template_output = cls.parse_show_as_object(template_raw_output)

- audit_output = cls.parse_show_as_object(cls.watcher(

- 'audit create -a %s' % template_output['Name']))

- cls.audit_uuid = audit_output['UUID']

- audit_created = test_utils.call_until_true(

- func=functools.partial(cls.has_audit_created, cls.audit_uuid),

- duration=600,

- sleep_for=2)

- if not audit_created:

- raise Exception('Audit has not been succeeded')

-

- @classmethod

- def tearDownClass(cls):

- # Delete Action Plan and all related actions.

- output = cls.parse_show(

- cls.watcher('actionplan list --audit %s' % cls.audit_uuid))

- action_plan_uuid = list(output[0])[0]

- raw_output = cls.watcher('actionplan delete %s' % action_plan_uuid)

- cls.assertOutput('', raw_output)

- # Delete audit

- raw_output = cls.watcher('audit delete %s' % cls.audit_uuid)

- cls.assertOutput('', raw_output)

- # Delete Template

- raw_output = cls.watcher(

- 'audittemplate delete %s' % cls.audit_template_name)

- cls.assertOutput('', raw_output)

-

- def test_action_list(self):

- raw_output = self.watcher('action list')

- self.assert_table_structure([raw_output], self.list_fields)

-

- def test_action_detailed_list(self):

- raw_output = self.watcher('action list --detail')

- self.assert_table_structure([raw_output], self.detailed_list_fields)

-

- def test_action_show(self):

- action_list = self.parse_show(self.watcher('action list --audit %s'

- % self.audit_uuid))

- action_uuid = list(action_list[0])[0]

- action = self.watcher('action show %s' % action_uuid)

- self.assertIn(action_uuid, action)

- self.assert_table_structure([action],

- self.detailed_list_fields)

-# Copyright (c) 2016 Servionica

-#

-# Licensed under the Apache License, Version 2.0 (the "License");

-# you may not use this file except in compliance with the License.

-# You may obtain a copy of the License at

-#

-# http://www.apache.org/licenses/LICENSE-2.0

-#

-# Unless required by applicable law or agreed to in writing, software

-# distributed under the License is distributed on an "AS IS" BASIS,

-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or

-# implied.

-# See the License for the specific language governing permissions and

-# limitations under the License.

-

-from oslo_utils import uuidutils

-

-import functools

-

-from tempest.lib.common.utils import test_utils

-

-from watcherclient.tests.functional.v1 import base

-

-

-class ActionPlanTests(base.TestCase):

- """Functional tests for action plan."""

-

- dummy_name = 'dummy'

- list_fields = ['UUID', 'Audit', 'State', 'Updated At', 'Global efficacy']

- detailed_list_fields = list_fields + ['Created At', 'Deleted At',

- 'Strategy', 'Efficacy indicators',

- 'Hostname']

- audit_template_name = 'a' + uuidutils.generate_uuid()

- audit_uuid = None

-

- @classmethod

- def setUpClass(cls):

- template_raw_output = cls.watcher(

- 'audittemplate create %s dummy -s dummy' % cls.audit_template_name)

- template_output = cls.parse_show_as_object(template_raw_output)

- audit_raw_output = cls.watcher('audit create -a %s'

- % template_output['Name'])

- audit_output = cls.parse_show_as_object(audit_raw_output)

- cls.audit_uuid = audit_output['UUID']

- audit_created = test_utils.call_until_true(

- func=functools.partial(cls.has_audit_created, cls.audit_uuid),

- duration=600,

- sleep_for=2)

- if not audit_created:

- raise Exception('Audit has not been succeeded')

-

- @classmethod

- def tearDownClass(cls):

- # Delete action plan

- output = cls.parse_show(

- cls.watcher('actionplan list --audit %s' % cls.audit_uuid))

- action_plan_uuid = list(output[0])[0]

- raw_output = cls.watcher('actionplan delete %s' % action_plan_uuid)

- cls.assertOutput('', raw_output)

- # Delete audit

- raw_output = cls.watcher('audit delete %s' % cls.audit_uuid)

- cls.assertOutput('', raw_output)

- # Delete Template

- raw_output = cls.watcher(

- 'audittemplate delete %s' % cls.audit_template_name)

- cls.assertOutput('', raw_output)

-

- def test_action_plan_list(self):

- raw_output = self.watcher('actionplan list')

- self.assert_table_structure([raw_output], self.list_fields)

-

- def test_action_plan_detailed_list(self):

- raw_output = self.watcher('actionplan list --detail')

- self.assert_table_structure([raw_output], self.detailed_list_fields)

-

- def test_action_plan_show(self):

- action_plan_list = self.parse_show(self.watcher('actionplan list'))

- action_plan_uuid = list(action_plan_list[0])[0]

- actionplan = self.watcher('actionplan show %s' % action_plan_uuid)

- self.assertIn(action_plan_uuid, actionplan)

- self.assert_table_structure([actionplan],

- self.detailed_list_fields)

-

- def test_action_plan_start(self):

- output = self.parse_show(self.watcher('actionplan list --audit %s'

- % self.audit_uuid))

- action_plan_uuid = list(output[0])[0]

- self.watcher('actionplan start %s' % action_plan_uuid)

- raw_output = self.watcher('actionplan show %s' % action_plan_uuid)

- self.assert_table_structure([raw_output], self.detailed_list_fields)

-

- self.assertTrue(test_utils.call_until_true(

- func=functools.partial(

- self.has_actionplan_succeeded, action_plan_uuid),

- duration=600,

- sleep_for=2

- ))

-# Copyright (c) 2016 Servionica

-#

-# Licensed under the Apache License, Version 2.0 (the "License");

-# you may not use this file except in compliance with the License.

-# You may obtain a copy of the License at

-#

-# http://www.apache.org/licenses/LICENSE-2.0

-#

-# Unless required by applicable law or agreed to in writing, software

-# distributed under the License is distributed on an "AS IS" BASIS,

-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or

-# implied.

-# See the License for the specific language governing permissions and

-# limitations under the License.

-

-from oslo_utils import uuidutils

-

-import functools

-

-from tempest.lib.common.utils import test_utils

-

-from watcherclient.tests.functional.v1 import base

-

-

-class AuditTests(base.TestCase):

- """Functional tests for audit."""

-

- dummy_name = 'dummy'

- list_fields = ['UUID', 'Name', 'Audit Type', 'State', 'Goal', 'Strategy']

- detailed_list_fields = list_fields + ['Created At', 'Updated At',

- 'Deleted At', 'Parameters',

- 'Interval', 'Audit Scope',

- 'Next Run Time', 'Hostname']

- audit_template_name = 'a' + uuidutils.generate_uuid()

- audit_uuid = None

-

- @classmethod

- def setUpClass(cls):

- raw_output = cls.watcher('audittemplate create %s dummy -s dummy'

- % cls.audit_template_name)

- template_output = cls.parse_show_as_object(raw_output)

- audit_raw_output = cls.watcher(

- 'audit create -a %s' % template_output['Name'])

- audit_output = cls.parse_show_as_object(audit_raw_output)

- cls.audit_uuid = audit_output['UUID']

- audit_created = test_utils.call_until_true(

- func=functools.partial(cls.has_audit_created, cls.audit_uuid),

- duration=600,

- sleep_for=2)

- if not audit_created:

- raise Exception('Audit has not been succeeded')

-

- @classmethod

- def tearDownClass(cls):

- output = cls.parse_show(

- cls.watcher('actionplan list --audit %s' % cls.audit_uuid))

- action_plan_uuid = list(output[0])[0]

- cls.watcher('actionplan delete %s' % action_plan_uuid)

- cls.watcher('audit delete %s' % cls.audit_uuid)

- cls.watcher('audittemplate delete %s' % cls.audit_template_name)

-

- def test_audit_list(self):

- raw_output = self.watcher('audit list')

- self.assert_table_structure([raw_output], self.list_fields)

-

- def test_audit_detailed_list(self):

- raw_output = self.watcher('audit list --detail')

- self.assert_table_structure([raw_output], self.detailed_list_fields)

-

- def test_audit_show(self):

- audit = self.watcher('audit show ' + self.audit_uuid)

- self.assertIn(self.audit_uuid, audit)

- self.assert_table_structure([audit], self.detailed_list_fields)

-

- def test_audit_update(self):

- audit_raw_output = self.watcher('audit update %s add interval=2'

- % self.audit_uuid)

- audit_output = self.parse_show_as_object(audit_raw_output)

- assert int(audit_output['Interval']) == 2

-

-

-class AuditActiveTests(base.TestCase):

-

- list_fields = ['UUID', 'Name', 'Audit Type', 'State', 'Goal', 'Strategy']

- detailed_list_fields = list_fields + ['Created At', 'Updated At',

- 'Deleted At', 'Parameters',

- 'Interval', 'Audit Scope']

- audit_template_name = 'a' + uuidutils.generate_uuid()

-

- @classmethod

- def setUpClass(cls):

- cls.watcher('audittemplate create %s dummy -s dummy'

- % cls.audit_template_name)

-

- @classmethod

- def tearDownClass(cls):

- cls.watcher('audittemplate delete %s' % cls.audit_template_name)

-

- def _create_audit(self):

- return self.parse_show_as_object(

- self.watcher('audit create -a %s'

- % self.audit_template_name))['UUID']

-

- def _delete_audit(self, audit_uuid):

- self.assertTrue(test_utils.call_until_true(

- func=functools.partial(

- self.has_audit_created, audit_uuid),

- duration=600,

- sleep_for=2

- ))

- output = self.parse_show(

- self.watcher('actionplan list --audit %s' % audit_uuid))

- action_plan_uuid = list(output[0])[0]

- self.watcher('actionplan delete %s' % action_plan_uuid)

- self.watcher('audit delete %s' % audit_uuid)

-

- def test_create_audit(self):

- audit = self.watcher('audit create -a %s' % self.audit_template_name)

- audit_uuid = self.parse_show_as_object(audit)['UUID']

- self.assert_table_structure([audit], self.detailed_list_fields)

- self._delete_audit(audit_uuid)

-

- def test_delete_audit(self):

- audit_uuid = self._create_audit()

- self.assertTrue(test_utils.call_until_true(

- func=functools.partial(

- self.has_audit_created, audit_uuid),

- duration=600,

- sleep_for=2

- ))

- raw_output = self.watcher('audit delete %s' % audit_uuid)

- self.assertOutput('', raw_output)

- output = self.parse_show(

- self.watcher('actionplan list --audit %s' % audit_uuid))

- action_plan_uuid = list(output[0])[0]

- self.watcher('actionplan delete %s' % action_plan_uuid)

-# Copyright (c) 2016 Servionica

-#

-# Licensed under the Apache License, Version 2.0 (the "License");

-# you may not use this file except in compliance with the License.

-# You may obtain a copy of the License at

-#

-# http://www.apache.org/licenses/LICENSE-2.0

-#

-# Unless required by applicable law or agreed to in writing, software

-# distributed under the License is distributed on an "AS IS" BASIS,

-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or

-# implied.

-# See the License for the specific language governing permissions and

-# limitations under the License.

-

-from oslo_utils import uuidutils

-

-from watcherclient.tests.functional.v1 import base

-

-

-class AuditTemplateTests(base.TestCase):

- """Functional tests for audit template."""

-

- dummy_name = 'dummy'

- list_fields = ['UUID', 'Name', 'Goal', 'Strategy']

- detailed_list_fields = list_fields + ['Created At', 'Updated At',

- 'Deleted At', 'Description',

- 'Audit Scope']

- audit_template_name = 'a' + uuidutils.generate_uuid()

-

- @classmethod

- def setUpClass(cls):

- cls.watcher('audittemplate create %s dummy -s dummy'

- % cls.audit_template_name)

-

- @classmethod

- def tearDownClass(cls):

- cls.watcher('audittemplate delete %s' % cls.audit_template_name)

-

- def test_audit_template_list(self):

- raw_output = self.watcher('audittemplate list')

- self.assert_table_structure([raw_output], self.list_fields)

-

- def test_audit_template_detailed_list(self):

- raw_output = self.watcher('audittemplate list --detail')

- self.assert_table_structure([raw_output], self.detailed_list_fields)

-

- def test_audit_template_show(self):

- audit_template = self.watcher(

- 'audittemplate show %s' % self.audit_template_name)

- self.assertIn(self.audit_template_name, audit_template)

- self.assert_table_structure([audit_template],

- self.detailed_list_fields)

-

- def test_audit_template_update(self):

- raw_output = self.watcher('audittemplate update %s replace '

- 'description="Updated Desc"'

- % self.audit_template_name)

- audit_template_output = self.parse_show_as_object(raw_output)

- assert audit_template_output['Description'] == 'Updated Desc'

-

-

-class AuditTemplateActiveTests(base.TestCase):

-

- audit_template_name = 'b' + uuidutils.generate_uuid()

- list_fields = ['UUID', 'Name', 'Goal', 'Strategy']

- detailed_list_fields = list_fields + ['Created At', 'Updated At',

- 'Deleted At', 'Description',

- 'Audit Scope']

-

- def _create_audit_template(self):

- self.watcher('audittemplate create %s dummy -s dummy '

- '-d "Test Audit Template"' % self.audit_template_name)

-

- def _delete_audit_template(self):

- self.watcher('audittemplate delete %s' % self.audit_template_name)

-

- def test_create_audit_template(self):

- raw_output = self.watcher('audittemplate create %s dummy '

- '-s dummy -d "Test Audit Template"'

- % self.audit_template_name)

- self.assert_table_structure([raw_output], self.detailed_list_fields)

- self._delete_audit_template()

-

- def test_delete_audit_template(self):

- self._create_audit_template()

- raw_output = self.watcher('audittemplate delete %s'

- % self.audit_template_name)

- self.assertOutput('', raw_output)

-# Copyright (c) 2016 Servionica

-#

-# Licensed under the Apache License, Version 2.0 (the "License");

-# you may not use this file except in compliance with the License.

-# You may obtain a copy of the License at

-#

-# http://www.apache.org/licenses/LICENSE-2.0

-#

-# Unless required by applicable law or agreed to in writing, software

-# distributed under the License is distributed on an "AS IS" BASIS,

-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or

-# implied.

-# See the License for the specific language governing permissions and

-# limitations under the License.

-

-from watcherclient.tests.functional.v1 import base

-

-

-class GoalTests(base.TestCase):

- """Functional tests for goal."""

-

- dummy_name = 'dummy'

- list_fields = ['UUID', 'Name', 'Display name']

-

- def test_goal_list(self):

- raw_output = self.watcher('goal list')

- self.assertIn(self.dummy_name, raw_output)

- self.assert_table_structure([raw_output], self.list_fields)

-

- def test_goal_detailed_list(self):

- raw_output = self.watcher('goal list --detail')

- self.assertIn(self.dummy_name, raw_output)

- self.assert_table_structure(

- [raw_output], self.list_fields + ['Efficacy specification'])

-

- def test_goal_show(self):

- raw_output = self.watcher('goal show %s' % self.dummy_name)

- self.assertIn(self.dummy_name, raw_output)

- self.assert_table_structure(

- [raw_output], self.list_fields + ['Efficacy specification'])

- self.assertNotIn('server_consolidation', raw_output)

-# Copyright (c) 2016 Servionica

-#

-# Licensed under the Apache License, Version 2.0 (the "License");

-# you may not use this file except in compliance with the License.

-# You may obtain a copy of the License at

-#

-# http://www.apache.org/licenses/LICENSE-2.0

-#

-# Unless required by applicable law or agreed to in writing, software

-# distributed under the License is distributed on an "AS IS" BASIS,

-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or

-# implied.

-# See the License for the specific language governing permissions and

-# limitations under the License.

-

-from watcherclient.tests.functional.v1 import base

-

-

-class ScoringEngineTests(base.TestCase):

- """Functional tests for scoring engine."""

-

- dummy_name = 'dummy_scorer'

- list_fields = ['UUID', 'Name', 'Description']

- detailed_list_fields = list_fields + ['Metainfo']

-

- def test_scoringengine_list(self):

- raw_output = self.watcher('scoringengine list')

- self.assertIn(self.dummy_name, raw_output)

- self.assert_table_structure([raw_output], self.list_fields)

-

- def test_scoringengine_detailed_list(self):

- raw_output = self.watcher('scoringengine list --detail')

- self.assertIn(self.dummy_name, raw_output)

- self.assert_table_structure([raw_output], self.detailed_list_fields)

-

- def test_scoringengine_show(self):

- raw_output = self.watcher('scoringengine show %s' % self.dummy_name)

- self.assertIn(self.dummy_name, raw_output)

- self.assert_table_structure([raw_output], self.detailed_list_fields)

- self.assertNotIn('dummy_avg_scorer', raw_output)

-# Copyright (c) 2016 Servionica

-#

-# Licensed under the Apache License, Version 2.0 (the "License");

-# you may not use this file except in compliance with the License.

-# You may obtain a copy of the License at

-#

-# http://www.apache.org/licenses/LICENSE-2.0

-#

-# Unless required by applicable law or agreed to in writing, software

-# distributed under the License is distributed on an "AS IS" BASIS,

-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or

-# implied.

-# See the License for the specific language governing permissions and

-# limitations under the License.

-

-from watcherclient.tests.functional.v1 import base

-

-

-class ServiceTests(base.TestCase):

- """Functional tests for service."""

-

- decision_engine_name = 'watcher-decision-engine'

- applier_name = 'watcher-applier'

- list_fields = ['ID', 'Name', 'Host', 'Status']

-

- def test_service_list(self):

- raw_output = self.watcher('service list')

- self.assertIn(self.decision_engine_name, raw_output)

- self.assertIn(self.applier_name, raw_output)

- self.assert_table_structure([raw_output], self.list_fields)

-

- def test_service_detailed_list(self):

- raw_output = self.watcher('service list --detail')

- self.assertIn(self.decision_engine_name, raw_output)

- self.assertIn(self.applier_name, raw_output)

- self.assert_table_structure([raw_output],

- self.list_fields + ['Last seen up'])

-

- def test_service_show(self):

- # TODO(alexchadin): this method should be refactored since Watcher will

- # get HA support soon.

- raw_output = self.watcher('service show %s'

- % self.decision_engine_name)

- self.assertIn(self.decision_engine_name, raw_output)

- self.assert_table_structure([raw_output],

- self.list_fields + ['Last seen up'])

- self.assertNotIn(self.applier_name, raw_output)

-# Copyright (c) 2016 Servionica

-#

-# Licensed under the Apache License, Version 2.0 (the "License");

-# you may not use this file except in compliance with the License.

-# You may obtain a copy of the License at

-#

-# http://www.apache.org/licenses/LICENSE-2.0

-#

-# Unless required by applicable law or agreed to in writing, software

-# distributed under the License is distributed on an "AS IS" BASIS,

-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or

-# implied.

-# See the License for the specific language governing permissions and

-# limitations under the License.

-

-from watcherclient.tests.functional.v1 import base

-

-

-class StrategyTests(base.TestCase):

- """Functional tests for strategy."""

-

- dummy_name = 'dummy'

- basic_strategy = 'basic'

- list_fields = ['UUID', 'Name', 'Display name', 'Goal']

- state_fields = ['Datasource', 'Metrics', 'CDM', 'Name']

-

- def test_strategy_list(self):

- raw_output = self.watcher('strategy list')

- self.assertIn(self.dummy_name, raw_output)

- self.assert_table_structure([raw_output], self.list_fields)

-

- def test_strategy_detailed_list(self):

- raw_output = self.watcher('strategy list --detail')

- self.assertIn(self.dummy_name, raw_output)

- self.assert_table_structure([raw_output],

- self.list_fields + ['Parameters spec'])

-

- def test_strategy_show(self):

- raw_output = self.watcher('strategy show %s' % self.dummy_name)

- self.assertIn(self.dummy_name, raw_output)

- self.assert_table_structure([raw_output],

- self.list_fields + ['Parameters spec'])

- self.assertNotIn('basic', raw_output)

-

- def test_strategy_state(self):

- raw_output = self.watcher('strategy state %s' % self.basic_strategy)

- self.assertIn(self.basic_strategy, raw_output)

- self.assert_table_structure([raw_output], self.state_fields)

- 'os_watcher_api_version': "latest",

- 'os_watcher_api_version': "1.4",

- 'os_watcher_api_version': 'latest',

- 'os_watcher_api_version': 'latest',

- 'os_watcher_api_version': None,

- 'os_watcher_api_version': None,

- def setUp(self):

- 'os_watcher_api_version': '1'}

- if formatting:

-from watcherclient.v1 import resource_fields

-AUDIT_1 = {

- 'uuid': '5869da81-4876-4687-a1ed-12cd64cf53d9',

- 'audit_type': 'ONESHOT',

- 'state': 'PENDING',

- 'audit_template_uuid': 'f8e47706-efcf-49a4-a5c4-af604eb492f2',

- 'audit_template_name': 'at1',

- 'goal_name': 'SERVER_CONSOLIDATION',

- 'strategy_name': 'basic',

- 'created_at': datetime.datetime.now().isoformat(),

- 'updated_at': None,

- 'deleted_at': None,

- 'parameters': None,

- 'interval': None,

- 'scope': '',

- 'auto_trigger': False,

- 'next_run_time': None,

- 'name': 'my_audit1',

- 'hostname': '',

-}

-

-AUDIT_2 = {

- 'uuid': 'a5199d0e-0702-4613-9234-5ae2af8dafea',

- 'audit_type': 'ONESHOT',

- 'state': 'PENDING',

- 'audit_template_uuid': 'f8e47706-efcf-49a4-a5c4-af604eb492f2',

- 'audit_template_name': 'at1',

- 'goal_name': 'fc087747-61be-4aad-8126-b701731ae836',

- 'strategy_name': 'auto',

- 'created_at': datetime.datetime.now().isoformat(),

- 'updated_at': None,

- 'deleted_at': None,

- 'parameters': None,

- 'interval': None,

- 'scope': '',

- 'auto_trigger': False,

- 'next_run_time': None,

- 'name': 'my_audit2',

- 'hostname': '',

-}

-

-AUDIT_3 = {

- 'uuid': '43199d0e-0712-1213-9674-5ae2af8dhgte',

- 'audit_type': 'ONESHOT',

- 'state': 'PENDING',

- 'audit_template_uuid': 'f8e47706-efcf-49a4-a5c4-af604eb492f2',

- 'audit_template_name': 'at1',

- 'goal_name': None,

- 'strategy_name': 'auto',

- 'created_at': datetime.datetime.now().isoformat(),

- 'updated_at': None,

- 'deleted_at': None,

- 'parameters': None,

- 'interval': 3600,

- 'scope': '',

- 'auto_trigger': True,

- 'next_run_time': None,

- 'name': 'my_audit3',

- 'hostname': '',

-}

-

- SHORT_LIST_FIELDS = resource_fields.AUDIT_SHORT_LIST_FIELDS

- SHORT_LIST_FIELD_LABELS = resource_fields.AUDIT_SHORT_LIST_FIELD_LABELS

- FIELDS = resource_fields.AUDIT_FIELDS

- FIELD_LABELS = resource_fields.AUDIT_FIELD_LABELS

-

- def setUp(self):

- super(self.__class__, self).setUp()

- audit1 = resource.Audit(mock.Mock(), AUDIT_1)

- audit2 = resource.Audit(mock.Mock(), AUDIT_2)

- audit2 = resource.Audit(mock.Mock(), AUDIT_2)

- audit1 = resource.Audit(mock.Mock(), AUDIT_1)

- audit2 = resource.Audit(mock.Mock(), AUDIT_2)

- audit = resource.Audit(mock.Mock(), AUDIT_1)

- audit = resource.Audit(mock.Mock(), AUDIT_1)

- audit = resource.Audit(mock.Mock(), AUDIT_1)

- audit = resource.Audit(mock.Mock(), AUDIT_1)

- audit = resource.Audit(mock.Mock(), AUDIT_3)

- audit = resource.Audit(mock.Mock(), AUDIT_3)

- audit = resource.Audit(mock.Mock(), AUDIT_1)

- audit = resource.Audit(mock.Mock(), AUDIT_1)

- audit = resource.Audit(mock.Mock(), AUDIT_1)

- audit = resource.Audit(mock.Mock(), AUDIT_1)

- audit = resource.Audit(mock.Mock(), AUDIT_1)

- audit = resource.Audit(mock.Mock(), AUDIT_1)

- "Service", "ServiceManager", "Strategy", "StrategyManager")

- 'name']

- 'interval', 'goal', 'strategy', 'auto_trigger', 'name']

- if kwargs.get('os_watcher_api_version'):

- _("Must provide 'endpoint' if os_watcher_api_version "

- kwargs['os_watcher_api_version'] = httpclient.DEFAULT_VER

- def list(self, limit=None, sort_key=None, sort_dir=None, detail=False):

- :returns: A list of audits.

- filters = utils.common_filters(limit, sort_key, sort_dir)

- 'hostname']

- 'Next Run Time', 'Hostname']

- sort_dir=None, detail=False):

-

- filters = utils.common_filters(limit, sort_key, sort_dir)

-