d0dc69c1eb
The "Compare switch VLANs to VLANs in nic config" validation breaks when run through Mistral. This is because the custom module used in the validation imports functions from another custom module. Custom modules are recognized by Ansible, but they don't necessarily live on the Python path, so when run through Mistral the imported module cannot be found and the validation breaks. This patch moves the two imported functions to different locations: - `library.network_environment.get_network_config` is refactored into a generic utility function in `tripleo_validations.utils`. - `library.network_environment.open_network_environment_files` is copied to `library.network_environment.switch_vlans` (this is an ugly duplication, but the function is going to be removed from the original location in a different patch: https://review.openstack.org/#/c/547518/) Change-Id: If3bc62df730739640341386022c0cd99295b880b Closes-Bug: #1761717
87 lines
3.0 KiB
Python
87 lines
3.0 KiB
Python
#!/usr/bin/env python
|
|
|
|
# Copyright 2017 Red Hat, Inc.
|
|
# 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.
|
|
|
|
from __future__ import print_function
|
|
|
|
import collections
|
|
|
|
from keystoneauth1.identity import generic as ks_id
|
|
from keystoneauth1 import session
|
|
from six import string_types
|
|
from swiftclient.client import Connection
|
|
|
|
|
|
def get_auth_session(auth_url, username, project_name, password=None,
|
|
auth_token=None, cacert=None):
|
|
if auth_token:
|
|
auth = ks_id.Token(auth_url=auth_url,
|
|
token=auth_token,
|
|
project_name=project_name,
|
|
project_domain_id='default')
|
|
else:
|
|
auth = ks_id.Password(auth_url=auth_url,
|
|
username=username,
|
|
password=password,
|
|
project_name=project_name,
|
|
user_domain_id='default',
|
|
project_domain_id='default')
|
|
return session.Session(auth=auth, verify=cacert)
|
|
|
|
|
|
def get_swift_client(preauthurl, preauthtoken):
|
|
return Connection(preauthurl=preauthurl,
|
|
preauthtoken=preauthtoken,
|
|
retries=10,
|
|
starting_backoff=3,
|
|
max_backoff=120)
|
|
|
|
|
|
def filtered(obj):
|
|
"""Only return properties of obj whose value can be properly serialized."""
|
|
return {k: v for k, v in obj.__dict__.items()
|
|
if isinstance(v, (string_types, int, list, dict, type(None)))}
|
|
|
|
|
|
def get_nested(data, name, path):
|
|
# Finds and returns a property from a nested dictionary by
|
|
# following a path of a defined set of property names and types.
|
|
|
|
def deep_find_key(key_data, data, name):
|
|
key, instance_type, instance_name = key_data
|
|
if key in data.keys():
|
|
if not isinstance(data[key], instance_type):
|
|
raise ValueError("The '{}' property of '{}' must be a {}."
|
|
"".format(key, name, instance_name))
|
|
return data[key]
|
|
for item in data.values():
|
|
if isinstance(item, collections.Mapping):
|
|
return deep_find_key(key_data, item, name)
|
|
return None
|
|
|
|
if not isinstance(data, collections.Mapping):
|
|
raise ValueError(
|
|
"'{}' is not a valid resource.".format(name))
|
|
|
|
current_value = data
|
|
while len(path) > 0:
|
|
key_data = path.pop(0)
|
|
current_value = deep_find_key(key_data, current_value, name)
|
|
if current_value is None:
|
|
break
|
|
|
|
return current_value
|