add system jobs to controller_node exceptions

This commit is contained in:
chris meyers
2018-06-06 15:04:41 -04:00
parent 8ee4b9680c
commit 74155dfc9d
4 changed files with 43 additions and 29 deletions

View File

@@ -13,9 +13,8 @@ from awx.main.models import (
JobTemplate,
User,
Job,
ProjectUpdate,
AdHocCommand,
InventoryUpdate,
ProjectUpdate,
)
from crum import impersonate
@@ -182,7 +181,11 @@ class TestControllerNode():
return AdHocCommand.objects.create(inventory=inventory)
@pytest.mark.django_db
def test_field_controller_node_exists(self, admin_user, job, project_update, inventory_update, adhoc, get):
def test_field_controller_node_exists(self, sqlite_copy_expert,
admin_user, job, project_update,
inventory_update, adhoc, get, system_job_factory):
system_job = system_job_factory()
r = get(reverse('api:unified_job_list') + '?id={}'.format(job.id), admin_user, expect=200)
assert 'controller_node' in r.data['results'][0]
@@ -197,3 +200,6 @@ class TestControllerNode():
r = get(reverse('api:inventory_update_detail', kwargs={'pk': inventory_update.pk}), admin_user, expect=200)
assert 'controller_node' not in r.data
r = get(reverse('api:system_job_detail', kwargs={'pk': system_job.pk}), admin_user, expect=200)
assert 'controller_node' not in r.data

View File

@@ -3,11 +3,8 @@
import base64
import json
import re
import shutil
import tempfile
from django.conf import settings
from django.db.backends.sqlite3.base import SQLiteCursorWrapper
import mock
import pytest
@@ -31,28 +28,6 @@ def _mk_inventory_update():
return iu
@pytest.fixture(scope='function')
def sqlite_copy_expert(request):
# copy_expert is postgres-specific, and SQLite doesn't support it; mock its
# behavior to test that it writes a file that contains stdout from events
path = tempfile.mkdtemp(prefix='job-event-stdout')
def write_stdout(self, sql, fd):
# simulate postgres copy_expert support with ORM code
parts = sql.split(' ')
tablename = parts[parts.index('from') + 1]
for cls in (JobEvent, AdHocCommandEvent, ProjectUpdateEvent,
InventoryUpdateEvent, SystemJobEvent):
if cls._meta.db_table == tablename:
for event in cls.objects.order_by('start_line').all():
fd.write(event.stdout.encode('utf-8'))
setattr(SQLiteCursorWrapper, 'copy_expert', write_stdout)
request.addfinalizer(lambda: shutil.rmtree(path))
request.addfinalizer(lambda: delattr(SQLiteCursorWrapper, 'copy_expert'))
return path
@pytest.mark.django_db
@pytest.mark.parametrize('Parent, Child, relation, view', [
[Job, JobEvent, 'job', 'api:job_stdout'],