mirror of
https://github.com/ansible/awx.git
synced 2026-01-11 01:57:35 -03:30
c43dfde45a
* Create test for using a manual project * Chang default project factory to git, remove project files monkeypatch * skip update of factory project * Initial file scaffolding for feature * Fill in galaxy and names * Add README, describe project folders and dependencies
884 lines
25 KiB
Python
884 lines
25 KiB
Python
# Python
|
|
import pytest
|
|
from unittest import mock
|
|
import urllib.parse
|
|
from unittest.mock import PropertyMock
|
|
import importlib
|
|
|
|
# Django
|
|
from django.urls import resolve
|
|
from django.http import Http404
|
|
from django.apps import apps
|
|
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
|
|
|
|
from django.db.models.signals import post_migrate
|
|
|
|
from awx.main.migrations._dab_rbac import setup_managed_role_definitions
|
|
|
|
# AWX
|
|
from awx.main.models.projects import Project
|
|
from awx.main.models.ha import Instance, InstanceGroup
|
|
|
|
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
|
|
from awx.main.models.organization import (
|
|
Organization,
|
|
Team,
|
|
)
|
|
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.execution_environments import ExecutionEnvironment
|
|
from awx.main.utils import is_testing
|
|
|
|
__SWAGGER_REQUESTS__ = {}
|
|
|
|
|
|
# HACK: the dab_resource_registry app required ServiceID in migrations which checks do not run
|
|
dab_rr_initial = importlib.import_module('ansible_base.resource_registry.migrations.0001_initial')
|
|
|
|
|
|
if is_testing():
|
|
post_migrate.connect(lambda **kwargs: dab_rr_initial.create_service_id(apps, None))
|
|
|
|
|
|
@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 execution_environment():
|
|
return ExecutionEnvironment.objects.create(name="test-ee", description="test-ee", managed=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def setup_managed_roles():
|
|
"Run the migration script to pre-create managed role definitions"
|
|
setup_managed_role_definitions(apps, None)
|
|
|
|
|
|
@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
|
|
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(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(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():
|
|
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=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',
|
|
inventory=inventory,
|
|
)
|
|
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)
|
|
u.is_system_auditor = True
|
|
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 constructed_inventory(organization):
|
|
"""
|
|
creates a new constructed inventory source
|
|
"""
|
|
return Inventory.objects.create(name='dummy1', kind='constructed', organization=organization)
|
|
|
|
|
|
@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 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 instance_group():
|
|
return InstanceGroup.objects.create(name="east")
|
|
|
|
|
|
@pytest.fixture
|
|
def workflow_job_template(organization):
|
|
wjt = WorkflowJobTemplate.objects.create(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
|
|
|
|
|
|
@pytest.fixture
|
|
def wfjt(workflow_job_template_factory, organization):
|
|
objects = workflow_job_template_factory('test_workflow', organization=organization, persisted=True)
|
|
return objects.workflow_job_template
|
|
|
|
|
|
@pytest.fixture
|
|
def wfjt_with_nodes(workflow_job_template_factory, organization, job_template):
|
|
objects = workflow_job_template_factory(
|
|
'test_workflow', organization=organization, workflow_job_template_nodes=[{'unified_job_template': job_template}], persisted=True
|
|
)
|
|
return objects.workflow_job_template
|
|
|
|
|
|
@pytest.fixture
|
|
def wfjt_node(wfjt_with_nodes):
|
|
return wfjt_with_nodes.workflow_job_template_nodes.all()[0]
|
|
|
|
|
|
@pytest.fixture
|
|
def workflow_job(wfjt):
|
|
return wfjt.workflow_jobs.create(name='test_workflow')
|
|
|
|
|
|
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
|
|
|
|
|
|
class MockCopy:
|
|
events = []
|
|
index = -1
|
|
|
|
def __init__(self, sql):
|
|
self.events = []
|
|
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():
|
|
self.events.append(event.stdout)
|
|
|
|
def read(self):
|
|
self.index = self.index + 1
|
|
if self.index < len(self.events):
|
|
return memoryview(self.events[self.index].encode())
|
|
|
|
return None
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def sqlite_copy(request, mocker):
|
|
# copy is postgres-specific, and SQLite doesn't support it; mock its
|
|
# behavior to test that it writes a file that contains stdout from events
|
|
|
|
def write_stdout(self, sql):
|
|
mock_copy = MockCopy(sql)
|
|
return mock_copy
|
|
|
|
mocker.patch.object(SQLiteCursorWrapper, 'copy', write_stdout, create=True)
|
|
|
|
|
|
@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
|
|
|
|
|
|
@pytest.fixture
|
|
def control_plane_execution_environment():
|
|
return ExecutionEnvironment.objects.create(name="Control Plane EE", managed=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def default_job_execution_environment():
|
|
return ExecutionEnvironment.objects.create(name="Default Job EE", managed=False)
|