| 195e6d61 |
import os
import unittest |
| ddd58226 |
from nose.tools import raises |
| 195e6d61 |
|
| ddd58226 |
from ansible import errors |
| 195e6d61 |
from ansible.inventory import Inventory
class TestInventory(unittest.TestCase):
def setUp(self):
self.cwd = os.getcwd()
self.test_dir = os.path.join(self.cwd, 'test')
|
| 6603737e |
self.inventory_file = os.path.join(self.test_dir, 'simple_hosts')
self.large_range_inventory_file = os.path.join(self.test_dir, 'large_range')
self.complex_inventory_file = os.path.join(self.test_dir, 'complex_hosts')
self.inventory_script = os.path.join(self.test_dir, 'inventory_api.py') |
| 1f8805d9 |
self.inventory_dir = os.path.join(self.test_dir, 'inventory_dir') |
| 195e6d61 |
os.chmod(self.inventory_script, 0755)
def tearDown(self):
os.chmod(self.inventory_script, 0644)
|
| 9041addd |
def compare(self, left, right, sort=True):
if sort:
left = sorted(left)
right = sorted(right) |
| 1a00e263 |
print left
print right
assert left == right
|
| 195e6d61 |
def simple_inventory(self): |
| 157d21b1 |
return Inventory(self.inventory_file) |
| 195e6d61 |
|
| 6603737e |
def large_range_inventory(self):
return Inventory(self.large_range_inventory_file)
|
| 195e6d61 |
def script_inventory(self): |
| 157d21b1 |
return Inventory(self.inventory_script) |
| 195e6d61 |
|
| 157d21b1 |
def complex_inventory(self):
return Inventory(self.complex_inventory_file)
|
| 1f8805d9 |
def dir_inventory(self):
return Inventory(self.inventory_dir)
|
| 6603737e |
all_simple_hosts=['jupiter', 'saturn', 'zeus', 'hera',
'cerberus001','cerberus002','cerberus003',
'cottus99', 'cottus100',
'poseidon', 'thor', 'odin', 'loki',
'thrudgelmir0', 'thrudgelmir1', 'thrudgelmir2',
'thrudgelmir3', 'thrudgelmir4', 'thrudgelmir5',
'Hotep-a', 'Hotep-b', 'Hotep-c',
'BastC', 'BastD', ]
|
| 157d21b1 |
#####################################
### Simple inventory format tests |
| 3a24aa9a |
|
| 195e6d61 |
def test_simple(self):
inventory = self.simple_inventory()
hosts = inventory.list_hosts() |
| 6603737e |
self.assertEqual(sorted(hosts), sorted(self.all_simple_hosts)) |
| 195e6d61 |
def test_simple_all(self):
inventory = self.simple_inventory()
hosts = inventory.list_hosts('all') |
| 6603737e |
self.assertEqual(sorted(hosts), sorted(self.all_simple_hosts)) |
| 195e6d61 |
|
| ddd58226 |
def test_get_hosts(self):
inventory = Inventory('127.0.0.1,192.168.1.1')
hosts = inventory.get_hosts('!10.0.0.1')
hosts_all = inventory.get_hosts('all')
self.assertEqual(sorted(hosts), sorted(hosts_all))
def test_no_src(self):
inventory = Inventory('127.0.0.1,')
self.assertEqual(inventory.src(), None)
|
| 195e6d61 |
def test_simple_norse(self):
inventory = self.simple_inventory()
hosts = inventory.list_hosts("norse")
expected_hosts=['thor', 'odin', 'loki'] |
| 0669ce85 |
assert sorted(hosts) == sorted(expected_hosts) |
| 195e6d61 |
|
| 961ccdb2 |
def test_simple_ungrouped(self):
inventory = self.simple_inventory()
hosts = inventory.list_hosts("ungrouped")
|
| 1c9b43df |
expected_hosts=['jupiter', 'saturn',
'thrudgelmir0', 'thrudgelmir1', 'thrudgelmir2', |
| e3b2521f |
'thrudgelmir3', 'thrudgelmir4', 'thrudgelmir5'] |
| 0669ce85 |
assert sorted(hosts) == sorted(expected_hosts) |
| 961ccdb2 |
|
| 195e6d61 |
def test_simple_combined(self):
inventory = self.simple_inventory()
hosts = inventory.list_hosts("norse:greek")
|
| 1c9b43df |
expected_hosts=['zeus', 'hera', 'poseidon', |
| e3b2521f |
'cerberus001','cerberus002','cerberus003',
'cottus99','cottus100',
'thor', 'odin', 'loki'] |
| 0669ce85 |
assert sorted(hosts) == sorted(expected_hosts) |
| 195e6d61 |
def test_simple_restrict(self):
inventory = self.simple_inventory()
restricted_hosts = ['hera', 'poseidon', 'thor'] |
| 1c9b43df |
expected_hosts=['zeus', 'hera', 'poseidon', |
| e3b2521f |
'cerberus001','cerberus002','cerberus003',
'cottus99', 'cottus100',
'thor', 'odin', 'loki'] |
| 195e6d61 |
inventory.restrict_to(restricted_hosts)
hosts = inventory.list_hosts("norse:greek")
|
| 0669ce85 |
print "Hosts=%s" % hosts
print "Restricted=%s" % restricted_hosts
assert sorted(hosts) == sorted(restricted_hosts) |
| 195e6d61 |
inventory.lift_restriction()
hosts = inventory.list_hosts("norse:greek")
|
| 0669ce85 |
print hosts
print expected_hosts
assert sorted(hosts) == sorted(expected_hosts) |
| 195e6d61 |
|
| b655d579 |
def test_simple_string_ipv4(self):
inventory = Inventory('127.0.0.1,192.168.1.1')
hosts = inventory.list_hosts()
self.assertEqual(sorted(hosts), sorted(['127.0.0.1','192.168.1.1']))
def test_simple_string_ipv4_port(self):
inventory = Inventory('127.0.0.1:2222,192.168.1.1')
hosts = inventory.list_hosts()
self.assertEqual(sorted(hosts), sorted(['127.0.0.1','192.168.1.1']))
def test_simple_string_ipv4_vars(self):
inventory = Inventory('127.0.0.1:2222,192.168.1.1')
var = inventory.get_variables('127.0.0.1')
self.assertEqual(var['ansible_ssh_port'], 2222)
def test_simple_string_ipv6(self):
inventory = Inventory('FE80:EF45::12:1,192.168.1.1')
hosts = inventory.list_hosts()
self.assertEqual(sorted(hosts), sorted(['FE80:EF45::12:1','192.168.1.1']))
def test_simple_string_ipv6_port(self):
inventory = Inventory('[FE80:EF45::12:1]:2222,192.168.1.1')
hosts = inventory.list_hosts()
self.assertEqual(sorted(hosts), sorted(['FE80:EF45::12:1','192.168.1.1']))
def test_simple_string_ipv6_vars(self):
inventory = Inventory('[FE80:EF45::12:1]:2222,192.168.1.1')
var = inventory.get_variables('FE80:EF45::12:1')
self.assertEqual(var['ansible_ssh_port'], 2222)
|
| 195e6d61 |
def test_simple_vars(self):
inventory = self.simple_inventory()
vars = inventory.get_variables('thor')
|
| 0669ce85 |
print vars |
| 03541bab |
assert vars == {'group_names': ['norse'], |
| fe558f68 |
'inventory_hostname': 'thor',
'inventory_hostname_short': 'thor'} |
| 195e6d61 |
|
| 54f45261 |
def test_simple_port(self):
inventory = self.simple_inventory()
vars = inventory.get_variables('hera')
|
| 0669ce85 |
print vars |
| fe558f68 |
expected = { 'ansible_ssh_port': 3000,
'group_names': ['greek'],
'inventory_hostname': 'hera',
'inventory_hostname_short': 'hera' } |
| 0669ce85 |
print expected
assert vars == expected |
| 54f45261 |
|
| 6603737e |
def test_large_range(self):
inventory = self.large_range_inventory()
hosts = inventory.list_hosts()
self.assertEqual(sorted(hosts), sorted('bob%03i' %i for i in range(0, 143)))
|
| ddd58226 |
def test_subset(self):
inventory = self.simple_inventory()
inventory.subset('odin;thor,loki')
self.assertEqual(sorted(inventory.list_hosts()), sorted(['thor','odin','loki']))
def test_subset_filename(self):
inventory = self.simple_inventory()
inventory.subset('@' + os.path.join(self.test_dir, 'restrict_pattern'))
self.assertEqual(sorted(inventory.list_hosts()), sorted(['thor','odin']))
@raises(errors.AnsibleError)
def testinvalid_entry(self):
Inventory('1234')
|
| 157d21b1 |
###################################################
### INI file advanced tests
def test_complex_vars(self):
inventory = self.complex_inventory()
vars = inventory.get_variables('rtp_a')
print vars
expected = dict( |
| daf79780 |
a='1', b='2', c='3', d='10002', e='10003', f='10004 != 10005',
g=' g ', h=' h ', i="' i \"", j='" j',
rga='1', rgb='2', rgc='3', |
| 1c9b43df |
inventory_hostname='rtp_a', inventory_hostname_short='rtp_a', |
| 8e88667f |
group_names=[ 'eastcoast', 'nc', 'redundantgroup', 'redundantgroup2', 'redundantgroup3', 'rtp', 'us' ] |
| 157d21b1 |
)
print vars
print expected
assert vars == expected
|
| 7fbb7e07 |
def test_complex_group_names(self):
inventory = self.complex_inventory()
tests = { |
| 7fec9c3e |
'host1': [ 'role1', 'role3' ], |
| 7fbb7e07 |
'host2': [ 'role1', 'role2' ], |
| 7fec9c3e |
'host3': [ 'role2', 'role3' ] |
| 7fbb7e07 |
}
for host, roles in tests.iteritems():
group_names = inventory.get_variables(host)['group_names']
assert sorted(group_names) == sorted(roles)
|
| 05a7a372 |
def test_complex_exclude(self):
inventory = self.complex_inventory() |
| 259f2dc4 |
hosts = inventory.list_hosts("nc:florida:!triangle:!orlando") |
| 9041addd |
expected_hosts = ['miami', 'rtp_a', 'rtp_b', 'rtp_c'] |
| 259f2dc4 |
print "HOSTS=%s" % sorted(hosts)
print "EXPECTED=%s" % sorted(expected_hosts) |
| 05a7a372 |
assert sorted(hosts) == sorted(expected_hosts)
|
| 8bc1b69c |
def test_regex_exclude(self):
inventory = self.complex_inventory()
hosts = inventory.list_hosts("~rtp_[ac]")
expected_hosts = ['rtp_a', 'rtp_c']
print "HOSTS=%s" % sorted(hosts)
print "EXPECTED=%s" % sorted(expected_hosts)
assert sorted(hosts) == sorted(expected_hosts)
|
| 9041addd |
def test_complex_enumeration(self):
expected1 = ['rtp_a', 'rtp_b']
expected2 = ['rtp_c', 'tri_a']
expected3 = ['rtp_b', 'rtp_c', 'tri_a', 'tri_b', 'tri_c']
expected4 = ['orlando', 'rtp_c', 'tri_a']
inventory = self.complex_inventory()
print "ALL NC=%s" % inventory.list_hosts("nc") |
| ddd58226 |
# use "-1" instead of 0-1 to test the syntax, on purpose
hosts = inventory.list_hosts("nc[-1]") |
| 9041addd |
self.compare(hosts, expected1, sort=False)
hosts = inventory.list_hosts("nc[2-3]")
self.compare(hosts, expected2, sort=False)
hosts = inventory.list_hosts("nc[1-99999]")
self.compare(hosts, expected3, sort=False)
hosts = inventory.list_hosts("nc[2-3]:florida[1-2]")
self.compare(hosts, expected4, sort=False)
|
| 54b45e9b |
def test_complex_intersect(self):
inventory = self.complex_inventory()
hosts = inventory.list_hosts("nc:&redundantgroup:!rtp_c")
self.compare(hosts, ['rtp_a'])
hosts = inventory.list_hosts("nc:&triangle:!tri_c")
self.compare(hosts, ['tri_a', 'tri_b'])
|
| 2aea8a63 |
@raises(errors.AnsibleError)
def test_invalid_range(self):
Inventory(os.path.join(self.test_dir, 'inventory','test_incorrect_range')) |
| 9041addd |
|
| 835fdd1e |
@raises(errors.AnsibleError)
def test_missing_end(self):
Inventory(os.path.join(self.test_dir, 'inventory','test_missing_end'))
|
| 4f69b63f |
@raises(errors.AnsibleError)
def test_incorrect_format(self):
Inventory(os.path.join(self.test_dir, 'inventory','test_incorrect_format'))
|
| 64d35cb3 |
@raises(errors.AnsibleError)
def test_alpha_end_before_beg(self):
Inventory(os.path.join(self.test_dir, 'inventory','test_alpha_end_before_beg'))
|
| a0b73b18 |
def test_combined_range(self):
i = Inventory(os.path.join(self.test_dir, 'inventory','test_combined_range'))
hosts = i.list_hosts('test')
expected_hosts=['host1A','host2A','host1B','host2B']
assert sorted(hosts) == sorted(expected_hosts) |
| 835fdd1e |
|
| 157d21b1 |
################################################### |
| 195e6d61 |
### Inventory API tests
def test_script(self):
inventory = self.script_inventory()
hosts = inventory.list_hosts()
expected_hosts=['jupiter', 'saturn', 'zeus', 'hera', 'poseidon', 'thor', 'odin', 'loki']
print "Expected: %s"%(expected_hosts)
print "Got : %s"%(hosts)
assert sorted(hosts) == sorted(expected_hosts)
def test_script_all(self):
inventory = self.script_inventory()
hosts = inventory.list_hosts('all')
expected_hosts=['jupiter', 'saturn', 'zeus', 'hera', 'poseidon', 'thor', 'odin', 'loki']
assert sorted(hosts) == sorted(expected_hosts)
def test_script_norse(self):
inventory = self.script_inventory()
hosts = inventory.list_hosts("norse")
expected_hosts=['thor', 'odin', 'loki']
assert sorted(hosts) == sorted(expected_hosts)
def test_script_combined(self):
inventory = self.script_inventory()
hosts = inventory.list_hosts("norse:greek")
expected_hosts=['zeus', 'hera', 'poseidon', 'thor', 'odin', 'loki']
assert sorted(hosts) == sorted(expected_hosts)
def test_script_restrict(self):
inventory = self.script_inventory()
restricted_hosts = ['hera', 'poseidon', 'thor']
expected_hosts=['zeus', 'hera', 'poseidon', 'thor', 'odin', 'loki']
inventory.restrict_to(restricted_hosts)
hosts = inventory.list_hosts("norse:greek")
assert sorted(hosts) == sorted(restricted_hosts)
inventory.lift_restriction()
hosts = inventory.list_hosts("norse:greek")
assert sorted(hosts) == sorted(expected_hosts)
def test_script_vars(self):
inventory = self.script_inventory()
vars = inventory.get_variables('thor')
|
| 25637716 |
print "VARS=%s" % vars
|
| 03541bab |
assert vars == {'hammer':True,
'group_names': ['norse'], |
| fe558f68 |
'inventory_hostname': 'thor',
'inventory_hostname_short': 'thor'} |
| 1f8696f5 |
def test_hosts_list(self): |
| 7116c83c |
# Test the case when playbook 'hosts' var is a list. |
| 1f8696f5 |
inventory = self.script_inventory()
host_names = sorted(['thor', 'loki', 'odin']) # Not sure if sorting is in the contract or not
actual_hosts = inventory.get_hosts(host_names)
actual_host_names = [host.name for host in actual_hosts]
assert host_names == actual_host_names |
| 23a62f5a |
def test_script_multiple_groups(self):
inventory = self.script_inventory()
vars = inventory.get_variables('zeus')
print "VARS=%s" % vars |
| 6603737e |
|
| 23a62f5a |
assert vars == {'inventory_hostname': 'zeus',
'inventory_hostname_short': 'zeus',
'group_names': ['greek', 'major-god']} |
| 1f8805d9 |
|
| 4ae3ef8a |
# test disabled as needs to be updated to model desired behavior
#
#def test_dir_inventory(self):
# inventory = self.dir_inventory()
# vars = inventory.get_variables('zeus')
#
# print "VARS=%s" % vars
#
# assert vars == {'inventory_hostname': 'zeus',
# 'inventory_hostname_short': 'zeus',
# 'group_names': ['greek', 'major-god', 'ungrouped'],
# 'var_a': '1#2'} |