openstack
/
fuel-qa
Code Issues Proposed changes

Divide the test and the class with useful methods 

Because CommandLine class used in
    create_backup_reset_restore_and_deploy_via_cli it should not have test
    decorator. Proboscis saves test function each time when module are
    imported

Change-Id: I30fd2596f03f10992b0d743c7e375325b45bf90c
Closes-bug: #1507999
This commit is contained in:
Dmitry Tyzhnenko 2015-10-28 16:44:49 +02:00 committed by Dmitry Tyzhnenko
parent c1ab3a2d0b
commit 4e402e32e3
4 changed files with 135 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
doc/base_tests.rst
View File

@ -46,6 +46,11 @@ Test Cli
.. automodule:: fuelweb_test.tests.test_cli
:members:
Test Cli Base
-------------
.. automodule:: fuelweb_test.tests.test_cli_base
:members:
Test Clone Environment
----------------------
.. automodule:: fuelweb_test.tests.test_clone_env

2
fuelweb_test/tests/test_backup_restore.py
View File

@ -31,7 +31,7 @@ from fuelweb_test.helpers.utils import run_on_remote
from fuelweb_test.settings import DEPLOYMENT_MODE
from fuelweb_test.settings import NEUTRON_SEGMENT_TYPE
from fuelweb_test.settings import OPENSTACK_RELEASE
from fuelweb_test.tests.test_cli import CommandLine
from fuelweb_test.tests.test_cli_base import CommandLine
from fuelweb_test import logger

105
fuelweb_test/tests/test_cli.py
View File

@ -12,9 +12,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import time
import json
from proboscis import test
from proboscis.asserts import assert_equal
from proboscis.asserts import assert_false
@ -26,14 +23,13 @@ from fuelweb_test.helpers.checkers import check_cluster_presence
from fuelweb_test.helpers.checkers import check_cobbler_node_exists
from fuelweb_test.helpers.decorators import log_snapshot_after_test
from fuelweb_test.helpers.utils import run_on_remote
from fuelweb_test.helpers.ssl import change_cluster_ssl_config
from fuelweb_test.settings import DEPLOYMENT_MODE
from fuelweb_test.settings import NEUTRON_ENABLE
from fuelweb_test.settings import NEUTRON_SEGMENT_TYPE
from fuelweb_test.settings import OPENSTACK_RELEASE
from fuelweb_test.tests.base_test_case import SetupEnvironment
from fuelweb_test.tests.base_test_case import TestBasic
from fuelweb_test import logwrap
from fuelweb_test.tests import test_cli_base
from fuelweb_test import logger
@ -91,106 +87,9 @@ class CommandLineMinimal(TestBasic):
@test(groups=["command_line"])
class CommandLine(TestBasic):
class CommandLineTest(test_cli_base.CommandLine):
"""CommandLine.""" # TODO documentation
@logwrap
def get_task(self, remote, task_id):
tasks = run_on_remote(remote, 'fuel task --task-id {0} --json'
.format(task_id), jsonify=True)
return tasks[0]
@logwrap
def get_network_filename(self, cluster_id, remote):
cmd = ('fuel --env {0} network --download --dir /tmp --json'
.format(cluster_id))
net_download = ''.join(run_on_remote(remote, cmd))
# net_download = 'Network ... downloaded to /tmp/network_1.json'
return net_download.split()[-1]
@logwrap
def get_networks(self, cluster_id, remote):
net_file = self.get_network_filename(cluster_id, remote)
return run_on_remote(remote, 'cat {0}'.format(net_file), jsonify=True)
@logwrap
def update_network(self, cluster_id, remote, net_config):
net_file = self.get_network_filename(cluster_id, remote)
data = json.dumps(net_config)
cmd = 'echo {data} > {net_file}'.format(data=json.dumps(data),
net_file=net_file)
run_on_remote(remote, cmd)
cmd = ('cd /tmp; fuel --env {0} network --upload --json'
.format(cluster_id))
run_on_remote(remote, cmd)
def assert_cli_task_success(
self, task, remote, timeout=70 * 60, interval=20):
logger.info('Wait {timeout} seconds for task: {task}'
.format(timeout=timeout, task=task))
start = time.time()
try:
wait(
lambda: self.get_task(
remote, task['id'])['status'] != 'running',
interval=interval,
timeout=timeout
)
except TimeoutError:
raise TimeoutError(
"Waiting timeout {timeout} sec was reached for task: {task}"
.format(task=task["name"], timeout=timeout))
took = time.time() - start
task = self.get_task(remote, task['id'])
logger.info('Task finished in {took} seconds with the result: {task}'
.format(took=took, task=task))
assert_equal(
task['status'], 'ready',
"Task '{name}' has incorrect status. {} != {}".format(
task['status'], 'ready', name=task["name"]
)
)
@logwrap
def update_cli_network_configuration(self, cluster_id, remote):
"""Update cluster network settings with custom configuration.
Place here an additional config changes if needed (e.g. nodegroups'
networking configuration.
Also this method checks downloading/uploading networks via cli.
"""
net_config = self.get_networks(cluster_id, remote)
new_settings = net_config
self.update_network(cluster_id, remote, new_settings)
def get_public_vip(self, cluster_id, remote):
networks = self.get_networks(cluster_id, remote)
return networks['public_vip']
def download_settings(self, cluster_id, remote):
cmd = ('fuel --env {0} settings --download --dir /tmp --json'.format(
cluster_id))
run_on_remote(remote, cmd)
return run_on_remote(remote,
'cd /tmp && cat settings_{0}.json'.format(
cluster_id), jsonify=True)
def upload_settings(self, cluster_id, remote, settings):
data = json.dumps(settings)
cmd = 'cd /tmp && echo {data} > settings_{id}.json'.format(
data=json.dumps(data),
id=cluster_id)
run_on_remote(remote, cmd)
cmd = ('fuel --env {0} settings --upload --dir /tmp --json'.format(
cluster_id))
run_on_remote(remote, cmd)
@logwrap
def update_ssl_configuration(self, cluster_id, remote):
settings = self.download_settings(cluster_id, remote)
cn = self.get_public_vip(cluster_id, remote)
change_cluster_ssl_config(settings, cn)
self.upload_settings(cluster_id, remote, settings)
@test(depends_on=[SetupEnvironment.prepare_slaves_3],
groups=["cli_selected_nodes_deploy"])
@log_snapshot_after_test

127
fuelweb_test/tests/test_cli_base.py Normal file
View File

@ -0,0 +1,127 @@
# Copyright 2015 Mirantis, 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.
import time
import json
from proboscis.asserts import assert_equal
from devops.error import TimeoutError
from devops.helpers.helpers import wait
from fuelweb_test.helpers.utils import run_on_remote
from fuelweb_test.helpers.ssl import change_cluster_ssl_config
from fuelweb_test.tests.base_test_case import TestBasic
from fuelweb_test import logwrap
from fuelweb_test import logger
class CommandLine(TestBasic):
"""CommandLine.""" # TODO documentation
@logwrap
def get_task(self, remote, task_id):
tasks = run_on_remote(remote, 'fuel task --task-id {0} --json'
.format(task_id), jsonify=True)
return tasks[0]
@logwrap
def get_network_filename(self, cluster_id, remote):
cmd = ('fuel --env {0} network --download --dir /tmp --json'
.format(cluster_id))
net_download = ''.join(run_on_remote(remote, cmd))
# net_download = 'Network ... downloaded to /tmp/network_1.json'
return net_download.split()[-1]
@logwrap
def get_networks(self, cluster_id, remote):
net_file = self.get_network_filename(cluster_id, remote)
return run_on_remote(remote, 'cat {0}'.format(net_file), jsonify=True)
@logwrap
def update_network(self, cluster_id, remote, net_config):
net_file = self.get_network_filename(cluster_id, remote)
data = json.dumps(net_config)
cmd = 'echo {data} > {net_file}'.format(data=json.dumps(data),
net_file=net_file)
run_on_remote(remote, cmd)
cmd = ('cd /tmp; fuel --env {0} network --upload --json'
.format(cluster_id))
run_on_remote(remote, cmd)
def assert_cli_task_success(
self, task, remote, timeout=70 * 60, interval=20):
logger.info('Wait {timeout} seconds for task: {task}'
.format(timeout=timeout, task=task))
start = time.time()
try:
wait(
lambda: self.get_task(
remote, task['id'])['status'] != 'running',
interval=interval,
timeout=timeout
)
except TimeoutError:
raise TimeoutError(
"Waiting timeout {timeout} sec was reached for task: {task}"
.format(task=task["name"], timeout=timeout))
took = time.time() - start
task = self.get_task(remote, task['id'])
logger.info('Task finished in {took} seconds with the result: {task}'
.format(took=took, task=task))
assert_equal(
task['status'], 'ready',
"Task '{name}' has incorrect status. {} != {}".format(
task['status'], 'ready', name=task["name"]
)
)
@logwrap
def update_cli_network_configuration(self, cluster_id, remote):
"""Update cluster network settings with custom configuration.
Place here an additional config changes if needed (e.g. nodegroups'
networking configuration.
Also this method checks downloading/uploading networks via cli.
"""
net_config = self.get_networks(cluster_id, remote)
new_settings = net_config
self.update_network(cluster_id, remote, new_settings)
def get_public_vip(self, cluster_id, remote):
networks = self.get_networks(cluster_id, remote)
return networks['public_vip']
def download_settings(self, cluster_id, remote):
cmd = ('fuel --env {0} settings --download --dir /tmp --json'.format(
cluster_id))
run_on_remote(remote, cmd)
return run_on_remote(remote,
'cd /tmp && cat settings_{0}.json'.format(
cluster_id), jsonify=True)
def upload_settings(self, cluster_id, remote, settings):
data = json.dumps(settings)
cmd = 'cd /tmp && echo {data} > settings_{id}.json'.format(
data=json.dumps(data),
id=cluster_id)
run_on_remote(remote, cmd)
cmd = ('fuel --env {0} settings --upload --dir /tmp --json'.format(
cluster_id))
run_on_remote(remote, cmd)
@logwrap
def update_ssl_configuration(self, cluster_id, remote):
settings = self.download_settings(cluster_id, remote)
cn = self.get_public_vip(cluster_id, remote)
change_cluster_ssl_config(settings, cn)
self.upload_settings(cluster_id, remote, settings)