mirror of
https://github.com/ansible/awx.git
synced 2026-02-12 07:04:45 -03:30
fefab89815
Revert the --force flags use the update id as metric for role caching Shift the movement of cache to job folder from rsync task to python Only install roles and collections if needed Deal with roles and collections for jobs without sync Skip local copy if roles or collections turned off update docs for content caching Design pivot - use empty cache dir to indicate lack of content Do not cache content if we did not install content Test changes to allay concerns about reliability of local_path Do not blow away cache for SCM inventory updates
853 lines
26 KiB
Python
853 lines
26 KiB
Python
# Python
|
|
import pytest
|
|
from unittest import mock
|
|
import tempfile
|
|
import shutil
|
|
import urllib.parse
|
|
from unittest.mock import PropertyMock
|
|
|
|
# Django
|
|
from django.urls import resolve
|
|
from django.http import Http404
|
|
from django.core.handlers.exception import response_for_exception
|
|
from django.contrib.auth.models import User
|
|
from django.core.serializers.json import DjangoJSONEncoder
|
|
from django.db.backends.sqlite3.base import SQLiteCursorWrapper
|
|
|
|
# AWX
|
|
from awx.main.fields import JSONBField
|
|
from awx.main.models.projects import Project
|
|
from awx.main.models.ha import Instance
|
|
|
|
from rest_framework.test import (
|
|
APIRequestFactory,
|
|
force_authenticate,
|
|
)
|
|
|
|
from awx.main.models.credential import CredentialType, Credential
|
|
from awx.main.models.jobs import JobTemplate, SystemJobTemplate
|
|
from awx.main.models.inventory import (
|
|
Group,
|
|
Inventory,
|
|
InventoryUpdate,
|
|
InventorySource,
|
|
CustomInventoryScript
|
|
)
|
|
from awx.main.models.organization import (
|
|
Organization,
|
|
Team,
|
|
)
|
|
from awx.main.models.rbac import Role
|
|
from awx.main.models.notifications import (
|
|
NotificationTemplate,
|
|
Notification
|
|
)
|
|
from awx.main.models.events import (
|
|
JobEvent,
|
|
AdHocCommandEvent,
|
|
ProjectUpdateEvent,
|
|
InventoryUpdateEvent,
|
|
SystemJobEvent,
|
|
)
|
|
from awx.main.models.workflow import WorkflowJobTemplate
|
|
from awx.main.models.ad_hoc_commands import AdHocCommand
|
|
from awx.main.models.oauth import OAuth2Application as Application
|
|
|
|
__SWAGGER_REQUESTS__ = {}
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def swagger_autogen(requests=__SWAGGER_REQUESTS__):
|
|
return requests
|
|
|
|
|
|
@pytest.fixture
|
|
def user():
|
|
def u(name, is_superuser=False):
|
|
try:
|
|
user = User.objects.get(username=name)
|
|
except User.DoesNotExist:
|
|
user = User(username=name, is_superuser=is_superuser)
|
|
user.set_password(name)
|
|
user.save()
|
|
return user
|
|
return u
|
|
|
|
|
|
@pytest.fixture
|
|
def check_jobtemplate(project, inventory, credential):
|
|
jt = JobTemplate.objects.create(
|
|
job_type='check',
|
|
project=project,
|
|
inventory=inventory,
|
|
name='check-job-template'
|
|
)
|
|
jt.credentials.add(credential)
|
|
return jt
|
|
|
|
|
|
@pytest.fixture
|
|
def deploy_jobtemplate(project, inventory, credential):
|
|
jt = JobTemplate.objects.create(
|
|
job_type='run',
|
|
project=project,
|
|
inventory=inventory,
|
|
name='deploy-job-template'
|
|
)
|
|
jt.credentials.add(credential)
|
|
return jt
|
|
|
|
|
|
@pytest.fixture
|
|
def team(organization):
|
|
return organization.teams.create(name='test-team')
|
|
|
|
|
|
@pytest.fixture
|
|
def team_member(user, team):
|
|
ret = user('team-member', False)
|
|
team.member_role.members.add(ret)
|
|
return ret
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def project_playbooks():
|
|
'''
|
|
Return playbook_files as playbooks for manual projects when testing.
|
|
'''
|
|
class PlaybooksMock(mock.PropertyMock):
|
|
def __get__(self, obj, obj_type):
|
|
return obj.playbook_files
|
|
mocked = mock.patch.object(Project, 'playbooks', new_callable=PlaybooksMock)
|
|
mocked.start()
|
|
|
|
|
|
@pytest.fixture
|
|
def run_computed_fields_right_away(request):
|
|
|
|
def run_me(inventory_id):
|
|
i = Inventory.objects.get(id=inventory_id)
|
|
i.update_computed_fields()
|
|
|
|
mocked = mock.patch(
|
|
'awx.main.signals.update_inventory_computed_fields.delay',
|
|
new=run_me
|
|
)
|
|
mocked.start()
|
|
|
|
request.addfinalizer(mocked.stop)
|
|
|
|
|
|
@pytest.fixture
|
|
@mock.patch.object(Project, "update", lambda self, **kwargs: None)
|
|
def project(instance, organization):
|
|
prj = Project.objects.create(name="test-proj",
|
|
description="test-proj-desc",
|
|
organization=organization,
|
|
playbook_files=['helloworld.yml', 'alt-helloworld.yml'],
|
|
scm_revision='1234567890123456789012345678901234567890',
|
|
scm_url='localhost',
|
|
scm_type='git'
|
|
)
|
|
return prj
|
|
|
|
|
|
@pytest.fixture
|
|
@mock.patch.object(Project, "update", lambda self, **kwargs: None)
|
|
def manual_project(instance, organization):
|
|
prj = Project.objects.create(name="test-manual-proj",
|
|
description="manual-proj-desc",
|
|
organization=organization,
|
|
playbook_files=['helloworld.yml', 'alt-helloworld.yml'],
|
|
local_path='_92__test_proj'
|
|
)
|
|
return prj
|
|
|
|
|
|
@pytest.fixture
|
|
def project_factory(organization):
|
|
def factory(name):
|
|
try:
|
|
prj = Project.objects.get(name=name)
|
|
except Project.DoesNotExist:
|
|
prj = Project.objects.create(name=name,
|
|
description="description for " + name,
|
|
organization=organization
|
|
)
|
|
return prj
|
|
return factory
|
|
|
|
|
|
@pytest.fixture
|
|
def job_factory(jt_linked, admin):
|
|
def factory(job_template=jt_linked, initial_state='new', created_by=admin):
|
|
return job_template.create_unified_job(_eager_fields={
|
|
'status': initial_state, 'created_by': created_by})
|
|
return factory
|
|
|
|
|
|
@pytest.fixture
|
|
def team_factory(organization):
|
|
def factory(name):
|
|
try:
|
|
t = Team.objects.get(name=name)
|
|
except Team.DoesNotExist:
|
|
t = Team.objects.create(name=name,
|
|
description="description for " + name,
|
|
organization=organization)
|
|
return t
|
|
return factory
|
|
|
|
|
|
@pytest.fixture
|
|
def user_project(user):
|
|
owner = user('owner')
|
|
return Project.objects.create(name="test-user-project", created_by=owner, description="test-user-project-desc")
|
|
|
|
|
|
@pytest.fixture
|
|
def insights_project():
|
|
return Project.objects.create(name="test-insights-project", scm_type="insights")
|
|
|
|
|
|
@pytest.fixture
|
|
def instance(settings):
|
|
return Instance.objects.create(uuid=settings.SYSTEM_UUID, hostname="instance.example.org", capacity=100)
|
|
|
|
|
|
@pytest.fixture
|
|
def organization(instance):
|
|
return Organization.objects.create(name="test-org", description="test-org-desc")
|
|
|
|
|
|
@pytest.fixture
|
|
def credentialtype_kube():
|
|
kube = CredentialType.defaults['kubernetes_bearer_token']()
|
|
kube.save()
|
|
return kube
|
|
|
|
|
|
@pytest.fixture
|
|
def credentialtype_ssh():
|
|
ssh = CredentialType.defaults['ssh']()
|
|
ssh.save()
|
|
return ssh
|
|
|
|
|
|
@pytest.fixture
|
|
def credentialtype_aws():
|
|
aws = CredentialType.defaults['aws']()
|
|
aws.save()
|
|
return aws
|
|
|
|
|
|
@pytest.fixture
|
|
def credentialtype_net():
|
|
net = CredentialType.defaults['net']()
|
|
net.save()
|
|
return net
|
|
|
|
|
|
@pytest.fixture
|
|
def credentialtype_vault():
|
|
vault_type = CredentialType.defaults['vault']()
|
|
vault_type.save()
|
|
return vault_type
|
|
|
|
|
|
@pytest.fixture
|
|
def credentialtype_scm():
|
|
scm_type = CredentialType.defaults['scm']()
|
|
scm_type.save()
|
|
return scm_type
|
|
|
|
|
|
@pytest.fixture
|
|
def credentialtype_insights():
|
|
insights_type = CredentialType.defaults['insights']()
|
|
insights_type.save()
|
|
return insights_type
|
|
|
|
|
|
@pytest.fixture
|
|
def credentialtype_external():
|
|
external_type_inputs = {
|
|
'fields': [{
|
|
'id': 'url',
|
|
'label': 'Server URL',
|
|
'type': 'string',
|
|
'help_text': 'The server url.'
|
|
}, {
|
|
'id': 'token',
|
|
'label': 'Token',
|
|
'type': 'string',
|
|
'secret': True,
|
|
'help_text': 'An access token for the server.'
|
|
}],
|
|
'metadata': [{
|
|
'id': 'key',
|
|
'label': 'Key',
|
|
'type': 'string'
|
|
}, {
|
|
'id': 'version',
|
|
'label': 'Version',
|
|
'type': 'string'
|
|
}],
|
|
'required': ['url', 'token', 'key'],
|
|
}
|
|
|
|
class MockPlugin(object):
|
|
def backend(self, **kwargs):
|
|
return 'secret'
|
|
|
|
with mock.patch('awx.main.models.credential.CredentialType.plugin', new_callable=PropertyMock) as mock_plugin:
|
|
mock_plugin.return_value = MockPlugin()
|
|
external_type = CredentialType(
|
|
kind='external',
|
|
managed_by_tower=True,
|
|
name='External Service',
|
|
inputs=external_type_inputs
|
|
)
|
|
external_type.save()
|
|
yield external_type
|
|
|
|
|
|
@pytest.fixture
|
|
def credential(credentialtype_aws):
|
|
return Credential.objects.create(credential_type=credentialtype_aws, name='test-cred',
|
|
inputs={'username': 'something', 'password': 'secret'})
|
|
|
|
|
|
@pytest.fixture
|
|
def net_credential(credentialtype_net):
|
|
return Credential.objects.create(credential_type=credentialtype_net, name='test-cred',
|
|
inputs={'username': 'something', 'password': 'secret'})
|
|
|
|
|
|
@pytest.fixture
|
|
def vault_credential(credentialtype_vault):
|
|
return Credential.objects.create(credential_type=credentialtype_vault, name='test-cred',
|
|
inputs={'vault_password': 'secret'})
|
|
|
|
|
|
@pytest.fixture
|
|
def machine_credential(credentialtype_ssh):
|
|
return Credential.objects.create(credential_type=credentialtype_ssh, name='machine-cred',
|
|
inputs={'username': 'test_user', 'password': 'pas4word'})
|
|
|
|
|
|
@pytest.fixture
|
|
def scm_credential(credentialtype_scm):
|
|
return Credential.objects.create(credential_type=credentialtype_scm, name='scm-cred',
|
|
inputs={'username': 'optimus', 'password': 'prime'})
|
|
|
|
|
|
@pytest.fixture
|
|
def insights_credential(credentialtype_insights):
|
|
return Credential.objects.create(credential_type=credentialtype_insights, name='insights-cred',
|
|
inputs={'username': 'morocco_mole', 'password': 'secret_squirrel'})
|
|
|
|
|
|
@pytest.fixture
|
|
def org_credential(organization, credentialtype_aws):
|
|
return Credential.objects.create(credential_type=credentialtype_aws, name='test-cred',
|
|
inputs={'username': 'something', 'password': 'secret'},
|
|
organization=organization)
|
|
|
|
|
|
@pytest.fixture
|
|
def external_credential(credentialtype_external):
|
|
return Credential.objects.create(credential_type=credentialtype_external, name='external-cred',
|
|
inputs={'url': 'http://testhost.com', 'token': 'secret1'})
|
|
|
|
|
|
@pytest.fixture
|
|
def other_external_credential(credentialtype_external):
|
|
return Credential.objects.create(credential_type=credentialtype_external, name='other-external-cred',
|
|
inputs={'url': 'http://testhost.com', 'token': 'secret2'})
|
|
|
|
|
|
@pytest.fixture
|
|
def kube_credential(credentialtype_kube):
|
|
return Credential.objects.create(credential_type=credentialtype_kube, name='kube-cred',
|
|
inputs={'host': 'my.cluster', 'bearer_token': 'my-token', 'verify_ssl': False})
|
|
|
|
|
|
@pytest.fixture
|
|
def inventory(organization):
|
|
return organization.inventories.create(name="test-inv")
|
|
|
|
|
|
@pytest.fixture
|
|
def insights_inventory(inventory):
|
|
inventory.scm_type = 'insights'
|
|
inventory.save()
|
|
return inventory
|
|
|
|
|
|
@pytest.fixture
|
|
def scm_inventory_source(inventory, project):
|
|
inv_src = InventorySource(
|
|
name="test-scm-inv",
|
|
source_project=project,
|
|
source='scm',
|
|
source_path='inventory_file',
|
|
update_on_project_update=True,
|
|
inventory=inventory,
|
|
scm_last_revision=project.scm_revision)
|
|
with mock.patch('awx.main.models.unified_jobs.UnifiedJobTemplate.update'):
|
|
inv_src.save()
|
|
return inv_src
|
|
|
|
|
|
@pytest.fixture
|
|
def inventory_factory(organization):
|
|
def factory(name, org=organization):
|
|
try:
|
|
inv = Inventory.objects.get(name=name, organization=org)
|
|
except Inventory.DoesNotExist:
|
|
inv = Inventory.objects.create(name=name, organization=org)
|
|
return inv
|
|
return factory
|
|
|
|
|
|
@pytest.fixture
|
|
def label(organization):
|
|
return organization.labels.create(name="test-label", description="test-label-desc")
|
|
|
|
|
|
@pytest.fixture
|
|
def notification_template(organization):
|
|
return NotificationTemplate.objects.create(name='test-notification_template',
|
|
organization=organization,
|
|
notification_type="webhook",
|
|
notification_configuration=dict(url="http://localhost",
|
|
username="",
|
|
password="",
|
|
headers={"Test": "Header",}))
|
|
|
|
|
|
@pytest.fixture
|
|
def notification_template_with_encrypt(organization):
|
|
return NotificationTemplate.objects.create(name='test-notification_template_with_encrypt',
|
|
organization=organization,
|
|
notification_type="slack",
|
|
notification_configuration=dict(channels=["Foo", "Bar"],
|
|
token="token"))
|
|
|
|
|
|
@pytest.fixture
|
|
def notification(notification_template):
|
|
return Notification.objects.create(notification_template=notification_template,
|
|
status='successful',
|
|
notifications_sent=1,
|
|
notification_type='email',
|
|
recipients='admin@redhat.com',
|
|
subject='email subject')
|
|
|
|
|
|
@pytest.fixture
|
|
def job_template_with_survey_passwords(job_template_with_survey_passwords_factory):
|
|
return job_template_with_survey_passwords_factory(persisted=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def admin(user):
|
|
return user('admin', True)
|
|
|
|
|
|
@pytest.fixture
|
|
def system_auditor(user):
|
|
u = user('an-auditor', False)
|
|
Role.singleton('system_auditor').members.add(u)
|
|
return u
|
|
|
|
|
|
@pytest.fixture
|
|
def alice(user):
|
|
return user('alice', False)
|
|
|
|
|
|
@pytest.fixture
|
|
def bob(user):
|
|
return user('bob', False)
|
|
|
|
|
|
@pytest.fixture
|
|
def rando(user):
|
|
"Rando, the random user that doesn't have access to anything"
|
|
return user('rando', False)
|
|
|
|
|
|
@pytest.fixture
|
|
def org_admin(user, organization):
|
|
ret = user('org-admin', False)
|
|
organization.admin_role.members.add(ret)
|
|
organization.member_role.members.add(ret)
|
|
return ret
|
|
|
|
|
|
@pytest.fixture
|
|
def org_auditor(user, organization):
|
|
ret = user('org-auditor', False)
|
|
organization.auditor_role.members.add(ret)
|
|
organization.member_role.members.add(ret)
|
|
return ret
|
|
|
|
|
|
@pytest.fixture
|
|
def org_member(user, organization):
|
|
ret = user('org-member', False)
|
|
organization.member_role.members.add(ret)
|
|
return ret
|
|
|
|
|
|
@pytest.fixture
|
|
def organizations(instance):
|
|
def rf(organization_count=1):
|
|
orgs = []
|
|
for i in range(0, organization_count):
|
|
o = Organization.objects.create(name="test-org-%d" % i, description="test-org-desc")
|
|
orgs.append(o)
|
|
return orgs
|
|
return rf
|
|
|
|
|
|
@pytest.fixture
|
|
def group_factory(inventory):
|
|
def g(name):
|
|
try:
|
|
return Group.objects.get(name=name, inventory=inventory)
|
|
except Exception:
|
|
return Group.objects.create(inventory=inventory, name=name)
|
|
return g
|
|
|
|
|
|
@pytest.fixture
|
|
def hosts(group_factory):
|
|
group1 = group_factory('group-1')
|
|
|
|
def rf(host_count=1):
|
|
hosts = []
|
|
for i in range(0, host_count):
|
|
name = '%s-host-%s' % (group1.name, i)
|
|
(host, created) = group1.inventory.hosts.get_or_create(name=name)
|
|
if created:
|
|
group1.hosts.add(host)
|
|
hosts.append(host)
|
|
return hosts
|
|
return rf
|
|
|
|
|
|
@pytest.fixture
|
|
def group(inventory):
|
|
return inventory.groups.create(name='single-group')
|
|
|
|
|
|
@pytest.fixture
|
|
def inventory_source(inventory):
|
|
# by making it ec2, the credential is not required
|
|
return InventorySource.objects.create(name='single-inv-src',
|
|
inventory=inventory, source='ec2')
|
|
|
|
|
|
@pytest.fixture
|
|
def inventory_source_factory(inventory_factory):
|
|
def invsrc(name, source=None, inventory=None):
|
|
if inventory is None:
|
|
inventory = inventory_factory("inv-is-%s" % name)
|
|
if source is None:
|
|
source = 'file'
|
|
try:
|
|
return inventory.inventory_sources.get(name=name)
|
|
except Exception:
|
|
return inventory.inventory_sources.create(name=name, source=source)
|
|
return invsrc
|
|
|
|
|
|
@pytest.fixture
|
|
def inventory_update(inventory_source):
|
|
return InventoryUpdate.objects.create(
|
|
inventory_source=inventory_source,
|
|
source=inventory_source.source
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def inventory_script(organization):
|
|
return CustomInventoryScript.objects.create(name='test inv script',
|
|
organization=organization,
|
|
script='#!/usr/bin/python')
|
|
|
|
|
|
@pytest.fixture
|
|
def host(group, inventory):
|
|
return group.hosts.create(name='single-host', inventory=inventory)
|
|
|
|
|
|
@pytest.fixture
|
|
def permissions():
|
|
return {
|
|
'admin':{'create':True, 'read':True, 'write':True,
|
|
'update':True, 'delete':True, 'scm_update':True, 'execute':True, 'use':True,},
|
|
|
|
'auditor':{'read':True, 'create':False, 'write':False,
|
|
'update':False, 'delete':False, 'scm_update':False, 'execute':False, 'use':False,},
|
|
|
|
'usage':{'read':False, 'create':False, 'write':False,
|
|
'update':False, 'delete':False, 'scm_update':False, 'execute':False, 'use':True,},
|
|
}
|
|
|
|
|
|
def _request(verb):
|
|
def rf(url, data_or_user=None, user=None, middleware=None, expect=None, **kwargs):
|
|
if type(data_or_user) is User and user is None:
|
|
user = data_or_user
|
|
elif 'data' not in kwargs:
|
|
kwargs['data'] = data_or_user
|
|
if 'format' not in kwargs and 'content_type' not in kwargs:
|
|
kwargs['format'] = 'json'
|
|
|
|
request = getattr(APIRequestFactory(), verb)(url, **kwargs)
|
|
request_error = None
|
|
try:
|
|
view, view_args, view_kwargs = resolve(urllib.parse.urlparse(url)[2])
|
|
except Http404 as e:
|
|
request_error = e
|
|
if isinstance(kwargs.get('cookies', None), dict):
|
|
for key, value in kwargs['cookies'].items():
|
|
request.COOKIES[key] = value
|
|
if middleware:
|
|
middleware.process_request(request)
|
|
if user:
|
|
force_authenticate(request, user=user)
|
|
|
|
if not request_error:
|
|
response = view(request, *view_args, **view_kwargs)
|
|
else:
|
|
response = response_for_exception(request, request_error)
|
|
if middleware:
|
|
middleware.process_response(request, response)
|
|
if expect:
|
|
if response.status_code != expect:
|
|
if getattr(response, 'data', None):
|
|
try:
|
|
data_copy = response.data.copy()
|
|
# Make translated strings printable
|
|
for key, value in response.data.items():
|
|
if isinstance(value, list):
|
|
response.data[key] = []
|
|
for item in value:
|
|
response.data[key].append(str(item))
|
|
else:
|
|
response.data[key] = str(value)
|
|
except Exception:
|
|
response.data = data_copy
|
|
assert response.status_code == expect, 'Response data: {}'.format(
|
|
getattr(response, 'data', None)
|
|
)
|
|
if hasattr(response, 'render'):
|
|
response.render()
|
|
__SWAGGER_REQUESTS__.setdefault(request.path, {})[
|
|
(request.method.lower(), response.status_code)
|
|
] = (response.get('Content-Type', None), response.content, kwargs.get('data'))
|
|
return response
|
|
return rf
|
|
|
|
|
|
@pytest.fixture
|
|
def post():
|
|
return _request('post')
|
|
|
|
|
|
@pytest.fixture
|
|
def get():
|
|
return _request('get')
|
|
|
|
|
|
@pytest.fixture
|
|
def put():
|
|
return _request('put')
|
|
|
|
|
|
@pytest.fixture
|
|
def patch():
|
|
return _request('patch')
|
|
|
|
|
|
@pytest.fixture
|
|
def delete():
|
|
return _request('delete')
|
|
|
|
|
|
@pytest.fixture
|
|
def head():
|
|
return _request('head')
|
|
|
|
|
|
@pytest.fixture
|
|
def options():
|
|
return _request('options')
|
|
|
|
|
|
@pytest.fixture
|
|
def ad_hoc_command_factory(inventory, machine_credential, admin):
|
|
def factory(inventory=inventory, credential=machine_credential, initial_state='new', created_by=admin):
|
|
adhoc = AdHocCommand(
|
|
name='test-adhoc', inventory=inventory, credential=credential,
|
|
status=initial_state, created_by=created_by
|
|
)
|
|
adhoc.save()
|
|
return adhoc
|
|
return factory
|
|
|
|
|
|
@pytest.fixture
|
|
def job_template():
|
|
return JobTemplate.objects.create(name='test-job_template')
|
|
|
|
|
|
@pytest.fixture
|
|
def job_template_labels(organization, job_template):
|
|
job_template.labels.create(name="label-1", organization=organization)
|
|
job_template.labels.create(name="label-2", organization=organization)
|
|
|
|
return job_template
|
|
|
|
|
|
@pytest.fixture
|
|
def jt_linked(organization, project, inventory, machine_credential, credential, net_credential, vault_credential):
|
|
'''
|
|
A job template with a reasonably complete set of related objects to
|
|
test RBAC and other functionality affected by related objects
|
|
'''
|
|
jt = JobTemplate.objects.create(
|
|
project=project, inventory=inventory, playbook='helloworld.yml',
|
|
organization=organization
|
|
)
|
|
jt.credentials.add(machine_credential, vault_credential, credential, net_credential)
|
|
return jt
|
|
|
|
|
|
@pytest.fixture
|
|
def workflow_job_template(organization):
|
|
wjt = WorkflowJobTemplate(name='test-workflow_job_template', organization=organization)
|
|
wjt.save()
|
|
|
|
return wjt
|
|
|
|
|
|
@pytest.fixture
|
|
def workflow_job_factory(workflow_job_template, admin):
|
|
def factory(workflow_job_template=workflow_job_template, initial_state='new', created_by=admin):
|
|
return workflow_job_template.create_unified_job(_eager_fields={
|
|
'status': initial_state, 'created_by': created_by})
|
|
return factory
|
|
|
|
|
|
@pytest.fixture
|
|
def system_job_template():
|
|
sys_jt = SystemJobTemplate(name='test-system_job_template', job_type='cleanup_jobs')
|
|
sys_jt.save()
|
|
return sys_jt
|
|
|
|
|
|
@pytest.fixture
|
|
def system_job_factory(system_job_template, admin):
|
|
def factory(system_job_template=system_job_template, initial_state='new', created_by=admin):
|
|
return system_job_template.create_unified_job(_eager_fields={
|
|
'status': initial_state, 'created_by': created_by})
|
|
return factory
|
|
|
|
|
|
def dumps(value):
|
|
return DjangoJSONEncoder().encode(value)
|
|
|
|
|
|
# Taken from https://github.com/django-extensions/django-extensions/blob/54fe88df801d289882a79824be92d823ab7be33e/django_extensions/db/fields/json.py
|
|
def get_db_prep_save(self, value, connection, **kwargs):
|
|
"""Convert our JSON object to a string before we save"""
|
|
if value is None and self.null:
|
|
return None
|
|
# default values come in as strings; only non-strings should be
|
|
# run through `dumps`
|
|
if not isinstance(value, str):
|
|
value = dumps(value)
|
|
|
|
return value
|
|
|
|
|
|
@pytest.fixture
|
|
def monkeypatch_jsonbfield_get_db_prep_save(mocker):
|
|
JSONBField.get_db_prep_save = get_db_prep_save
|
|
|
|
|
|
@pytest.fixture
|
|
def oauth_application(admin):
|
|
return Application.objects.create(
|
|
name='test app', user=admin, client_type='confidential',
|
|
authorization_grant_type='password'
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
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)
|
|
|
|
setattr(SQLiteCursorWrapper, 'copy_expert', write_stdout)
|
|
request.addfinalizer(lambda: shutil.rmtree(path))
|
|
request.addfinalizer(lambda: delattr(SQLiteCursorWrapper, 'copy_expert'))
|
|
return path
|
|
|
|
|
|
@pytest.fixture
|
|
def disable_database_settings(mocker):
|
|
m = mocker.patch('awx.conf.settings.SettingsWrapper.all_supported_settings', new_callable=PropertyMock)
|
|
m.return_value = []
|
|
|
|
|
|
@pytest.fixture
|
|
def slice_jt_factory(inventory):
|
|
def r(N, jt_kwargs=None):
|
|
for i in range(N):
|
|
inventory.hosts.create(name='foo{}'.format(i))
|
|
if not jt_kwargs:
|
|
jt_kwargs = {}
|
|
return JobTemplate.objects.create(
|
|
name='slice-jt-from-factory',
|
|
job_slice_count=N,
|
|
inventory=inventory,
|
|
**jt_kwargs
|
|
)
|
|
return r
|
|
|
|
|
|
@pytest.fixture
|
|
def slice_job_factory(slice_jt_factory):
|
|
def r(N, jt_kwargs=None, prompts=None, spawn=False):
|
|
slice_jt = slice_jt_factory(N, jt_kwargs=jt_kwargs)
|
|
if not prompts:
|
|
prompts = {}
|
|
slice_job = slice_jt.create_unified_job(**prompts)
|
|
if spawn:
|
|
for node in slice_job.workflow_nodes.all():
|
|
# does what the task manager does for spawning workflow jobs
|
|
kv = node.get_job_kwargs()
|
|
job = node.unified_job_template.create_unified_job(**kv)
|
|
node.job = job
|
|
node.save()
|
|
return slice_job
|
|
return r
|