fuel-qa/fuelweb_test/tests/test_cli_base.py
Alexey Stepanov 235959b861 Pylint: fix trivial issues 8
fix: make methods static, if self is not used inside
(call is not changed, backward change is possible without code change,
  if self will be required)
Related-bug: #1556791

Change-Id: I48175c0179e0fde69ed9b527ab570a267b9191a5
2016-03-18 10:35:15 +03:00

297 lines
11 KiB
Python

# 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 urlparse import urlparse
from proboscis.asserts import assert_equal
from devops.error import TimeoutError
from devops.helpers.helpers import wait
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
from fuelweb_test.helpers.utils import hiera_json_out
from fuelweb_test.settings import SSL_CN
class CommandLine(TestBasic):
"""CommandLine.""" # TODO documentation
@logwrap
def get_task(self, task_id):
tasks = self.ssh_manager.execute_on_remote(
ip=self.ssh_manager.admin_ip,
cmd='fuel task --task-id {0} --json'.format(task_id),
jsonify=True
)['stdout_json']
return tasks[0]
@logwrap
def get_network_filename(self, cluster_id):
cmd = ('fuel --env {0} network --download --dir /tmp --json'
.format(cluster_id))
out = self.ssh_manager.execute_on_remote(
ip=self.ssh_manager.admin_ip,
cmd=cmd
)['stdout']
net_download = ''.join(out)
# net_download = 'Network ... downloaded to /tmp/network_1.json'
return net_download.split()[-1]
@logwrap
def get_networks(self, cluster_id):
net_file = self.get_network_filename(cluster_id)
out = self.ssh_manager.execute_on_remote(
ip=self.ssh_manager.admin_ip,
cmd='cat {0}'.format(net_file),
jsonify=True
)['stdout_json']
return out
@logwrap
def update_network(self, cluster_id, net_config):
net_file = self.get_network_filename(cluster_id)
data = json.dumps(net_config)
cmd = 'echo {data} > {net_file}'.format(data=json.dumps(data),
net_file=net_file)
self.ssh_manager.execute_on_remote(
ip=self.ssh_manager.admin_ip,
cmd=cmd
)
cmd = ('cd /tmp; fuel --env {0} network --upload --json'
.format(cluster_id))
self.ssh_manager.execute_on_remote(
ip=self.ssh_manager.admin_ip,
cmd=cmd
)
def assert_cli_task_success(self, task, 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(task['id'])['status'] not in
('pending', '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(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"]
)
)
@staticmethod
@logwrap
def hiera_floating_ranges(node_ip):
"""
1. SSH to controller node
2. Get network settings from controller node
3. Convert to json network settings in variable config_json
4. Get new list of floating ranges in variable floating ranges
5. Convert to sublist floating ranges in variable floating_ranges_json
"""
config_json = hiera_json_out(node_ip, 'quantum_settings')
floating_ranges = \
config_json[
"predefined_networks"][
"admin_floating_net"][
"L3"]["floating"]
floating_ranges_json = [
[float_address[0], float_address[1]] for float_address in (
float_address.split(':') for float_address in floating_ranges)]
return floating_ranges_json
@logwrap
def get_floating_ranges(self, cluster_id):
"""
This method using for get floating ranges from master node before
cluster will be deployed.
1. SSH to master node
2. Get networks from master node
3. Save floating ranges from master node
"""
net_config = self.get_networks(cluster_id)
floating_ranges =\
net_config[u'networking_parameters'][u'floating_ranges']
return floating_ranges
@logwrap
def change_floating_ranges(self, cluster_id, floating_range):
net_config = self.get_networks(cluster_id)
net_config[u'networking_parameters'][u'floating_ranges'] = \
floating_range
new_settings = net_config
self.update_network(cluster_id, new_settings)
@logwrap
def update_cli_network_configuration(self, cluster_id):
"""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)
new_settings = net_config
self.update_network(cluster_id, new_settings)
def get_public_vip(self, cluster_id):
networks = self.get_networks(cluster_id)
return networks['public_vip']
def download_settings(self, cluster_id):
cmd = ('fuel --env {0} settings --download --dir /tmp --json'.format(
cluster_id))
self.ssh_manager.execute_on_remote(
ip=self.ssh_manager.admin_ip,
cmd=cmd
)
out = self.ssh_manager.execute_on_remote(
ip=self.ssh_manager.admin_ip,
cmd='cd /tmp && cat settings_{0}.json'.format(cluster_id),
jsonify=True
)['stdout_json']
return out
def upload_settings(self, cluster_id, settings):
data = json.dumps(settings)
cmd = 'cd /tmp && echo {data} > settings_{id}.json'.format(
data=json.dumps(data),
id=cluster_id)
self.ssh_manager.execute_on_remote(
ip=self.ssh_manager.admin_ip,
cmd=cmd
)
cmd = ('fuel --env {0} settings --upload --dir /tmp --json'.format(
cluster_id))
self.ssh_manager.execute_on_remote(
ip=self.ssh_manager.admin_ip,
cmd=cmd
)
@logwrap
def update_ssl_configuration(self, cluster_id):
settings = self.download_settings(cluster_id)
change_cluster_ssl_config(settings, SSL_CN)
self.upload_settings(cluster_id, settings)
def add_nodes_to_cluster(self, cluster_id, node_ids, roles):
if isinstance(node_ids, int):
node_ids_str = str(node_ids)
else:
node_ids_str = ','.join(str(n) for n in node_ids)
cmd = ('fuel --env-id={0} node set --node {1} --role={2}'.format(
cluster_id, node_ids_str, ','.join(roles)))
self.ssh_manager.execute_on_remote(
ip=self.ssh_manager.admin_ip,
cmd=cmd
)
@logwrap
def use_ceph_for_volumes(self, cluster_id):
settings = self.download_settings(cluster_id)
settings['editable']['storage']['volumes_lvm'][
'value'] = False
settings['editable']['storage']['volumes_ceph'][
'value'] = True
self.upload_settings(cluster_id, settings)
@logwrap
def use_ceph_for_images(self, cluster_id):
settings = self.download_settings(cluster_id)
settings['editable']['storage']['images_ceph'][
'value'] = True
self.upload_settings(cluster_id, settings)
@logwrap
def use_ceph_for_ephemeral(self, cluster_id):
settings = self.download_settings(cluster_id)
settings['editable']['storage']['ephemeral_ceph'][
'value'] = True
self.upload_settings(cluster_id, settings)
@logwrap
def change_osd_pool_size(self, cluster_id, replication_factor):
settings = self.download_settings(cluster_id)
settings['editable']['storage']['osd_pool_size'][
'value'] = replication_factor
self.upload_settings(cluster_id, settings)
@logwrap
def use_radosgw_for_objects(self, cluster_id):
settings = self.download_settings(cluster_id)
ceph_for_images = settings['editable']['storage']['images_ceph'][
'value']
if ceph_for_images:
settings['editable']['storage']['objects_ceph'][
'value'] = True
else:
settings['editable']['storage']['images_ceph'][
'value'] = True
settings['editable']['storage']['objects_ceph'][
'value'] = True
self.upload_settings(cluster_id, settings)
@logwrap
def get_current_ssl_cn(self, controller_ip):
cmd = "openssl x509 -noout -subject -in \
/var/lib/astute/haproxy/public_haproxy.pem \
| sed -n '/^subject/s/^.*CN=//p'"
ssl_cn = self.ssh_manager.execute_on_remote(
ip=controller_ip,
cmd=cmd)['stdout_str']
return ssl_cn
@logwrap
def get_current_ssl_keypair(self, controller_ip):
cmd = "cat /var/lib/astute/haproxy/public_haproxy.pem"
current_ssl_keypair = self.ssh_manager.execute_on_remote(
ip=controller_ip,
cmd=cmd)['stdout_str']
return current_ssl_keypair
@logwrap
def get_endpoints(self, controller_ip):
cmd = "source openrc;export OS_IDENTITY_API_VERSION=3;" \
"openstack endpoint list -f json"
endpoints = []
endpoint_list =\
self.ssh_manager.execute_on_remote(ip=controller_ip,
cmd=cmd,
jsonify=True)['stdout_json']
for endpoint in endpoint_list:
if endpoint['Interface'] == 'public':
url = urlparse(endpoint['URL'])
endpoint_info = {'service_name': endpoint['Service Name'],
'protocol': url.scheme,
'domain': url.hostname}
endpoints.append(endpoint_info)
return endpoints