External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-01-13 19:10:07 -03:30
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #5914 from AlanCoding/ii_tests

Browse Source 
Port inventory_import tests to new system
This commit is contained in:
Alan Rominger 2017-04-12 15:07:56 -04:00 committed by GitHub
commit 047672a5b7
5 changed files with 233 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
awx/main/management/commands/inventory_import.py
View File

@ -675,7 +675,6 @@ class Command(NoArgsCommand):
with ignore_inventory_computed_fields():
self.inventory_source, created = InventorySource.objects.get_or_create(
inventory=self.inventory,
group=None,
source='file',
source_path=os.path.abspath(self.source),
overwrite=self.overwrite,

166
awx/main/tests/functional/commands/test_inventory_import.py Normal file
View File

@ -0,0 +1,166 @@
# Copyright (c) 2017 Ansible by Red Hat
# All Rights Reserved
# Python
import pytest
import mock
# Django
from django.core.management.base import CommandError
# AWX
from awx.main.management.commands.inventory_import import (
Command
)
from awx.main.models import Inventory, Host, Group
TEST_INVENTORY_INI = '''\
# Some comment about blah blah blah...
[webservers]
web1.example.com ansible_ssh_host=w1.example.net
web2.example.com
web3.example.com:1022
[webservers:vars] # Comment on a section
webvar=blah # Comment on an option
[dbservers]
db1.example.com
db2.example.com
[dbservers:vars]
dbvar=ugh
[servers:children]
webservers
dbservers
[servers:vars]
varb=B
[all:vars]
vara=A
[others]
10.11.12.13
10.12.14.16:8022
fe80::1610:9fff:fedd:654b
[fe80::1610:9fff:fedd:b654]:1022
::1
'''
@pytest.fixture(scope='session')
def test_dir(tmpdir_factory):
return tmpdir_factory.mktemp('inv_files', numbered=False)
@pytest.fixture(scope='session')
def ini_file(test_dir):
fn = test_dir.join('test_hosts')
fn.write(TEST_INVENTORY_INI)
return fn
@pytest.mark.django_db
@mock.patch.object(Command, 'check_license', mock.MagicMock())
class TestInvalidOptionsFunctional:
def test_invalid_options_invalid_source(self, inventory):
# Give invalid file to the command
cmd = Command()
with mock.patch('django.db.transaction.rollback'):
with pytest.raises(IOError) as err:
cmd.handle_noargs(
inventory_id=inventory.id,
source='/tmp/pytest-of-root/pytest-7/inv_files0-invalid')
assert 'Source does not exist' in err.value.message
def test_invalid_inventory_id(self, ini_file):
cmd = Command()
with pytest.raises(CommandError) as err:
cmd.handle_noargs(inventory_id=42, source=ini_file.dirname)
assert 'id = 42' in err.value.message
assert 'cannot be found' in err.value.message
def test_invalid_inventory_name(self, ini_file):
cmd = Command()
with pytest.raises(CommandError) as err:
cmd.handle_noargs(inventory_name='fooservers', source=ini_file.dirname)
assert 'name = fooservers' in err.value.message
assert 'cannot be found' in err.value.message
@pytest.mark.django_db
@mock.patch.object(Command, 'check_license', mock.MagicMock())
class TestINIImports:
def test_inventory_single_ini_import(self, inventory, ini_file, capsys):
cmd = Command()
r = cmd.handle_noargs(inventory_id=inventory.pk, source=ini_file.dirname)
out, err = capsys.readouterr()
assert r is None
assert out == ''
assert set(inventory.groups.values_list('name', flat=True)) == set([
'servers', 'dbservers', 'webservers', 'others'])
assert set(inventory.hosts.values_list('name', flat=True)) == set([
'web1.example.com', 'web2.example.com',
'web3.example.com', 'db1.example.com',
'db2.example.com', '10.11.12.13',
'10.12.14.16', 'fe80::1610:9fff:fedd:654b',
'fe80::1610:9fff:fedd:b654', '::1'])
reloaded_inv = Inventory.objects.get(pk=inventory.pk)
assert reloaded_inv.variables_dict == {'vara': 'A'}
assert Host.objects.get(name='web1.example.com').variables_dict == {'ansible_ssh_host': 'w1.example.net'}
assert Host.objects.get(name='web3.example.com').variables_dict == {'ansible_ssh_port': 1022}
assert Host.objects.get(name='fe80::1610:9fff:fedd:b654').variables_dict == {'ansible_ssh_port': 1022}
assert Host.objects.get(name='10.12.14.16').variables_dict == {'ansible_ssh_port': 8022}
servers = Group.objects.get(name='servers')
assert servers.variables_dict == {'varb': 'B'}
assert set(servers.children.values_list('name', flat=True)) == set(['dbservers', 'webservers'])
assert servers.hosts.count() == 0
servers = Group.objects.get(name='dbservers')
assert servers.variables_dict == {'dbvar': 'ugh'}
assert servers.children.count() == 0
assert set(servers.hosts.values_list('name', flat=True)) == set(['db1.example.com','db2.example.com'])
servers = Group.objects.get(name='webservers')
assert servers.variables_dict == {'webvar': 'blah'}
assert servers.children.count() == 0
assert set(servers.hosts.values_list('name', flat=True)) == set(['web1.example.com','web2.example.com', 'web3.example.com'])
assert reloaded_inv.inventory_sources.filter().count() == 1
invsrc = reloaded_inv.inventory_sources.first()
assert invsrc.source == 'file'
assert invsrc.inventory_updates.count() == 1
assert invsrc.inventory_updates.first().status == 'successful'
def test_inventory_import_group_vars_file(self, inventory, ini_file, tmpdir_factory):
# Create an extra group_vars file for group webservers
gvarf = tmpdir_factory.mktemp('inv_files/group_vars', numbered=False).join('webservers')
gvarf.write('''webservers_only_variable: foobar\n''')
cmd = Command()
cmd.handle_noargs(inventory_id=inventory.pk, source=ini_file.dirname)
servers = Group.objects.get(name='webservers')
assert servers.variables_dict == {'webvar': 'blah', 'webservers_only_variable': 'foobar'}
def test_inventory_import_host_vars_file(self, inventory, ini_file, tmpdir_factory):
# Create an extra host_vars file for one specific host
gvarf = tmpdir_factory.mktemp('inv_files/host_vars', numbered=False).join('web1.example.com')
gvarf.write('''host_only_variable: foobar\n''')
cmd = Command()
cmd.handle_noargs(inventory_id=inventory.pk, source=ini_file.dirname)
Host.objects.get(name='web1.example.com').variables_dict == {
'ansible_ssh_host': 'w1.example.net', 'host_only_variable': 'foobar'}

124
awx/main/tests/old/commands/commands_monolithic.py
View File

@ -16,7 +16,6 @@ import unittest2 as unittest
# Django
from django.conf import settings
from django.core.management import call_command
from django.core.management.base import CommandError
from django.utils.timezone import now
from django.test.utils import override_settings
@ -528,129 +527,6 @@ class InventoryImportTest(BaseCommandMixin, BaseLiveServerTest):
source_pks = group.inventory_sources.values_list('pk', flat=True)
self.assertTrue(inventory_source.pk in source_pks)
def test_invalid_options(self):
inventory_id = self.inventories[0].pk
inventory_name = self.inventories[0].name
# No options specified.
result, stdout, stderr = self.run_command('inventory_import')
self.assertTrue(isinstance(result, CommandError), result)
self.assertTrue('inventory-id' in str(result))
self.assertTrue('required' in str(result))
# Both inventory ID and name.
result, stdout, stderr = self.run_command('inventory_import',
inventory_id=inventory_id,
inventory_name=inventory_name)
self.assertTrue(isinstance(result, CommandError), result)
self.assertTrue('inventory-id' in str(result))
self.assertTrue('exclusive' in str(result))
# Inventory ID with overwrite and keep_vars.
result, stdout, stderr = self.run_command('inventory_import',
inventory_id=inventory_id,
overwrite=True, keep_vars=True)
self.assertTrue(isinstance(result, CommandError), result)
self.assertTrue('overwrite-vars' in str(result))
self.assertTrue('exclusive' in str(result))
result, stdout, stderr = self.run_command('inventory_import',
inventory_id=inventory_id,
overwrite_vars=True,
keep_vars=True)
self.assertTrue(isinstance(result, CommandError), result)
self.assertTrue('overwrite-vars' in str(result))
self.assertTrue('exclusive' in str(result))
# Inventory ID, but no source.
result, stdout, stderr = self.run_command('inventory_import',
inventory_id=inventory_id)
self.assertTrue(isinstance(result, CommandError), result)
self.assertTrue('--source' in str(result))
self.assertTrue('required' in str(result))
# Inventory ID, with invalid source.
invalid_source = ''.join([os.path.splitext(self.ini_path)[0] + '-invalid',
os.path.splitext(self.ini_path)[1]])
result, stdout, stderr = self.run_command('inventory_import',
inventory_id=inventory_id,
source=invalid_source)
self.assertTrue(isinstance(result, IOError), result)
self.assertTrue('not exist' in str(result))
# Invalid inventory ID.
invalid_id = Inventory.objects.order_by('-pk')[0].pk + 1
result, stdout, stderr = self.run_command('inventory_import',
inventory_id=invalid_id,
source=self.ini_path)
self.assertTrue(isinstance(result, CommandError), result)
self.assertTrue('found' in str(result))
# Invalid inventory name.
invalid_name = 'invalid inventory name'
result, stdout, stderr = self.run_command('inventory_import',
inventory_name=invalid_name,
source=self.ini_path)
self.assertTrue(isinstance(result, CommandError), result)
self.assertTrue('found' in str(result))
def test_ini_file(self, source=None):
inv_src = source or self.ini_path
# New empty inventory.
new_inv = self.organizations[0].inventories.create(name=os.path.basename(inv_src))
self.assertEqual(new_inv.hosts.count(), 0)
self.assertEqual(new_inv.groups.count(), 0)
result, stdout, stderr = self.run_command('inventory_import',
inventory_id=new_inv.pk,
source=inv_src)
self.assertEqual(result, None, stdout + stderr)
# Check that inventory is populated as expected.
new_inv = Inventory.objects.get(pk=new_inv.pk)
expected_group_names = set(['servers', 'dbservers', 'webservers', 'others'])
group_names = set(new_inv.groups.values_list('name', flat=True))
self.assertEqual(expected_group_names, group_names)
expected_host_names = set(['web1.example.com', 'web2.example.com',
'web3.example.com', 'db1.example.com',
'db2.example.com', '10.11.12.13',
'10.12.14.16', 'fe80::1610:9fff:fedd:654b',
'fe80::1610:9fff:fedd:b654', '::1'])
host_names = set(new_inv.hosts.values_list('name', flat=True))
self.assertEqual(expected_host_names, host_names)
if source and os.path.isdir(source):
self.assertEqual(new_inv.variables_dict, {
'vara': 'A',
'test_group_name': 'all',
})
else:
self.assertEqual(new_inv.variables_dict, {'vara': 'A'})
for host in new_inv.hosts.all():
if host.name == 'web1.example.com':
self.assertEqual(host.variables_dict,
{'ansible_ssh_host': 'w1.example.net'})
elif host.name in ('db1.example.com', 'db2.example.com') and source and os.path.isdir(source):
self.assertEqual(host.variables_dict, {'test_host_name': host.name})
elif host.name in ('web3.example.com', 'fe80::1610:9fff:fedd:b654'):
self.assertEqual(host.variables_dict, {'ansible_ssh_port': 1022})
elif host.name == '10.12.14.16':
self.assertEqual(host.variables_dict, {'ansible_ssh_port': 8022})
else:
self.assertEqual(host.variables_dict, {})
for group in new_inv.groups.all():
if group.name == 'servers':
self.assertEqual(group.variables_dict, {'varb': 'B'})
children = set(group.children.values_list('name', flat=True))
self.assertEqual(children, set(['dbservers', 'webservers']))
self.assertEqual(group.hosts.count(), 0)
elif group.name == 'dbservers':
if source and os.path.isdir(source):
self.assertEqual(group.variables_dict, {'dbvar': 'ugh', 'test_group_name': 'dbservers'})
else:
self.assertEqual(group.variables_dict, {'dbvar': 'ugh'})
self.assertEqual(group.children.count(), 0)
hosts = set(group.hosts.values_list('name', flat=True))
host_names = set(['db1.example.com','db2.example.com'])
self.assertEqual(hosts, host_names)
elif group.name == 'webservers':
self.assertEqual(group.variables_dict, {'webvar': 'blah'})
self.assertEqual(group.children.count(), 0)
hosts = set(group.hosts.values_list('name', flat=True))
host_names = set(['web1.example.com','web2.example.com',
'web3.example.com'])
self.assertEqual(hosts, host_names)
self.check_adhoc_inventory_source(new_inv)
def test_dir_with_ini_file(self):
self.create_test_dir(host_names=['db1.example.com', 'db2.example.com'],
group_names=['dbservers'], suffix='')

0
awx/main/tests/unit/commands/__init__.py Normal file
View File

67
awx/main/tests/unit/commands/test_inventory_import.py Normal file
View File

@ -0,0 +1,67 @@
# Copyright (c) 2017 Ansible by Red Hat
# All Rights Reserved
# Python
import pytest
# Django
from django.core.management.base import CommandError
# AWX
from awx.main.management.commands.inventory_import import (
Command
)
class TestInvalidOptions:
def test_invalid_options_no_options_specified(self):
cmd = Command()
with pytest.raises(CommandError) as err:
cmd.handle_noargs()
assert 'inventory-id' in err.value.message
assert 'required' in err.value.message
def test_invalid_options_name_and_id(self):
# You can not specify both name and if of the inventory
cmd = Command()
with pytest.raises(CommandError) as err:
cmd.handle_noargs(
inventory_id=42, inventory_name='my-inventory'
)
assert 'inventory-id' in err.value.message
assert 'exclusive' in err.value.message
def test_invalid_options_id_and_keep_vars(self):
# You can't overwrite and keep_vars at the same time, that wouldn't make sense
cmd = Command()
with pytest.raises(CommandError) as err:
cmd.handle_noargs(
inventory_id=42, overwrite=True, keep_vars=True
)
assert 'overwrite-vars' in err.value.message
assert 'exclusive' in err.value.message
def test_invalid_options_id_but_no_source(self):
# Need a source to import
cmd = Command()
with pytest.raises(CommandError) as err:
cmd.handle_noargs(
inventory_id=42, overwrite=True, keep_vars=True
)
assert 'overwrite-vars' in err.value.message
assert 'exclusive' in err.value.message
with pytest.raises(CommandError) as err:
cmd.handle_noargs(
inventory_id=42, overwrite_vars=True, keep_vars=True
)
assert 'overwrite-vars' in err.value.message
assert 'exclusive' in err.value.message
def test_invalid_options_missing_source(self):
cmd = Command()
with pytest.raises(CommandError) as err:
cmd.handle_noargs(inventory_id=42)
assert '--source' in err.value.message
assert 'required' in err.value.message