set schedule.next_run to datetime.datetime

* Update JobNotificationMixin.context_stub
* .. TestJobNotificationMixin.CONTEXT_STRUCTURE
This commit is contained in:
Jim Ladd
2021-02-05 12:15:46 -08:00
parent 7b757f17a9
commit 5b3bb1e81d
2 changed files with 37 additions and 19 deletions

View File

@@ -348,7 +348,7 @@ class JobNotificationMixin(object):
'schedule': {'description': 'Sample schedule', 'schedule': {'description': 'Sample schedule',
'id': 42, 'id': 42,
'name': 'Stub schedule', 'name': 'Stub schedule',
'next_run': '2038-01-01T00:00:00Z'}, 'next_run': datetime.datetime(2038, 1, 1, 0, 0, 0, 0, tzinfo=datetime.timezone.utc)},
'unified_job_template': {'description': 'Sample unified job template description', 'unified_job_template': {'description': 'Sample unified job template description',
'id': 39, 'id': 39,
'name': 'Stub Job Template', 'name': 'Stub Job Template',

View File

@@ -6,7 +6,7 @@ import pytest
#from awx.main.models import NotificationTemplates, Notifications, JobNotificationMixin #from awx.main.models import NotificationTemplates, Notifications, JobNotificationMixin
from awx.main.models import (AdHocCommand, InventoryUpdate, Job, JobNotificationMixin, ProjectUpdate, from awx.main.models import (AdHocCommand, InventoryUpdate, Job, JobNotificationMixin, ProjectUpdate,
SystemJob, WorkflowJob) Schedule, SystemJob, WorkflowJob)
from awx.api.serializers import UnifiedJobSerializer from awx.api.serializers import UnifiedJobSerializer
@@ -75,7 +75,7 @@ class TestJobNotificationMixin(object):
'schedule': {'description': str, 'schedule': {'description': str,
'id': int, 'id': int,
'name': str, 'name': str,
'next_run': str}, 'next_run': datetime.datetime},
'unified_job_template': {'description': str, 'unified_job_template': {'description': str,
'id': int, 'id': int,
'name': str, 'name': str,
@@ -93,27 +93,27 @@ class TestJobNotificationMixin(object):
'workflow_url': str, 'workflow_url': str,
'url': str} 'url': str}
def check_structure(self, 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)
self.check_structure(expected_structure[key], obj[key])
else:
if key == 'job_explanation':
assert isinstance(str(obj[key]), expected_structure[key])
else:
assert isinstance(obj[key], expected_structure[key])
@pytest.mark.django_db @pytest.mark.django_db
@pytest.mark.parametrize('JobClass', [AdHocCommand, InventoryUpdate, Job, ProjectUpdate, SystemJob, WorkflowJob]) @pytest.mark.parametrize('JobClass', [AdHocCommand, InventoryUpdate, Job, ProjectUpdate, SystemJob, WorkflowJob])
def test_context(self, JobClass, sqlite_copy_expert, project, inventory_source): 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 """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.""" 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:
if key == 'job_explanation':
assert isinstance(str(obj[key]), expected_structure[key])
else:
assert isinstance(obj[key], expected_structure[key])
kwargs = {} kwargs = {}
if JobClass is InventoryUpdate: if JobClass is InventoryUpdate:
kwargs['inventory_source'] = inventory_source kwargs['inventory_source'] = inventory_source
@@ -125,8 +125,26 @@ class TestJobNotificationMixin(object):
job_serialization = UnifiedJobSerializer(job).to_representation(job) job_serialization = UnifiedJobSerializer(job).to_representation(job)
context = job.context(job_serialization) context = job.context(job_serialization)
check_structure(TestJobNotificationMixin.CONTEXT_STRUCTURE, context) self.check_structure(TestJobNotificationMixin.CONTEXT_STRUCTURE, context)
@pytest.mark.django_db
def test_schedule_context(self, job_template, admin_user):
schedule = Schedule.objects.create(
name='job-schedule',
rrule='DTSTART:20171129T155939z\nFREQ=MONTHLY',
unified_job_template=job_template
)
job = Job.objects.create(
name='fake-job',
launch_type='workflow',
schedule=schedule,
job_template=job_template
)
job_serialization = UnifiedJobSerializer(job).to_representation(job)
context = job.context(job_serialization)
self.check_structure(TestJobNotificationMixin.CONTEXT_STRUCTURE, context)
@pytest.mark.django_db @pytest.mark.django_db
def test_context_job_metadata_with_unicode(self): def test_context_job_metadata_with_unicode(self):