# Copyright 2016 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 inventory import unittest from unittest import mock from collections import OrderedDict import sys path = "./contrib/inventory_builder/" if path not in sys.path: sys.path.append(path) import inventory # noqa class TestInventory(unittest.TestCase): @mock.patch('inventory.sys') def setUp(self, sys_mock): sys_mock.exit = mock.Mock() super(TestInventory, self).setUp() self.data = ['10.90.3.2', '10.90.3.3', '10.90.3.4'] self.inv = inventory.KubesprayInventory() def test_get_ip_from_opts(self): optstring = {'ansible_host': '10.90.3.2', 'ip': '10.90.3.2', 'access_ip': '10.90.3.2'} expected = "10.90.3.2" result = self.inv.get_ip_from_opts(optstring) self.assertEqual(expected, result) def test_get_ip_from_opts_invalid(self): optstring = "notanaddr=value something random!chars:D" self.assertRaisesRegex(ValueError, "IP parameter not found", self.inv.get_ip_from_opts, optstring) def test_ensure_required_groups(self): groups = ['group1', 'group2'] self.inv.ensure_required_groups(groups) for group in groups: self.assertIn(group, self.inv.yaml_config['all']['children']) def test_get_host_id(self): hostnames = ['node99', 'no99de01', '01node01', 'node1.domain', 'node3.xyz123.aaa'] expected = [99, 1, 1, 1, 3] for hostname, expected in zip(hostnames, expected): result = self.inv.get_host_id(hostname) self.assertEqual(expected, result) def test_get_host_id_invalid(self): bad_hostnames = ['node', 'no99de', '01node', 'node.111111'] for hostname in bad_hostnames: self.assertRaisesRegex(ValueError, "Host name must end in an", self.inv.get_host_id, hostname) def test_build_hostnames_add_one(self): changed_hosts = ['10.90.0.2'] expected = OrderedDict([('node1', {'ansible_host': '10.90.0.2', 'ip': '10.90.0.2', 'access_ip': '10.90.0.2'})]) result = self.inv.build_hostnames(changed_hosts) self.assertEqual(expected, result) def test_build_hostnames_add_duplicate(self): changed_hosts = ['10.90.0.2'] expected = OrderedDict([('node1', {'ansible_host': '10.90.0.2', 'ip': '10.90.0.2', 'access_ip': '10.90.0.2'})]) self.inv.yaml_config['all']['hosts'] = expected result = self.inv.build_hostnames(changed_hosts) self.assertEqual(expected, result) def test_build_hostnames_add_two(self): changed_hosts = ['10.90.0.2', '10.90.0.3'] expected = OrderedDict([ ('node1', {'ansible_host': '10.90.0.2', 'ip': '10.90.0.2', 'access_ip': '10.90.0.2'}), ('node2', {'ansible_host': '10.90.0.3', 'ip': '10.90.0.3', 'access_ip': '10.90.0.3'})]) self.inv.yaml_config['all']['hosts'] = OrderedDict() result = self.inv.build_hostnames(changed_hosts) self.assertEqual(expected, result) def test_build_hostnames_delete_first(self): changed_hosts = ['-10.90.0.2'] existing_hosts = OrderedDict([ ('node1', {'ansible_host': '10.90.0.2', 'ip': '10.90.0.2', 'access_ip': '10.90.0.2'}), ('node2', {'ansible_host': '10.90.0.3', 'ip': '10.90.0.3', 'access_ip': '10.90.0.3'})]) self.inv.yaml_config['all']['hosts'] = existing_hosts expected = OrderedDict([ ('node2', {'ansible_host': '10.90.0.3', 'ip': '10.90.0.3', 'access_ip': '10.90.0.3'})]) result = self.inv.build_hostnames(changed_hosts) self.assertEqual(expected, result) def test_exists_hostname_positive(self): hostname = 'node1' expected = True existing_hosts = OrderedDict([ ('node1', {'ansible_host': '10.90.0.2', 'ip': '10.90.0.2', 'access_ip': '10.90.0.2'}), ('node2', {'ansible_host': '10.90.0.3', 'ip': '10.90.0.3', 'access_ip': '10.90.0.3'})]) result = self.inv.exists_hostname(existing_hosts, hostname) self.assertEqual(expected, result) def test_exists_hostname_negative(self): hostname = 'node99' expected = False existing_hosts = OrderedDict([ ('node1', {'ansible_host': '10.90.0.2', 'ip': '10.90.0.2', 'access_ip': '10.90.0.2'}), ('node2', {'ansible_host': '10.90.0.3', 'ip': '10.90.0.3', 'access_ip': '10.90.0.3'})]) result = self.inv.exists_hostname(existing_hosts, hostname) self.assertEqual(expected, result) def test_exists_ip_positive(self): ip = '10.90.0.2' expected = True existing_hosts = OrderedDict([ ('node1', {'ansible_host': '10.90.0.2', 'ip': '10.90.0.2', 'access_ip': '10.90.0.2'}), ('node2', {'ansible_host': '10.90.0.3', 'ip': '10.90.0.3', 'access_ip': '10.90.0.3'})]) result = self.inv.exists_ip(existing_hosts, ip) self.assertEqual(expected, result) def test_exists_ip_negative(self): ip = '10.90.0.200' expected = False existing_hosts = OrderedDict([ ('node1', {'ansible_host': '10.90.0.2', 'ip': '10.90.0.2', 'access_ip': '10.90.0.2'}), ('node2', {'ansible_host': '10.90.0.3', 'ip': '10.90.0.3', 'access_ip': '10.90.0.3'})]) result = self.inv.exists_ip(existing_hosts, ip) self.assertEqual(expected, result) def test_delete_host_by_ip_positive(self): ip = '10.90.0.2' expected = OrderedDict([ ('node2', {'ansible_host': '10.90.0.3', 'ip': '10.90.0.3', 'access_ip': '10.90.0.3'})]) existing_hosts = OrderedDict([ ('node1', {'ansible_host': '10.90.0.2', 'ip': '10.90.0.2', 'access_ip': '10.90.0.2'}), ('node2', {'ansible_host': '10.90.0.3', 'ip': '10.90.0.3', 'access_ip': '10.90.0.3'})]) self.inv.delete_host_by_ip(existing_hosts, ip) self.assertEqual(expected, existing_hosts) def test_delete_host_by_ip_negative(self): ip = '10.90.0.200' existing_hosts = OrderedDict([ ('node1', {'ansible_host': '10.90.0.2', 'ip': '10.90.0.2', 'access_ip': '10.90.0.2'}), ('node2', {'ansible_host': '10.90.0.3', 'ip': '10.90.0.3', 'access_ip': '10.90.0.3'})]) self.assertRaisesRegex(ValueError, "Unable to find host", self.inv.delete_host_by_ip, existing_hosts, ip) def test_purge_invalid_hosts(self): proper_hostnames = ['node1', 'node2'] bad_host = 'doesnotbelong2' existing_hosts = OrderedDict([ ('node1', {'ansible_host': '10.90.0.2', 'ip': '10.90.0.2', 'access_ip': '10.90.0.2'}), ('node2', {'ansible_host': '10.90.0.3', 'ip': '10.90.0.3', 'access_ip': '10.90.0.3'}), ('doesnotbelong2', {'whateveropts=ilike'})]) self.inv.yaml_config['all']['hosts'] = existing_hosts self.inv.purge_invalid_hosts(proper_hostnames) self.assertNotIn( bad_host, self.inv.yaml_config['all']['hosts'].keys()) def test_add_host_to_group(self): group = 'etcd' host = 'node1' opts = {'ip': '10.90.0.2'} self.inv.add_host_to_group(group, host, opts) self.assertEqual( self.inv.yaml_config['all']['children'][group]['hosts'].get(host), None) def test_set_kube_control_plane(self): group = 'kube_control_plane' host = 'node1' self.inv.set_kube_control_plane([host]) self.assertIn( host, self.inv.yaml_config['all']['children'][group]['hosts']) def test_set_all(self): hosts = OrderedDict([ ('node1', 'opt1'), ('node2', 'opt2')]) self.inv.set_all(hosts) for host, opt in hosts.items(): self.assertEqual( self.inv.yaml_config['all']['hosts'].get(host), opt) def test_set_k8s_cluster(self): group = 'k8s_cluster' expected_hosts = ['kube_node', 'kube_control_plane'] self.inv.set_k8s_cluster() for host in expected_hosts: self.assertIn( host, self.inv.yaml_config['all']['children'][group]['children']) def test_set_kube_node(self): group = 'kube_node' host = 'node1' self.inv.set_kube_node([host]) self.assertIn( host, self.inv.yaml_config['all']['children'][group]['hosts']) def test_set_etcd(self): group = 'etcd' host = 'node1' self.inv.set_etcd([host]) self.assertIn( host, self.inv.yaml_config['all']['children'][group]['hosts']) def test_scale_scenario_one(self): num_nodes = 50 hosts = OrderedDict() for hostid in range(1, num_nodes+1): hosts["node" + str(hostid)] = "" self.inv.set_all(hosts) self.inv.set_etcd(list(hosts.keys())[0:3]) self.inv.set_kube_control_plane(list(hosts.keys())[0:2]) self.inv.set_kube_node(hosts.keys()) for h in range(3): self.assertFalse( list(hosts.keys())[h] in self.inv.yaml_config['all']['children']['kube_node']['hosts']) def test_scale_scenario_two(self): num_nodes = 500 hosts = OrderedDict() for hostid in range(1, num_nodes+1): hosts["node" + str(hostid)] = "" self.inv.set_all(hosts) self.inv.set_etcd(list(hosts.keys())[0:3]) self.inv.set_kube_control_plane(list(hosts.keys())[3:5]) self.inv.set_kube_node(hosts.keys()) for h in range(5): self.assertFalse( list(hosts.keys())[h] in self.inv.yaml_config['all']['children']['kube_node']['hosts']) def test_range2ips_range(self): changed_hosts = ['10.90.0.2', '10.90.0.4-10.90.0.6', '10.90.0.8'] expected = ['10.90.0.2', '10.90.0.4', '10.90.0.5', '10.90.0.6', '10.90.0.8'] result = self.inv.range2ips(changed_hosts) self.assertEqual(expected, result) def test_range2ips_incorrect_range(self): host_range = ['10.90.0.4-a.9b.c.e'] self.assertRaisesRegex(Exception, "Range of ip_addresses isn't valid", self.inv.range2ips, host_range) def test_build_hostnames_different_ips_add_one(self): changed_hosts = ['10.90.0.2,192.168.0.2'] expected = OrderedDict([('node1', {'ansible_host': '192.168.0.2', 'ip': '10.90.0.2', 'access_ip': '192.168.0.2'})]) result = self.inv.build_hostnames(changed_hosts) self.assertEqual(expected, result) def test_build_hostnames_different_ips_add_duplicate(self): changed_hosts = ['10.90.0.2,192.168.0.2'] expected = OrderedDict([('node1', {'ansible_host': '192.168.0.2', 'ip': '10.90.0.2', 'access_ip': '192.168.0.2'})]) self.inv.yaml_config['all']['hosts'] = expected result = self.inv.build_hostnames(changed_hosts) self.assertEqual(expected, result) def test_build_hostnames_different_ips_add_two(self): changed_hosts = ['10.90.0.2,192.168.0.2', '10.90.0.3,192.168.0.3'] expected = OrderedDict([ ('node1', {'ansible_host': '192.168.0.2', 'ip': '10.90.0.2', 'access_ip': '192.168.0.2'}), ('node2', {'ansible_host': '192.168.0.3', 'ip': '10.90.0.3', 'access_ip': '192.168.0.3'})]) self.inv.yaml_config['all']['hosts'] = OrderedDict() result = self.inv.build_hostnames(changed_hosts) self.assertEqual(expected, result)