# 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 mock import unittest from collections import OrderedDict import sys path = "./contrib/inventory_builder/" if path not in sys.path: sys.path.append(path) import inventory 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.KargoInventory() def test_get_ip_from_opts(self): optstring = "ansible_host=10.90.3.2 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.assertRaisesRegexp(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.assertTrue(group in self.inv.config.sections()) 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.assertRaisesRegexp(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')]) 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')]) self.inv.config['all'] = 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'), ('node2', 'ansible_host=10.90.0.3 ip=10.90.0.3')]) self.inv.config['all'] = 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'), ('node2', 'ansible_host=10.90.0.3 ip=10.90.0.3')]) self.inv.config['all'] = existing_hosts expected = OrderedDict([ ('node2', 'ansible_host=10.90.0.3 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'), ('node2', 'ansible_host=10.90.0.3 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'), ('node2', 'ansible_host=10.90.0.3 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'), ('node2', 'ansible_host=10.90.0.3 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'), ('node2', 'ansible_host=10.90.0.3 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')]) existing_hosts = OrderedDict([ ('node1', 'ansible_host=10.90.0.2 ip=10.90.0.2'), ('node2', 'ansible_host=10.90.0.3 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'), ('node2', 'ansible_host=10.90.0.3 ip=10.90.0.3')]) self.assertRaisesRegexp(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'), ('node2', 'ansible_host=10.90.0.3 ip=10.90.0.3'), ('doesnotbelong2', 'whateveropts=ilike')]) self.inv.config['all'] = existing_hosts self.inv.purge_invalid_hosts(proper_hostnames) self.assertTrue(bad_host not in self.inv.config['all'].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.config[group].get(host), opts) def test_set_kube_master(self): group = 'kube-master' host = 'node1' self.inv.set_kube_master([host]) self.assertTrue(host in self.inv.config[group]) def test_set_all(self): group = 'all' hosts = OrderedDict([ ('node1', 'opt1'), ('node2', 'opt2')]) self.inv.set_all(hosts) for host, opt in hosts.items(): self.assertEqual(self.inv.config[group].get(host), opt) def test_set_k8s_cluster(self): group = 'k8s-cluster:children' expected_hosts = ['kube-node', 'kube-master'] self.inv.set_k8s_cluster() for host in expected_hosts: self.assertTrue(host in self.inv.config[group]) def test_set_kube_node(self): group = 'kube-node' host = 'node1' self.inv.set_kube_node([host]) self.assertTrue(host in self.inv.config[group]) def test_set_etcd(self): group = 'etcd' host = 'node1' self.inv.set_etcd([host]) self.assertTrue(host in self.inv.config[group])