zware/awx
1
0
Fork 0
mirror of https://github.com/ZwareBear/awx.git synced 2026-04-28 10:51:49 -05:00
Code Issues Packages Projects Releases Wiki Activity

Start on tests for inventory_import command.

Browse Source
This commit is contained in:
Chris Church
2013-07-21 19:47:36 -04:00
parent 6fda31e790
commit 5ef8a48600
2 changed files with 195 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

216
awx/main/tests/commands.py
View File

@@ -7,6 +7,7 @@ import os
import StringIO
import sys
import tempfile
import time
# Django
from django.conf import settings
@@ -16,10 +17,38 @@ from django.core.management.base import CommandError
from django.utils.timezone import now
# AWX
from awx.main.licenses import LicenseWriter
from awx.main.models import *
from awx.main.tests.base import BaseTest
__all__ = ['CleanupDeletedTest']
__all__ = ['CleanupDeletedTest', 'InventoryImportTest']
TEST_INVENTORY_INI = '''\
[webservers]
web1.example.com
web2.example.com
web3.example.com
[webservers:vars]
webvar=blah
[dbservers]
db1.example.com
db2.example.com
[dbservers:vars]
dbvar=ugh
[servers:children]
webservers
dbservers
[servers:vars]
varb=B
[all:vars]
vara=A
'''
class BaseCommandTest(BaseTest):
'''
@@ -45,44 +74,7 @@ class BaseCommandTest(BaseTest):
if os.path.exists(tf):
os.remove(tf)
def run_command(self, name, *args, **options):
'''
Run a management command and capture its stdout/stderr along with any
exceptions.
'''
command_runner = options.pop('command_runner', call_command)
stdin_fileobj = options.pop('stdin_fileobj', None)
options.setdefault('verbosity', 1)
options.setdefault('interactive', False)
original_stdin = sys.stdin
original_stdout = sys.stdout
original_stderr = sys.stderr
if stdin_fileobj:
sys.stdin = stdin_fileobj
sys.stdout = StringIO.StringIO()
sys.stderr = StringIO.StringIO()
result = None
try:
result = command_runner(name, *args, **options)
except Exception, e:
result = e
except SystemExit, e:
result = e
finally:
captured_stdout = sys.stdout.getvalue()
captured_stderr = sys.stderr.getvalue()
sys.stdin = original_stdin
sys.stdout = original_stdout
sys.stderr = original_stderr
return result, captured_stdout, captured_stderr
class CleanupDeletedTest(BaseCommandTest):
'''
Test cases for cleanup_deleted management command.
'''
def setUp(self):
super(CleanupDeletedTest, self).setUp()
def create_test_inventories(self):
self.setup_users()
self.organizations = self.make_organizations(self.super_django_user, 2)
self.projects = self.make_projects(self.normal_django_user, 2)
@@ -124,6 +116,47 @@ class CleanupDeletedTest(BaseCommandTest):
group.parents.add(groups[3])
self.groups.extend(groups)
def run_command(self, name, *args, **options):
'''
Run a management command and capture its stdout/stderr along with any
exceptions.
'''
command_runner = options.pop('command_runner', call_command)
stdin_fileobj = options.pop('stdin_fileobj', None)
options.setdefault('verbosity', 1)
options.setdefault('interactive', False)
original_stdin = sys.stdin
original_stdout = sys.stdout
original_stderr = sys.stderr
if stdin_fileobj:
sys.stdin = stdin_fileobj
sys.stdout = StringIO.StringIO()
sys.stderr = StringIO.StringIO()
result = None
try:
result = command_runner(name, *args, **options)
except Exception, e:
result = e
except SystemExit, e:
result = e
finally:
captured_stdout = sys.stdout.getvalue()
captured_stderr = sys.stderr.getvalue()
sys.stdin = original_stdin
sys.stdout = original_stdout
sys.stderr = original_stderr
return result, captured_stdout, captured_stderr
class CleanupDeletedTest(BaseCommandTest):
'''
Test cases for cleanup_deleted management command.
'''
def setUp(self):
super(CleanupDeletedTest, self).setUp()
self.create_test_inventories()
def get_model_counts(self):
def get_models(m):
if not m._meta.abstract:
@@ -204,3 +237,108 @@ class CleanupDeletedTest(BaseCommandTest):
counts_after = self.get_user_counts()
self.assertNotEqual(counts_before, counts_after)
self.assertFalse(counts_after[1])
class InventoryImportTest(BaseCommandTest):
'''
Test cases for inventory_import management command.
'''
def setUp(self):
super(InventoryImportTest, self).setUp()
self.create_test_inventories()
self.create_test_ini()
self.create_test_license_file()
def create_test_license_file(self):
writer = LicenseWriter(
company_name='AWX',
contact_name='AWX Admin',
contact_email='awx@example.com',
license_date=int(time.time() + 3600),
instance_count=500,
)
handle, license_path = tempfile.mkstemp(suffix='.json')
os.close(handle)
writer.write_file(license_path)
self._temp_files.append(license_path)
os.environ['AWX_LICENSE_FILE'] = license_path
def create_test_ini(self):
handle, self.ini_path = tempfile.mkstemp(suffix='.txt')
ini_file = os.fdopen(handle, 'w')
ini_file.write(TEST_INVENTORY_INI)
ini_file.close()
self._temp_files.append(self.ini_path)
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, CommandError), 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('matched' 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('matched' in str(result))
def test_ini_file(self):
# New empty inventory.
new_inv = self.organizations[0].inventories.create(name='newb')
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=self.ini_path)
self.assertEqual(result, None)
# FIXME
def test_executable_file(self):
pass
# FIXME