openstack/openstack-ansible
Code Issues Proposed changes
416b813776
openstack-ansible/tests/test_inventory.py
Kevin Carter beb36b8ab9 Remove the "is_ssh_address" option from inventory 
The key **is_ssh_address** has been removed from the
`openstack_user_config.yml` and the dynamic inventory. This key was
responsible mapping an address to the container which was used for SSH
connectivity. Because we've created the SSH connectivity plugin, which
allows us the ability to connect to remote containers without SSH, this
option is no longer useful. To keep the `openstack_user_config.yml` clean
deployers can remove the option however moving forward it no longer has any
effect.

Change-Id: I9264b73dcef71eba9ac29b238683a3a8c53e7121
Signed-off-by: Kevin Carter <kevin.carter@rackspace.com>
2018-03-15 00:58:13 +00:00

1462 lines
49 KiB
Python
Raw Blame History

# 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 collections
import copy
import json
import mock
import os
from os import path
try:
import Queue
except ImportError:
import queue as Queue
import sys
import unittest
import warnings
import yaml
INV_DIR = 'inventory'
sys.path.append(path.join(os.getcwd(), INV_DIR))
from osa_toolkit import dictutils
import dynamic_inventory
from osa_toolkit import filesystem as fs
from osa_toolkit import generate as di
from osa_toolkit import tools
TARGET_DIR = path.join(os.getcwd(), 'tests', 'inventory')
BASE_ENV_DIR = INV_DIR
CONFIGS_DIR = path.join(os.getcwd(), 'etc', 'openstack_deploy')
CONFD = os.path.join(CONFIGS_DIR, 'conf.d')
AIO_CONFIG_FILE = path.join(CONFIGS_DIR, 'openstack_user_config.yml.aio')
USER_CONFIG_FILE = path.join(TARGET_DIR, 'openstack_user_config.yml')
# These files will be placed in TARGET_DIR by the inventory functions
# They should be cleaned up between each test.
CLEANUP = [
'openstack_inventory.json',
'openstack_hostnames_ips.yml',
'backup_openstack_inventory.tar'
]
# Base config is a global configuration accessible for convenience.
# It should *not* be mutated outside of setUpModule, which populates it.
_BASE_CONFIG = {}
def get_config():
"""Return a copy of the original config so original isn't modified."""
global _BASE_CONFIG
return copy.deepcopy(_BASE_CONFIG)
def make_config():
"""Build an inventory configuration from the sample AIO files.
Take any files specified as '.aio' and load their keys into a
configuration dict and write them out to a file for consumption by
the tests.
"""
# Allow access here so we can populate the dictionary.
global _BASE_CONFIG
_BASE_CONFIG = tools.make_example_config(AIO_CONFIG_FILE, CONFD)
tools.write_example_config(USER_CONFIG_FILE, _BASE_CONFIG)
def setUpModule():
# The setUpModule function is used by the unittest framework.
make_config()
def tearDownModule():
# This file should only be removed after all tests are run,
# thus it is excluded from cleanup.
os.remove(USER_CONFIG_FILE)
def cleanup():
for f_name in CLEANUP:
f_file = path.join(TARGET_DIR, f_name)
if os.path.exists(f_file):
os.remove(f_file)
def get_inventory(clean=True, extra_args=None):
"Return the inventory mapping in a dict."
# Use the list argument to more closely mirror
# Ansible's use of the callable.
args = {'config': TARGET_DIR, 'list': True,
'environment': BASE_ENV_DIR}
if extra_args:
args.update(extra_args)
try:
inventory_string = di.main(**args)
inventory = json.loads(inventory_string)
return inventory
finally:
if clean:
# Remove the file system artifacts since we want to force
# fresh runs
cleanup()
class TestArgParser(unittest.TestCase):
def test_no_args(self):
arg_dict = dynamic_inventory.args([])
self.assertIsNone(arg_dict['config'])
self.assertEqual(arg_dict['list'], False)
def test_list_arg(self):
arg_dict = dynamic_inventory.args(['--list'])
self.assertEqual(arg_dict['list'], True)
def test_config_arg(self):
arg_dict = dynamic_inventory.args(['--config',
'/etc/openstack_deploy'])
self.assertEqual(arg_dict['config'], '/etc/openstack_deploy')
class TestAnsibleInventoryFormatConstraints(unittest.TestCase):
inventory = None
expected_groups = [
'aio1-host_containers',
'all',
'all_containers',
'aodh_alarm_evaluator',
'aodh_alarm_notifier',
'aodh_all',
'aodh_api',
'aodh_container',
'aodh_listener',
'barbican_all',
'barbican_api',
'barbican_container',
'ceilometer_all',
'ceilometer_agent_central',
'ceilometer_agent_compute',
'ceilometer_agent_notification',
'ceilometer_central_container',
'ceph_all',
'ceph-mon_all',
'ceph-mon_containers',
'ceph-mon_container',
'ceph-mon_hosts',
'ceph-mon',
'ceph-osd_all',
'ceph-osd_containers',
'ceph-osd_container',
'ceph-osd_hosts',
'ceph-osd',
'ceph-rgw_all',
'ceph-rgw_containers',
'ceph-rgw_container',
'ceph-rgw_hosts',
'ceph-rgw',
'cinder_all',
'cinder_api',
'cinder_api_container',
'cinder_backup',
'cinder_scheduler',
'cinder_volume',
'cinder_volumes_container',
'compute-infra_all',
'compute-infra_containers',
'compute-infra_hosts',
'compute_all',
'compute_containers',
'compute_hosts',
'dashboard_all',
'dashboard_containers',
'dashboard_hosts',
'database_containers',
'database_hosts',
'dnsaas_all',
'dnsaas_containers',
'dnsaas_hosts',
'designate_all',
'designate_container',
'designate_api',
'designate_central',
'designate_mdns',
'designate_worker',
'designate_producer',
'designate_sink',
'etcd',
'etcd_all',
'etcd_container',
'etcd_containers',
'etcd_hosts',
'galera',
'galera_all',
'galera_container',
'glance_all',
'glance_api',
'glance_container',
'glance_registry',
'gnocchi_all',
'gnocchi_api',
'gnocchi_container',
'gnocchi_metricd',
'haproxy',
'haproxy_all',
'haproxy_container',
'haproxy_containers',
'haproxy_hosts',
'heat_all',
'heat_api',
'heat_api_cfn',
'heat_api_container',
'heat_engine',
'horizon',
'horizon_all',
'horizon_container',
'hosts',
'identity_all',
'identity_containers',
'identity_hosts',
'image_all',
'image_containers',
'image_hosts',
'ironic-infra_all',
'ironic-infra_containers',
'ironic-infra_hosts',
'ironic-server_containers',
'ironic-server_hosts',
'ironic_all',
'ironic_api',
'ironic_api_container',
'ironic_conductor',
'ironic_server',
'ironic_server_container',
'ironic_servers',
'ironic_compute',
'ironic_compute_container',
'ironic-compute_containers',
'ironic-compute_all',
'ironic-compute_hosts',
'key-manager_containers',
'key-manager_hosts',
'keystone',
'keystone_all',
'keystone_container',
'kvm-compute_containers',
'kvm-compute_hosts',
'log_all',
'log_containers',
'log_hosts',
'lxc_hosts',
'lxd-compute_containers',
'lxd-compute_hosts',
'magnum',
'magnum-infra_all',
'magnum-infra_containers',
'magnum-infra_hosts',
'magnum_all',
'magnum_container',
'mano_all',
'mano_containers',
'mano_hosts',
'nspawn_hosts',
'octavia-infra_hosts',
'octavia_all',
'octavia-api',
'octavia_server_container',
'octavia-worker',
'octavia-housekeeping',
'octavia-health-manager',
'octavia-infra_containers',
'octavia-infra_all',
'powervm-compute_containers',
'powervm-compute_hosts',
'qemu-compute_containers',
'qemu-compute_hosts',
'trove_all',
'trove_api',
'trove_conductor',
'trove_taskmanager',
'trove_api_container',
'trove-infra_containers',
'trove-infra_hosts',
'trove-infra_all',
'memcached',
'memcached_all',
'memcached_container',
'memcaching_containers',
'memcaching_hosts',
'metering-alarm_all',
'metering-alarm_containers',
'metering-alarm_hosts',
'metering-compute_all',
'metering-compute_container',
'metering-compute_containers',
'metering-compute_hosts',
'metering-infra_all',
'metering-infra_containers',
'metering-infra_hosts',
'metrics_all',
'metrics_containers',
'metrics_hosts',
'molteniron',
'molteniron-server_all',
'molteniron-server_containers',
'molteniron-server_container',
'molteniron-server_hosts',
'molteniron_all',
'mq_containers',
'mq_hosts',
'network_all',
'network_containers',
'network_hosts',
'neutron_agent',
'neutron_agents_container',
'neutron_all',
'neutron_bgp_dragent',
'neutron_dhcp_agent',
'neutron_l3_agent',
'neutron_lbaas_agent',
'neutron_linuxbridge_agent',
'neutron_metadata_agent',
'neutron_metering_agent',
'neutron_openvswitch_agent',
'neutron_sriov_nic_agent',
'neutron_server',
'neutron_server_container',
'nova_all',
'nova_api_metadata',
'nova_api_os_compute',
'nova_api_container',
'nova_api_placement',
'nova_compute',
'nova_compute_container',
'nova_conductor',
'nova_console',
'nova_scheduler',
'opendaylight',
'operator_containers',
'operator_hosts',
'orchestration_all',
'orchestration_containers',
'orchestration_hosts',
'os-infra_containers',
'os-infra_hosts',
'pkg_repo',
'rabbit_mq_container',
'rabbitmq',
'rabbitmq_all',
'remote',
'remote_containers',
'repo-infra_all',
'repo-infra_containers',
'repo-infra_hosts',
'repo_all',
'repo_container',
'rsyslog',
'rsyslog_all',
'rsyslog_container',
'sahara-infra_all',
'sahara-infra_containers',
'sahara-infra_hosts',
'sahara_all',
'sahara_api',
'sahara_container',
'sahara_engine',
'shared-infra_all',
'shared-infra_containers',
'shared-infra_hosts',
'storage-infra_all',
'storage-infra_containers',
'storage-infra_hosts',
'storage_all',
'storage_containers',
'storage_hosts',
'swift-proxy_all',
'swift-proxy_containers',
'swift-proxy_hosts',
'swift-remote_containers',
'swift-remote_hosts',
'swift_acc',
'swift_acc_container',
'swift_all',
'swift_cont',
'swift_cont_container',
'swift_containers',
'swift_hosts',
'swift_obj',
'swift_obj_container',
'swift_proxy',
'swift_proxy_container',
'swift_remote',
'swift_remote_all',
'swift_remote_container',
'tacker_all',
'tacker_container',
'tacker_server',
'unbound',
'unbound_all',
'unbound_container',
'unbound_containers',
'unbound_hosts',
'utility',
'utility_all',
'utility_container'
]
@classmethod
def setUpClass(cls):
cls.inventory = get_inventory()
def test_meta(self):
meta = self.inventory['_meta']
self.assertIsNotNone(meta, "_meta missing from inventory")
self.assertIsInstance(meta, dict, "_meta is not a dict")
def test_hostvars(self):
hostvars = self.inventory['_meta']['hostvars']
self.assertIsNotNone(hostvars, "hostvars missing from _meta")
self.assertIsInstance(hostvars, dict, "hostvars is not a dict")
def test_group_vars_all(self):
group_vars_all = self.inventory['all']
self.assertIsNotNone(group_vars_all,
"group vars all missing from inventory")
self.assertIsInstance(group_vars_all, dict,
"group vars all is not a dict")
the_vars = group_vars_all['vars']
self.assertIsNotNone(the_vars,
"vars missing from group vars all")
self.assertIsInstance(the_vars, dict,
"vars in group vars all is not a dict")
def test_expected_host_groups_present(self):
for group in self.expected_groups:
the_group = self.inventory[group]
self.assertIsNotNone(the_group,
"Required host group: %s is missing "
"from inventory" % group)
self.assertIsInstance(the_group, dict)
if group != 'all':
self.assertIn('hosts', the_group)
self.assertIsInstance(the_group['hosts'], list)
def test_only_expected_host_groups_present(self):
all_keys = list(self.expected_groups)
all_keys.append('_meta')
self.assertEqual(set(all_keys), set(self.inventory.keys()))
def test_configured_groups_have_hosts(self):
config = get_config()
groups = self.inventory.keys()
for group in groups:
if group in config.keys():
self.assertTrue(0 < len(self.inventory[group]['hosts']))
class TestUserConfiguration(unittest.TestCase):
def setUp(self):
self.longMessage = True
self.loaded_user_configuration = fs.load_user_configuration(TARGET_DIR)
def test_loading_user_configuration(self):
"""Test that the user configuration can be loaded"""
self.assertIsInstance(self.loaded_user_configuration, dict)
class TestEnvironments(unittest.TestCase):
def setUp(self):
self.longMessage = True
self.loaded_environment = fs.load_environment(BASE_ENV_DIR, {})
def test_loading_environment(self):
"""Test that the environment can be loaded"""
self.assertIsInstance(self.loaded_environment, dict)
def test_envd_read(self):
"""Test that the env.d contents are inserted into the environment"""
expected_keys = [
'component_skel',
'container_skel',
'physical_skel',
]
for key in expected_keys:
self.assertIn(key, self.loaded_environment)
class TestIps(unittest.TestCase):
def setUp(self):
# Allow custom assertion errors.
self.longMessage = True
self.env = fs.load_environment(BASE_ENV_DIR, {})
@mock.patch('osa_toolkit.filesystem.load_environment')
@mock.patch('osa_toolkit.filesystem.load_user_configuration')
def test_duplicates(self, mock_load_config, mock_load_env):
"""Test that no duplicate IPs are made on any network."""
# Grab our values read from the file system just once.
mock_load_config.return_value = get_config()
mock_load_env.return_value = self.env
mock_open = mock.mock_open()
for i in range(0, 99):
# tearDown is ineffective for this loop, so clean the USED_IPs
# on each run
inventory = None
di.ip.USED_IPS = set()
# Mock out the context manager being used to write files.
# We don't need to hit the file system for this test.
with mock.patch('__main__.open', mock_open):
inventory = get_inventory()
ips = collections.defaultdict(int)
hostvars = inventory['_meta']['hostvars']
for host, var_dict in hostvars.items():
nets = var_dict['container_networks']
for net, vals in nets.items():
if 'address' in vals.keys():
addr = vals['address']
ips[addr] += 1
self.assertEqual(1, ips[addr],
msg="IP %s duplicated." % addr)
def test_empty_ip_queue(self):
q = Queue.Queue()
with self.assertRaises(SystemExit) as context:
# TODO(nrb): import and use ip module directly
di.ip.get_ip_address('test', q)
expectedLog = ("Cannot retrieve requested amount of IP addresses. "
"Increase the test range in your "
"openstack_user_config.yml.")
self.assertEqual(str(context.exception), expectedLog)
def tearDown(self):
# Since the get_ip_address function touches USED_IPS,
# and USED_IPS is currently a global var, make sure we clean it out
di.ip.USED_IPS = set()
class TestConfigCheckBase(unittest.TestCase):
def setUp(self):
self.config_changed = False
self.user_defined_config = get_config()
def delete_config_key(self, user_defined_config, key):
try:
if key in user_defined_config:
del user_defined_config[key]
elif key in user_defined_config['global_overrides']:
del user_defined_config['global_overrides'][key]
else:
raise KeyError("can't find specified key in user config")
finally:
self.write_config()
def add_config_key(self, key, value):
self.user_defined_config[key] = value
self.write_config()
def add_provider_network(self, net_name, cidr):
self.user_defined_config['cidr_networks'][net_name] = cidr
self.write_config()
def delete_provider_network(self, net_name):
del self.user_defined_config['cidr_networks'][net_name]
self.write_config()
def add_provider_network_key(self, net_name, key, value):
pns = self.user_defined_config['global_overrides']['provider_networks']
for net in pns:
if 'ip_from_q' in net['network']:
if net['network']['ip_from_q'] == net_name:
net['network'][key] = value
def delete_provider_network_key(self, net_name, key):
pns = self.user_defined_config['global_overrides']['provider_networks']
for net in pns:
if 'ip_from_q' in net['network']:
if net['network']['ip_from_q'] == net_name:
if key in net['network']:
del net['network'][key]
def write_config(self):
self.config_changed = True
# Save new user_config_file
with open(USER_CONFIG_FILE, 'wb') as f:
f.write(yaml.dump(self.user_defined_config).encode('ascii'))
def restore_config(self):
# get back our initial user config file
self.user_defined_config = get_config()
self.write_config()
def set_new_hostname(self, user_defined_config, group,
old_hostname, new_hostname):
# set a new name for the specified hostname
old_hostname_settings = user_defined_config[group].pop(old_hostname)
user_defined_config[group][new_hostname] = old_hostname_settings
self.write_config()
def set_new_ip(self, user_defined_config, group, hostname, ip):
# Sets an IP address for a specified host.
user_defined_config[group][hostname]['ip'] = ip
self.write_config()
def add_host(self, group, host_name, ip):
self.user_defined_config[group][host_name] = {'ip': ip}
self.write_config()
def tearDown(self):
if self.config_changed:
self.restore_config()
class TestConfigChecks(TestConfigCheckBase):
def test_missing_container_cidr_network(self):
self.delete_provider_network('container')
with self.assertRaises(SystemExit) as context:
get_inventory()
expectedLog = ("No container or management network specified in "
"user config.")
self.assertEqual(str(context.exception), expectedLog)
def test_management_network_malformed(self):
self.delete_provider_network_key('container', 'is_container_address')
self.write_config()
with self.assertRaises(di.ProviderNetworkMisconfiguration) as context:
get_inventory()
expectedLog = ("Provider network with queue 'container' "
"requires 'is_container_address' "
"to be set to True.")
self.assertEqual(str(context.exception), expectedLog)
self.restore_config()
def test_missing_cidr_network_present_in_provider(self):
self.delete_provider_network('storage')
with self.assertRaises(SystemExit) as context:
get_inventory()
expectedLog = "can't find storage in cidr_networks"
self.assertEqual(str(context.exception), expectedLog)
def test_missing_cidr_networks_key(self):
del self.user_defined_config['cidr_networks']
self.write_config()
with self.assertRaises(SystemExit) as context:
get_inventory()
expectedLog = "No container CIDR specified in user config"
self.assertEqual(str(context.exception), expectedLog)
def test_provider_networks_check(self):
# create config file without provider networks
self.delete_config_key(self.user_defined_config, 'provider_networks')
# check if provider networks absence is Caught
with self.assertRaises(SystemExit) as context:
get_inventory()
expectedLog = "provider networks can't be found under global_overrides"
self.assertIn(expectedLog, str(context.exception))
def test_global_overrides_check(self):
# create config file without global_overrides
self.delete_config_key(self.user_defined_config, 'global_overrides')
# check if global_overrides absence is Caught
with self.assertRaises(SystemExit) as context:
get_inventory()
expectedLog = "global_overrides can't be found in user config"
self.assertEqual(str(context.exception), expectedLog)
def test_two_hosts_same_ip(self):
# Use an OrderedDict to be certain our testing order is preserved
# Even with the same hash seed, different OSes get different results,
# eg. local OS X vs gate's Linux
config = collections.OrderedDict()
config['shared-infra_hosts'] = {
'host1': {
'ip': '192.168.1.1'
}
}
config['compute_hosts'] = {
'host2': {
'ip': '192.168.1.1'
}
}
with self.assertRaises(di.MultipleHostsWithOneIPError) as context:
di._check_same_ip_to_multiple_host(config)
self.assertEqual(context.exception.ip, '192.168.1.1')
self.assertEqual(context.exception.assigned_host, 'host1')
self.assertEqual(context.exception.new_host, 'host2')
def test_two_hosts_same_ip_externally(self):
self.set_new_hostname(self.user_defined_config, "haproxy_hosts",
"aio1", "hap")
with self.assertRaises(di.MultipleHostsWithOneIPError) as context:
get_inventory()
expectedLog = ("Both host:aio1 and host:hap have "
"address:172.29.236.100 assigned. Cannot "
"assign same ip to both hosts")
self.assertEqual(str(context.exception), expectedLog)
def test_one_host_two_ips_externally(self):
# haproxy chosen because it was last in the config file as of
# writing
self.set_new_ip(self.user_defined_config, 'haproxy_hosts', 'aio1',
'172.29.236.101')
with self.assertRaises(di.MultipleIpForHostError) as context:
get_inventory()
expectedLog = ("Host aio1 has both 172.29.236.100 and 172.29.236.101 "
"assigned")
self.assertEqual(str(context.exception), expectedLog)
def test_two_ips(self):
# Use an OrderedDict to be certain our testing order is preserved
# Even with the same hash seed, different OSes get different results,
# eg. local OS X vs gate's Linux
config = collections.OrderedDict()
config['shared-infra_hosts'] = {
'host1': {
'ip': '192.168.1.1'
}
}
config['compute_hosts'] = {
'host1': {
'ip': '192.168.1.2'
}
}
with self.assertRaises(di.MultipleIpForHostError) as context:
di._check_multiple_ips_to_host(config)
self.assertEqual(context.exception.current_ip, '192.168.1.1')
self.assertEqual(context.exception.new_ip, '192.168.1.2')
self.assertEqual(context.exception.hostname, 'host1')
def test_correct_hostname_ip_map(self):
config = {
'shared-infra_hosts': {
'host1': {
'ip': '192.168.1.1'
}
},
'compute_hosts': {
'host2': {
'ip': '192.168.1.2'
}
},
}
ret = di._check_multiple_ips_to_host(config)
self.assertTrue(ret)
class TestStaticRouteConfig(TestConfigCheckBase):
def setUp(self):
super(TestStaticRouteConfig, self).setUp()
self.expectedMsg = ("Static route provider network with queue "
"'container' needs both 'cidr' and 'gateway' "
"values.")
def add_static_route(self, q_name, route_dict):
"""Adds a static route to a provider network."""
pn = self.user_defined_config['global_overrides']['provider_networks']
for net in pn:
net_dict = net['network']
q = net_dict.get('ip_from_q', None)
if q == q_name:
net_dict['static_routes'] = [route_dict]
self.write_config()
def test_setting_static_route(self):
route_dict = {'cidr': '10.176.0.0/12',
'gateway': '172.29.248.1'}
self.add_static_route('container', route_dict)
inventory = get_inventory()
# Use aio1 and 'container_address' since they're known keys.
hostvars = inventory['_meta']['hostvars']['aio1']
cont_add = hostvars['container_networks']['container_address']
self.assertIn('static_routes', cont_add)
first_route = cont_add['static_routes'][0]
self.assertIn('cidr', first_route)
self.assertIn('gateway', first_route)
def test_setting_bad_static_route_only_cidr(self):
route_dict = {'cidr': '10.176.0.0/12'}
self.add_static_route('container', route_dict)
with self.assertRaises(di.MissingStaticRouteInfo) as context:
get_inventory()
exception = context.exception
self.assertEqual(str(exception), self.expectedMsg)
def test_setting_bad_static_route_only_gateway(self):
route_dict = {'gateway': '172.29.248.1'}
self.add_static_route('container', route_dict)
with self.assertRaises(di.MissingStaticRouteInfo) as context:
get_inventory()
exception = context.exception
self.assertEqual(exception.message, self.expectedMsg)
def test_setting_bad_gateway_value(self):
route_dict = {'cidr': '10.176.0.0/12',
'gateway': None}
self.add_static_route('container', route_dict)
with self.assertRaises(di.MissingStaticRouteInfo) as context:
get_inventory()
exception = context.exception
self.assertEqual(exception.message, self.expectedMsg)
def test_setting_bad_cidr_value(self):
route_dict = {'cidr': None,
'gateway': '172.29.248.1'}
self.add_static_route('container', route_dict)
with self.assertRaises(di.MissingStaticRouteInfo) as context:
get_inventory()
exception = context.exception
self.assertEqual(exception.message, self.expectedMsg)
def test_setting_bad_cidr_gateway_value(self):
route_dict = {'cidr': None,
'gateway': None}
self.add_static_route('container', route_dict)
with self.assertRaises(di.MissingStaticRouteInfo) as context:
get_inventory()
exception = context.exception
self.assertEqual(exception.message, self.expectedMsg)
class TestGlobalOverridesConfigDeletion(TestConfigCheckBase):
def setUp(self):
super(TestGlobalOverridesConfigDeletion, self).setUp()
self.inventory = get_inventory()
def add_global_override(self, var_name, var_value):
"""Adds an arbitrary name and value to the global_overrides dict."""
overrides = self.user_defined_config['global_overrides']
overrides[var_name] = var_value
def remove_global_override(self, var_name):
"""Removes target key from the global_overrides dict."""
overrides = self.user_defined_config['global_overrides']
del overrides[var_name]
def test_global_overrides_delete_when_merge(self):
"""Vars removed from global overrides are removed from inventory"""
self.add_global_override('foo', 'bar')
di._parse_global_variables({}, self.inventory,
self.user_defined_config)
self.remove_global_override('foo')
di._parse_global_variables({}, self.inventory,
self.user_defined_config)
self.assertNotIn('foo', self.inventory['all']['vars'],
"foo var not removed from group_vars_all")
def test_global_overrides_merge(self):
self.add_global_override('foo', 'bar')
di._parse_global_variables({}, self.inventory,
self.user_defined_config)
self.assertEqual('bar', self.inventory['all']['vars']['foo'])
def test_container_cidr_key_retained(self):
user_cidr = self.user_defined_config['cidr_networks']['container']
di._parse_global_variables(user_cidr, self.inventory,
self.user_defined_config)
self.assertIn('container_cidr', self.inventory['all']['vars'])
self.assertEqual(self.inventory['all']['vars']['container_cidr'],
user_cidr)
def test_only_old_vars_deleted(self):
self.inventory['all']['vars']['foo'] = 'bar'
di._parse_global_variables('', self.inventory,
self.user_defined_config)
self.assertNotIn('foo', self.inventory['all']['vars'])
def test_empty_vars(self):
del self.inventory['all']
di._parse_global_variables('', self.inventory,
self.user_defined_config)
self.assertIn('container_cidr', self.inventory['all']['vars'])
for key in self.user_defined_config['global_overrides']:
self.assertIn(key, self.inventory['all']['vars'])
class TestEnsureInventoryUptoDate(unittest.TestCase):
def setUp(self):
self.env = fs.load_environment(BASE_ENV_DIR, {})
# Copy because we manipulate the structure in each test;
# not copying would modify the global var in the target code
self.inv = copy.deepcopy(di.INVENTORY_SKEL)
# Since we're not running skel_setup, add necessary keys
self.host_vars = self.inv['_meta']['hostvars']
# The _ensure_inventory_uptodate function depends on values inserted
# by the skel_setup function
di.skel_setup(self.env, self.inv)
def test_missing_required_host_vars(self):
self.host_vars['host1'] = {}
di._ensure_inventory_uptodate(self.inv, self.env['container_skel'])
for required_key in di.REQUIRED_HOSTVARS:
self.assertIn(required_key, self.host_vars['host1'])
def test_missing_container_name(self):
self.host_vars['host1'] = {}
di._ensure_inventory_uptodate(self.inv, self.env['container_skel'])
self.assertIn('container_name', self.host_vars['host1'])
self.assertEqual(self.host_vars['host1']['container_name'], 'host1')
def test_inserting_container_networks_is_dict(self):
self.host_vars['host1'] = {}
di._ensure_inventory_uptodate(self.inv, self.env['container_skel'])
self.assertIsInstance(self.host_vars['host1']['container_networks'],
dict)
def test_populating_inventory_info(self):
skel = self.env['container_skel']
di._ensure_inventory_uptodate(self.inv, skel)
for container_type, type_vars in skel.items():
hosts = self.inv[container_type]['hosts']
if hosts:
for host in hosts:
host_var_entries = self.inv['_meta']['hostvars'][host]
if 'properties' in type_vars:
self.assertEqual(host_var_entries['properties'],
type_vars['properties'])
def tearDown(self):
self.env = None
self.host_vars = None
self.inv = None
class OverridingEnvBase(unittest.TestCase):
def setUp(self):
self.base_env = fs.load_environment(BASE_ENV_DIR, {})
# Use the cinder configuration as our sample for override testing
with open(path.join(BASE_ENV_DIR, 'env.d', 'cinder.yml'), 'r') as f:
self.cinder_config = yaml.safe_load(f.read())
self.override_path = path.join(TARGET_DIR, 'env.d')
os.mkdir(self.override_path)
def write_override_env(self):
with open(path.join(self.override_path, 'cinder.yml'), 'w') as f:
f.write(yaml.safe_dump(self.cinder_config))
def tearDown(self):
os.remove(path.join(self.override_path, 'cinder.yml'))
os.rmdir(self.override_path)
class TestOverridingEnvVars(OverridingEnvBase):
def test_cinder_metal_override(self):
vol = self.cinder_config['container_skel']['cinder_volumes_container']
vol['properties']['is_metal'] = False
self.write_override_env()
fs.load_environment(TARGET_DIR, self.base_env)
test_vol = self.base_env['container_skel']['cinder_volumes_container']
self.assertFalse(test_vol['properties']['is_metal'])
def test_deleting_elements(self):
# Leave only the 'properties' dictionary attached to simulate writing
# a partial override file
vol = self.cinder_config['container_skel']['cinder_volumes_container']
keys = vol.keys()
to_delete = []
for key in vol.keys():
if not key == 'properties':
to_delete.append(key)
for key in to_delete:
del vol[key]
self.write_override_env()
fs.load_environment(TARGET_DIR, self.base_env)
test_vol = self.base_env['container_skel']['cinder_volumes_container']
self.assertIn('belongs_to', test_vol)
def test_adding_new_keys(self):
vol = self.cinder_config['container_skel']['cinder_volumes_container']
vol['a_new_key'] = 'Added'
self.write_override_env()
fs.load_environment(TARGET_DIR, self.base_env)
test_vol = self.base_env['container_skel']['cinder_volumes_container']
self.assertIn('a_new_key', test_vol)
self.assertEqual(test_vol['a_new_key'], 'Added')
def test_emptying_dictionaries(self):
self.cinder_config['container_skel']['cinder_volumes_container'] = {}
self.write_override_env()
fs.load_environment(TARGET_DIR, self.base_env)
test_vol = self.base_env['container_skel']['cinder_volumes_container']
self.assertNotIn('belongs_to', test_vol)
def test_emptying_lists(self):
vol = self.cinder_config['container_skel']['cinder_volumes_container']
vol['belongs_to'] = []
self.write_override_env()
fs.load_environment(TARGET_DIR, self.base_env)
test_vol = self.base_env['container_skel']['cinder_volumes_container']
self.assertEqual(test_vol['belongs_to'], [])
class TestOverridingEnvIntegration(OverridingEnvBase):
def setUp(self):
super(TestOverridingEnvIntegration, self).setUp()
self.user_defined_config = get_config()
# Inventory is necessary since keys are assumed present
self.inv, path = fs.load_inventory(TARGET_DIR, di.INVENTORY_SKEL)
def skel_setup(self):
self.environment = fs.load_environment(TARGET_DIR, self.base_env)
di.skel_setup(self.environment, self.inv)
di.skel_load(
self.environment.get('physical_skel'),
self.inv
)
def test_emptying_container_integration(self):
self.cinder_config = {}
self.cinder_config['container_skel'] = {'cinder_volumes_container': {}}
self.write_override_env()
self.skel_setup()
di.container_skel_load(
self.environment.get('container_skel'),
self.inv,
self.user_defined_config
)
test_vol = self.base_env['container_skel']['cinder_volumes_container']
self.assertNotIn('belongs_to', test_vol)
self.assertNotIn('contains', test_vol)
def test_empty_contains(self):
vol = self.cinder_config['container_skel']['cinder_volumes_container']
vol['contains'] = []
self.write_override_env()
self.skel_setup()
di.container_skel_load(
self.environment.get('container_skel'),
self.inv,
self.user_defined_config
)
test_vol = self.base_env['container_skel']['cinder_volumes_container']
self.assertEqual(test_vol['contains'], [])
def test_empty_belongs_to(self):
vol = self.cinder_config['container_skel']['cinder_volumes_container']
vol['belongs_to'] = []
self.write_override_env()
self.skel_setup()
di.container_skel_load(
self.environment.get('container_skel'),
self.inv,
self.user_defined_config
)
test_vol = self.base_env['container_skel']['cinder_volumes_container']
self.assertEqual(test_vol['belongs_to'], [])
def tearDown(self):
super(TestOverridingEnvIntegration, self).tearDown()
self.user_defined_config = None
self.inv = None
class TestSetUsedIPS(unittest.TestCase):
def setUp(self):
# Clean up the used ips in case other tests didn't.
di.ip.USED_IPS = set()
# Create a fake inventory just for this test.
self.inventory = {'_meta': {'hostvars': {
'host1': {'container_networks': {
'net': {'address': '172.12.1.1'}
}},
'host2': {'container_networks': {
'net': {'address': '172.12.1.2'}
}},
}}}
def test_adding_inventory_used_ips(self):
config = {'used_ips': None}
# TODO(nrb): This is a smell, needs to set more directly
di.ip.set_used_ips(config, self.inventory)
self.assertEqual(len(di.ip.USED_IPS), 2)
self.assertIn('172.12.1.1', di.ip.USED_IPS)
self.assertIn('172.12.1.2', di.ip.USED_IPS)
def tearDown(self):
di.ip.USED_IPS = set()
class TestConfigCheckFunctional(TestConfigCheckBase):
def duplicate_ip(self):
ip = self.user_defined_config['log_hosts']['aio1']
self.user_defined_config['log_hosts']['bogus'] = ip
def test_checking_good_config(self):
output = di.main(config=TARGET_DIR, check=True,
environment=BASE_ENV_DIR)
self.assertEqual(output, 'Configuration ok!')
def test_duplicated_ip(self):
self.duplicate_ip()
self.write_config()
with self.assertRaises(di.MultipleHostsWithOneIPError) as context:
di.main(config=TARGET_DIR, check=True, environment=BASE_ENV_DIR)
self.assertEqual(context.exception.ip, '172.29.236.100')
class TestNetworkEntry(unittest.TestCase):
def test_all_args_filled(self):
entry = di.network_entry(True, 'eth1', 'br-mgmt', 'my_type', '1700')
self.assertNotIn('interface', entry.keys())
self.assertEqual(entry['bridge'], 'br-mgmt')
self.assertEqual(entry['type'], 'my_type')
self.assertEqual(entry['mtu'], '1700')
def test_container_dict(self):
entry = di.network_entry(False, 'eth1', 'br-mgmt', 'my_type', '1700')
self.assertEqual(entry['interface'], 'eth1')
class TestDebugLogging(unittest.TestCase):
@mock.patch('osa_toolkit.generate.logging')
@mock.patch('osa_toolkit.generate.logger')
def test_logging_enabled(self, mock_logger, mock_logging):
# Shadow the real value so tests don't complain about it
mock_logging.DEBUG = 10
get_inventory(extra_args={"debug": True})
self.assertTrue(mock_logging.basicConfig.called)
self.assertTrue(mock_logger.info.called)
self.assertTrue(mock_logger.debug.called)
@mock.patch('osa_toolkit.generate.logging')
@mock.patch('osa_toolkit.generate.logger')
def test_logging_disabled(self, mock_logger, mock_logging):
get_inventory(extra_args={"debug": False})
self.assertFalse(mock_logging.basicConfig.called)
# Even though logging is disabled, we still call these
# all over the place; they just choose not to do anything.
# NOTE: No info messages are published when debug is False
self.assertTrue(mock_logger.debug.called)
class TestLxcHosts(TestConfigCheckBase):
def test_lxc_hosts_group_present(self):
inventory = get_inventory()
self.assertIn('lxc_hosts', inventory)
def test_lxc_hosts_only_inserted_once(self):
inventory = get_inventory()
self.assertEqual(1, len(inventory['lxc_hosts']['hosts']))
def test_lxc_hosts_members(self):
self.add_host('shared-infra_hosts', 'aio2', '172.29.236.101')
inventory = get_inventory()
self.assertIn('aio2', inventory['lxc_hosts']['hosts'])
self.assertIn('aio1', inventory['lxc_hosts']['hosts'])
def test_lxc_hosts_in_config_raises_error(self):
self.add_config_key('lxc_hosts', {})
with self.assertRaises(di.LxcHostsDefined):
get_inventory()
def test_host_without_containers(self):
self.add_host('compute_hosts', 'compute1', '172.29.236.102')
inventory = get_inventory()
self.assertNotIn('compute1', inventory['lxc_hosts']['hosts'])
def test_cleaning_bad_hosts(self):
self.add_host('compute_hosts', 'compute1', '172.29.236.102')
inventory = get_inventory()
# insert compute1 into lxc_hosts, which mimicks bug behavior
inventory['lxc_hosts']['hosts'].append('compute1')
faked_path = INV_DIR
with mock.patch('osa_toolkit.filesystem.load_inventory') as inv_mock:
inv_mock.return_value = (inventory, faked_path)
new_inventory = get_inventory()
# host should no longer be in lxc_hosts
self.assertNotIn('compute1', new_inventory['lxc_hosts']['hosts'])
def test_emptying_lxc_hosts(self):
"""If lxc_hosts is deleted between runs, it should re-populate"""
inventory = get_inventory()
original_lxc_hosts = inventory.pop('lxc_hosts')
self.assertNotIn('lxc_hosts', inventory.keys())
faked_path = INV_DIR
with mock.patch('osa_toolkit.filesystem.load_inventory') as inv_mock:
inv_mock.return_value = (inventory, faked_path)
new_inventory = get_inventory()
self.assertEqual(original_lxc_hosts, new_inventory['lxc_hosts'])
class TestConfigMatchesEnvironment(unittest.TestCase):
def setUp(self):
self.env = fs.load_environment(BASE_ENV_DIR, {})
def test_matching_keys(self):
config = get_config()
result = di._check_all_conf_groups_present(config, self.env)
self.assertTrue(result)
def test_failed_match(self):
bad_config = get_config()
bad_config['bogus_key'] = []
result = di._check_all_conf_groups_present(bad_config, self.env)
self.assertFalse(result)
def test_extra_config_key_warning(self):
bad_config = get_config()
bad_config['bogus_key'] = []
with warnings.catch_warnings(record=True) as wl:
di._check_all_conf_groups_present(bad_config, self.env)
self.assertEqual(1, len(wl))
self.assertIn('bogus_key', str(wl[0].message))
def test_multiple_extra_keys(self):
bad_config = get_config()
bad_config['bogus_key1'] = []
bad_config['bogus_key2'] = []
with warnings.catch_warnings(record=True) as wl:
di._check_all_conf_groups_present(bad_config, self.env)
self.assertEqual(2, len(wl))
warn_msgs = [str(warn.message) for warn in wl]
warn_msgs.sort()
self.assertIn('bogus_key1', warn_msgs[0])
self.assertIn('bogus_key2', warn_msgs[1])
def test_confirm_exclusions(self):
"""Ensure the excluded keys in the function are present."""
config = get_config()
excluded_keys = ('global_overrides', 'cidr_networks', 'used_ips')
for key in excluded_keys:
config[key] = 'sentinel value'
with warnings.catch_warnings(record=True) as wl:
di._check_all_conf_groups_present(config, self.env)
self.assertEqual(0, len(wl))
for key in excluded_keys:
self.assertIn(key, config.keys())
class TestInventoryGroupConstraints(unittest.TestCase):
def setUp(self):
self.env = fs.load_environment(BASE_ENV_DIR, {})
def test_group_with_hosts_dont_have_children(self):
"""Require that groups have children groups or hosts, not both."""
inventory = get_inventory()
# This should only work on groups, but stuff like '_meta' and 'all'
# are in here, too.
for key, values in inventory.items():
# The keys for children/hosts can exist, the important part is being empty lists.
has_children = bool(inventory.get('children'))
has_hosts = bool(inventory.get('hosts'))
self.assertFalse(has_children and has_hosts)
def _create_bad_env(self, env):
# This environment setup is used because it was reported with
# bug #1646136
override = """
physical_skel:
local-compute_containers:
belongs_to:
- compute_containers
local-compute_hosts:
belongs_to:
- compute_hosts
rbd-compute_containers:
belongs_to:
- compute_containers
rbd-compute_hosts:
belongs_to:
- compute_hosts
"""
bad_env = yaml.safe_load(override)
# This is essentially what load_environment does, after all the file
# system walking
dictutils.merge_dict(env, bad_env)
return env
def test_group_with_hosts_and_children_fails(self):
"""Integration test making sure the whole script fails."""
env = self._create_bad_env(self.env)
config = get_config()
kwargs = {
'load_environment': mock.DEFAULT,
'load_user_configuration': mock.DEFAULT
}
with mock.patch.multiple('osa_toolkit.filesystem', **kwargs) as mocks:
mocks['load_environment'].return_value = env
mocks['load_user_configuration'].return_value = config
with self.assertRaises(di.GroupConflict) as context:
get_inventory()
def test_group_validation_unit(self):
env = self._create_bad_env(self.env)
config = get_config()
with self.assertRaises(di.GroupConflict):
di._check_group_branches(config, env['physical_skel'])
def test_group_validation_no_config(self):
result = di._check_group_branches(None, self.env)
self.assertTrue(result)
def test_group_validation_passes_defaults(self):
config = get_config()
result = di._check_group_branches(config, self.env['physical_skel'])
self.assertTrue(result)
class TestL3ProviderNetworkConfig(TestConfigCheckBase):
def setUp(self):
super(TestL3ProviderNetworkConfig, self).setUp()
self.delete_provider_network('container')
self.add_provider_network('pod1_container', '172.29.236.0/22')
self.add_provider_network_key('container', 'ip_from_q',
'pod1_container')
self.add_provider_network_key('pod1_container', 'address_prefix',
'management')
self.add_provider_network_key('pod1_container', 'reference_group',
'pod1_hosts')
self.add_config_key('pod1_hosts', {})
self.add_host('pod1_hosts', 'aio2', '172.29.236.101')
self.add_host('compute_hosts', 'aio2', '172.29.236.101')
self.write_config()
self.inventory = get_inventory()
def test_address_prefix_name_applied(self):
aio2_host_vars = self.inventory['_meta']['hostvars']['aio2']
aio2_container_networks = aio2_host_vars['container_networks']
self.assertIsInstance(aio2_container_networks['management_address'],
dict)
def test_host_outside_reference_group_excluded(self):
aio1_host_vars = self.inventory['_meta']['hostvars']['aio1']
aio1_container_networks = aio1_host_vars['container_networks']
self.assertNotIn('management_address', aio1_container_networks)
if __name__ == '__main__':
unittest.main(catchbreak=True)
View Git Blame Copy Permalink