mirror of
https://github.com/ansible/awx.git
synced 2026-05-11 03:17:38 -02:30
create jinja context based on job serialization
This commit is contained in:
Jim Ladd
148
awx/main/tests/functional/models/test_notifications.py
Normal file
148
awx/main/tests/functional/models/test_notifications.py
Normal file
@@ -0,0 +1,148 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from copy import deepcopy
|
||||
import datetime
|
||||
|
||||
import pytest
|
||||
|
||||
#from awx.main.models import NotificationTemplates, Notifications, JobNotificationMixin
|
||||
from awx.main.models import (AdHocCommand, InventoryUpdate, Job, JobNotificationMixin, ProjectUpdate,
|
||||
SystemJob, WorkflowJob)
|
||||
from awx.api.serializers import UnifiedJobSerializer
|
||||
|
||||
|
||||
class TestJobNotificationMixin(object):
|
||||
CONTEXT_STRUCTURE = {'job': {'allow_simultaneous': bool,
|
||||
'custom_virtualenv': str,
|
||||
'controller_node': str,
|
||||
'created': datetime.datetime,
|
||||
'description': str,
|
||||
'diff_mode': bool,
|
||||
'elapsed': float,
|
||||
'execution_node': str,
|
||||
'failed': bool,
|
||||
'finished': bool,
|
||||
'force_handlers': bool,
|
||||
'forks': int,
|
||||
'host_status_counts': {'skipped': int, 'ok': int, 'changed': int,
|
||||
'failures': int, 'dark': int},
|
||||
'id': int,
|
||||
'job_explanation': str,
|
||||
'job_slice_count': int,
|
||||
'job_slice_number': int,
|
||||
'job_tags': str,
|
||||
'job_type': str,
|
||||
'launch_type': str,
|
||||
'limit': str,
|
||||
'modified': datetime.datetime,
|
||||
'name': str,
|
||||
'playbook': str,
|
||||
'playbook_counts': {'play_count': int, 'task_count': int},
|
||||
'scm_revision': str,
|
||||
'skip_tags': str,
|
||||
'start_at_task': str,
|
||||
'started': str,
|
||||
'status': str,
|
||||
'summary_fields': {'created_by': {'first_name': str,
|
||||
'id': int,
|
||||
'last_name': str,
|
||||
'username': str},
|
||||
'instance_group': {'id': int, 'name': str},
|
||||
'inventory': {'description': str,
|
||||
'groups_with_active_failures': int,
|
||||
'has_active_failures': bool,
|
||||
'has_inventory_sources': bool,
|
||||
'hosts_with_active_failures': int,
|
||||
'id': int,
|
||||
'inventory_sources_with_failures': int,
|
||||
'kind': str,
|
||||
'name': str,
|
||||
'organization_id': int,
|
||||
'total_groups': int,
|
||||
'total_hosts': int,
|
||||
'total_inventory_sources': int},
|
||||
'job_template': {'description': str,
|
||||
'id': int,
|
||||
'name': str},
|
||||
'labels': {'count': int, 'results': list},
|
||||
'project': {'description': str,
|
||||
'id': int,
|
||||
'name': str,
|
||||
'scm_type': str,
|
||||
'status': str},
|
||||
'project_update': {'id': int, 'name': str, 'description': str, 'status': str, 'failed': bool},
|
||||
'unified_job_template': {'description': str,
|
||||
'id': int,
|
||||
'name': str,
|
||||
'unified_job_type': str},
|
||||
'source_workflow_job': {'description': str,
|
||||
'elapsed': float,
|
||||
'failed': bool,
|
||||
'id': int,
|
||||
'name': str,
|
||||
'status': str}},
|
||||
|
||||
'timeout': int,
|
||||
'type': str,
|
||||
'url': str,
|
||||
'use_fact_cache': bool,
|
||||
'verbosity': int},
|
||||
'job_friendly_name': str,
|
||||
'job_summary_dict': str,
|
||||
'url': str}
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.parametrize('JobClass', [AdHocCommand, InventoryUpdate, Job, ProjectUpdate, SystemJob, WorkflowJob])
|
||||
def test_context(self, JobClass, sqlite_copy_expert, project, inventory_source):
|
||||
"""The Jinja context defines all of the fields that can be used by a template. Ensure that the context generated
|
||||
for each job type has the expected structure."""
|
||||
def check_structure(expected_structure, obj):
|
||||
if isinstance(expected_structure, dict):
|
||||
assert isinstance(obj, dict)
|
||||
for key in obj:
|
||||
assert key in expected_structure
|
||||
if obj[key] is None:
|
||||
continue
|
||||
if isinstance(expected_structure[key], dict):
|
||||
assert isinstance(obj[key], dict)
|
||||
check_structure(expected_structure[key], obj[key])
|
||||
else:
|
||||
assert isinstance(obj[key], expected_structure[key])
|
||||
kwargs = {}
|
||||
if JobClass is InventoryUpdate:
|
||||
kwargs['inventory_source'] = inventory_source
|
||||
elif JobClass is ProjectUpdate:
|
||||
kwargs['project'] = project
|
||||
|
||||
job = JobClass.objects.create(name='foo', **kwargs)
|
||||
job_serialization = UnifiedJobSerializer(job).to_representation(job)
|
||||
|
||||
context = job.context(job_serialization)
|
||||
check_structure(TestJobNotificationMixin.CONTEXT_STRUCTURE, context)
|
||||
|
||||
def test_context_stub(self):
|
||||
"""The context stub is a fake context used to validate custom notification messages. Ensure that
|
||||
this also has the expected structure. Furthermore, ensure that the stub context contains
|
||||
*all* fields that could possibly be included in a context."""
|
||||
def check_structure_and_completeness(expected_structure, obj):
|
||||
expected_structure = deepcopy(expected_structure)
|
||||
if isinstance(expected_structure, dict):
|
||||
assert isinstance(obj, dict)
|
||||
for key in obj:
|
||||
assert key in expected_structure
|
||||
# Context stub should not have any undefined fields
|
||||
assert obj[key] is not None
|
||||
if isinstance(expected_structure[key], dict):
|
||||
assert isinstance(obj[key], dict)
|
||||
check_structure_and_completeness(expected_structure[key], obj[key])
|
||||
expected_structure.pop(key)
|
||||
else:
|
||||
assert isinstance(obj[key], expected_structure[key])
|
||||
expected_structure.pop(key)
|
||||
# Ensure all items in expected structure were present
|
||||
assert not len(expected_structure)
|
||||
|
||||
context_stub = JobNotificationMixin.context_stub()
|
||||
check_structure_and_completeness(TestJobNotificationMixin.CONTEXT_STRUCTURE, context_stub)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user