External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-03-16 08:27:29 -02:30
Code Issues Packages Projects Releases Wiki Activity

Updated callback module to delegate to acom_callback_event management command.

Browse Source
This commit is contained in:
Chris Church
2013-04-04 13:59:32 -04:00
parent 0a306ee0ad
commit 1b93886be2
8 changed files with 216 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
lib/main/tests/__init__.py
View File

@@ -20,5 +20,5 @@ from lib.main.tests.organizations import OrganizationsTest
from lib.main.tests.users import UsersTest
from lib.main.tests.inventory import InventoryTest
from lib.main.tests.projects import ProjectsTest
from lib.main.tests.commands import AcomInventoryTest
from lib.main.tests.commands import *
from lib.main.tests.tasks import RunLaunchJobTest

156
lib/main/tests/commands.py
View File

@@ -20,11 +20,15 @@ import json
import os
import StringIO
import sys
import tempfile
from django.core.management import call_command
from django.core.management.base import CommandError
from django.utils.timezone import now
from lib.main.models import *
from lib.main.tests.base import BaseTest
__all__ = ['AcomInventoryTest', 'AcomCallbackEventTest']
class BaseCommandTest(BaseTest):
'''
Base class for tests that run management commands.
@@ -33,6 +37,7 @@ class BaseCommandTest(BaseTest):
def setUp(self):
super(BaseCommandTest, self).setUp()
self._environ = dict(os.environ.items())
self._temp_files = []
def tearDown(self):
super(BaseCommandTest, self).tearDown()
@@ -42,16 +47,23 @@ class BaseCommandTest(BaseTest):
for k,v in os.environ.items():
if k not in self._environ.keys():
del os.environ[k]
for tf in self._temp_files:
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.
'''
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
@@ -64,6 +76,7 @@ class BaseCommandTest(BaseTest):
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
@@ -244,4 +257,145 @@ class AcomInventoryTest(BaseCommandTest):
host='blah')
self.assertTrue(isinstance(result, CommandError))
self.assertEqual(json.loads(stdout), {})
class AcomCallbackEventTest(BaseCommandTest):
'''
Test cases for acom_callback_event management command.
'''
def setUp(self):
super(AcomCallbackEventTest, self).setUp()
self.setup_users()
self.organization = self.make_organizations(self.super_django_user, 1)[0]
self.project = self.make_projects(self.normal_django_user, 1)[0]
self.organization.projects.add(self.project)
self.inventory = Inventory.objects.create(name='test-inventory',
organization=self.organization)
self.host = self.inventory.hosts.create(name='host.example.com',
inventory=self.inventory)
self.group = self.inventory.groups.create(name='test-group',
inventory=self.inventory)
self.group.hosts.add(self.host)
self.launch_job = LaunchJob.objects.create(name='test-launch-job',
inventory=self.inventory,
project=self.project)
self.launch_job_status = self.launch_job.launch_job_statuses.create(
name='launch-job-status-%s' % now().isoformat())
self.valid_kwargs = {
'launch_job_status_id': self.launch_job_status.id,
'event_type': 'playbook_on_start',
'event_data_json': json.dumps({'test_event_data': [2,4,6]}),
}
def test_with_launch_job_status_not_running(self):
# Events can only be added when the launch job is running.
self.assertEqual(self.launch_job_status.status, 'pending')
result, stdout, stderr = self.run_command('acom_callback_event',
**self.valid_kwargs)
self.assertTrue(isinstance(result, CommandError))
self.assertTrue('unable to add event ' in str(result).lower())
self.launch_job_status.status = 'successful'
self.launch_job_status.save()
result, stdout, stderr = self.run_command('acom_callback_event',
**self.valid_kwargs)
self.assertTrue(isinstance(result, CommandError))
self.assertTrue('unable to add event ' in str(result).lower())
self.launch_job_status.status = 'failed'
self.launch_job_status.save()
result, stdout, stderr = self.run_command('acom_callback_event',
**self.valid_kwargs)
self.assertTrue(isinstance(result, CommandError))
self.assertTrue('unable to add event ' in str(result).lower())
def test_with_invalid_args(self):
self.launch_job_status.status = 'running'
self.launch_job_status.save()
# Event type not given.
kwargs = dict(self.valid_kwargs.items())
kwargs.pop('event_type')
result, stdout, stderr = self.run_command('acom_callback_event', **kwargs)
self.assertTrue(isinstance(result, CommandError))
self.assertTrue('no event specified' in str(result).lower())
# Invalid event type.
kwargs = dict(self.valid_kwargs.items())
kwargs['event_type'] = 'invalid_event_type'
result, stdout, stderr = self.run_command('acom_callback_event', **kwargs)
self.assertTrue(isinstance(result, CommandError))
self.assertTrue('unsupported event' in str(result).lower())
# Neither file or data specified.
kwargs = dict(self.valid_kwargs.items())
kwargs.pop('event_data_json')
result, stdout, stderr = self.run_command('acom_callback_event', **kwargs)
self.assertTrue(isinstance(result, CommandError))
self.assertTrue('either --file or --data' in str(result).lower())
# Non-integer launch job status ID.
kwargs = dict(self.valid_kwargs.items())
kwargs['launch_job_status_id'] = 'foo'
result, stdout, stderr = self.run_command('acom_callback_event', **kwargs)
self.assertTrue(isinstance(result, CommandError))
self.assertTrue('id must be an integer' in str(result).lower())
# No launch job status ID.
kwargs = dict(self.valid_kwargs.items())
kwargs.pop('launch_job_status_id')
result, stdout, stderr = self.run_command('acom_callback_event', **kwargs)
self.assertTrue(isinstance(result, CommandError))
self.assertTrue('no launch job status id' in str(result).lower())
# Invalid launch job status ID.
kwargs = dict(self.valid_kwargs.items())
kwargs['launch_job_status_id'] = 9999
result, stdout, stderr = self.run_command('acom_callback_event', **kwargs)
self.assertTrue(isinstance(result, CommandError))
self.assertTrue('not found' in str(result).lower())
# Invalid inline JSON data.
kwargs = dict(self.valid_kwargs.items())
kwargs['event_data_json'] = 'invalid json'
result, stdout, stderr = self.run_command('acom_callback_event', **kwargs)
self.assertTrue(isinstance(result, CommandError))
self.assertTrue('error parsing json' in str(result).lower())
# Invalid file specified.
kwargs = dict(self.valid_kwargs.items())
kwargs.pop('event_data_json')
h, tf = tempfile.mkstemp()
os.close(h)
os.remove(tf)
kwargs['event_data_file'] = '%s.json' % tf
result, stdout, stderr = self.run_command('acom_callback_event', **kwargs)
self.assertTrue(isinstance(result, CommandError))
self.assertTrue('reading from' in str(result).lower())
def test_with_valid_args(self):
self.launch_job_status.status = 'running'
self.launch_job_status.save()
# Default valid args.
kwargs = dict(self.valid_kwargs.items())
result, stdout, stderr = self.run_command('acom_callback_event', **kwargs)
self.assertEqual(result, None)
self.assertEqual(self.launch_job_status.launch_job_status_events.count(), 1)
# Pass launch job status in environment instead.
kwargs = dict(self.valid_kwargs.items())
kwargs.pop('launch_job_status_id')
os.environ['ACOM_LAUNCH_JOB_STATUS_ID'] = str(self.launch_job_status.id)
result, stdout, stderr = self.run_command('acom_callback_event', **kwargs)
self.assertEqual(result, None)
self.assertEqual(self.launch_job_status.launch_job_status_events.count(), 2)
os.environ.pop('ACOM_LAUNCH_JOB_STATUS_ID', None)
# Test with JSON data in a file instead.
kwargs = dict(self.valid_kwargs.items())
kwargs.pop('event_data_json')
h, tf = tempfile.mkstemp(suffix='.json')
self._temp_files.append(tf)
f = os.fdopen(h, 'w')
json.dump({'some_event_data': [1, 2, 3]}, f)
f.close()
kwargs['event_data_file'] = tf
result, stdout, stderr = self.run_command('acom_callback_event', **kwargs)
self.assertEqual(result, None)
self.assertEqual(self.launch_job_status.launch_job_status_events.count(), 3)
# Test with JSON data from stdin.
kwargs = dict(self.valid_kwargs.items())
kwargs.pop('event_data_json')
kwargs['event_data_file'] = '-'
kwargs['stdin_fileobj'] = StringIO.StringIO(json.dumps({'blah': 'bleep'}))
result, stdout, stderr = self.run_command('acom_callback_event', **kwargs)
self.assertEqual(result, None)
self.assertEqual(self.launch_job_status.launch_job_status_events.count(), 4)

10
lib/main/tests/tasks.py
View File

@@ -88,12 +88,10 @@ class RunLaunchJobTest(BaseCeleryTest):
launch_job_status = self.launch_job.start()
self.assertEqual(launch_job_status.status, 'pending')
launch_job_status = LaunchJobStatus.objects.get(pk=launch_job_status.pk)
print 'stdout:', launch_job_status.result_stdout
print 'stderr:', launch_job_status.result_stderr
print launch_job_status.status
print settings.DATABASES
#print 'stdout:', launch_job_status.result_stdout
#print 'stderr:', launch_job_status.result_stderr
#print launch_job_status.status
#print settings.DATABASES
self.assertEqual(launch_job_status.status, 'successful')
self.assertTrue(launch_job_status.result_stdout)
launch_job_status_events = launch_job_status.launch_job_status_events.all()