Add YamlEditor from fuel-qa repo

YamlEditor is a simple way to manipulate YAML content
in the files stored on the localhost or on the remote host.

- YamlEditor moved from fuel-qa/fuelweb-test/helpers/utils.py
- add possibility to work with multi-document YAMLs (variable
  document_id sets the current document to edit)
- add 'multi_constructor' and 'representer' to work with YAMLs
  that are using different tags (for example "!include") without
  parsing these tags.

Examples:

// Local YAML file
with underlay.yaml_editor('/path/to/file') as editor:
    editor.content[key] = "value"

// Remote YAML file on k8s host
with underlay.yaml_editor('/path/to/file',
                          host=config.k8s.kube_host) as editor:
    editor.content[key] = "value"

Change-Id: I4741197125c508a90dac5c8df93381a040542a4a
This commit is contained in:
Dennis Dmitriev 2016-10-11 17:26:57 +03:00
parent 07204d077d
commit 6854c4f447
2 changed files with 166 additions and 1 deletions

View File

@ -12,7 +12,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import copy
import os
import shutil
import tempfile
@ -22,6 +22,7 @@ import traceback
import paramiko
import yaml
from devops.helpers import helpers
from devops.helpers import ssh_client
from elasticsearch import Elasticsearch
from fuel_ccp_tests import logger
@ -277,3 +278,136 @@ def rm_files(node, pod, path,
pod.name,
namespace,
'rm -- {}'.format(path)))
class YamlEditor(object):
"""Manipulations with local or remote .yaml files.
Usage:
with YamlEditor("tasks.yaml") as editor:
editor.content[key] = "value"
with YamlEditor("astute.yaml", ip=self.admin_ip) as editor:
editor.content[key] = "value"
"""
def __init__(self, file_path, host=None, port=None,
username=None, password=None, private_keys=None,
document_id=0,
default_flow_style=False, default_style=None):
self.__file_path = file_path
self.host = host
self.port = port or 22
self.username = username
self.__password = password
self.__private_keys = private_keys or []
self.__content = None
self.__documents = [{}, ]
self.__document_id = document_id
self.__original_content = None
self.default_flow_style = default_flow_style
self.default_style = default_style
@property
def file_path(self):
"""Open file path
:rtype: str
"""
return self.__file_path
@property
def content(self):
if self.__content is None:
self.__content = self.get_content()
return self.__content
@content.setter
def content(self, new_content):
self.__content = new_content
def __get_file(self, mode="r"):
if self.host:
remote = ssh_client.SSHClient(
host=self.host,
port=self.port,
username=self.username,
password=self.__password,
private_keys=self.__private_keys)
return remote.open(self.__file_path, mode=mode)
else:
return open(self.__file_path, mode=mode)
def get_content(self):
"""Return a single document from YAML"""
def multi_constructor(loader, tag_suffix, node):
"""Stores all unknown tags content into a dict
Original yaml:
!unknown_tag
- some content
Python object:
{"!unknown_tag": ["some content", ]}
"""
if type(node.value) is list:
if type(node.value[0]) is tuple:
return {node.tag: loader.construct_mapping(node)}
else:
return {node.tag: loader.construct_sequence(node)}
else:
return {node.tag: loader.construct_scalar(node)}
yaml.add_multi_constructor("!", multi_constructor)
with self.__get_file() as file_obj:
self.__documents = [x for x in yaml.load_all(file_obj)]
return self.__documents[self.__document_id]
def write_content(self, content=None):
if content:
self.content = content
self.__documents[self.__document_id] = self.content
def representer(dumper, data):
"""Represents a dict key started with '!' as a YAML tag
Assumes that there is only one !tag in the dict at the
current indent.
Python object:
{"!unknown_tag": ["some content", ]}
Resulting yaml:
!unknown_tag
- some content
"""
key = data.keys()[0]
if key.startswith("!"):
value = data[key]
if type(value) is dict:
node = dumper.represent_mapping(key, value)
elif type(value) is list:
node = dumper.represent_sequence(key, value)
else:
node = dumper.represent_scalar(key, value)
else:
node = dumper.represent_mapping(u'tag:yaml.org,2002:map', data)
return node
yaml.add_representer(dict, representer)
with self.__get_file("w") as file_obj:
yaml.dump_all(self.__documents, file_obj,
default_flow_style=self.default_flow_style,
default_style=self.default_style)
def __enter__(self):
self.__content = self.get_content()
self.__original_content = copy.deepcopy(self.content)
return self
def __exit__(self, x, y, z):
if self.content == self.__original_content:
return
self.write_content()

View File

@ -19,6 +19,7 @@ from devops.helpers import ssh_client
from paramiko import rsakey
from fuel_ccp_tests import logger
from fuel_ccp_tests.helpers import utils
LOG = logger.logger
@ -328,3 +329,33 @@ class UnderlaySSHManager(object):
:return: str, name of node
"""
return random.choice(self.node_names())
def yaml_editor(self, file_path, node_name=None, host=None,
address_pool=None):
"""Returns an initialized YamlEditor instance for context manager
Usage (with 'underlay' fixture):
# Local YAML file
with underlay.yaml_editor('/path/to/file') as editor:
editor.content[key] = "value"
# Remote YAML file on k8s host
with underlay.yaml_editor('/path/to/file',
host=config.k8s.kube_host) as editor:
editor.content[key] = "value"
"""
# Local YAML file
if node_name is None and host is None:
return utils.YamlEditor(file_path=file_path)
# Remote YAML file
ssh_data = self.__ssh_data(node_name=node_name, host=host,
address_pool=address_pool)
return utils.YamlEditor(
file_path=file_path,
host=ssh_data['host'],
port=ssh_data['port'] or 22,
username=ssh_data['login'],
password=ssh_data['password'],
private_keys=ssh_data['keys'])