mirror of
https://github.com/ZwareBear/awx.git
synced 2026-05-04 07:51:58 -05:00
Fix some notifications issues and write some tests
* Fixes some notifier merging issues * Fixes some more unicode problems * Implements unit tests
This commit is contained in:
Matthew Jones
160
awx/main/tests/functional/conftest.py
Normal file
160
awx/main/tests/functional/conftest.py
Normal file
@@ -0,0 +1,160 @@
|
||||
import pytest
|
||||
import mock
|
||||
|
||||
from django.core.urlresolvers import resolve
|
||||
from django.utils.six.moves.urllib.parse import urlparse
|
||||
|
||||
from awx.main.models.organization import Organization
|
||||
from awx.main.models.projects import Project
|
||||
from awx.main.models.ha import Instance
|
||||
from django.contrib.auth.models import User
|
||||
from rest_framework.test import (
|
||||
APIRequestFactory,
|
||||
force_authenticate,
|
||||
)
|
||||
|
||||
@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, password=name)
|
||||
user.save()
|
||||
return user
|
||||
return u
|
||||
|
||||
@pytest.fixture
|
||||
def post():
|
||||
def rf(url, data, user=None, middleware=None, **kwargs):
|
||||
view, view_args, view_kwargs = resolve(urlparse(url)[2])
|
||||
if 'format' not in kwargs:
|
||||
kwargs['format'] = 'json'
|
||||
request = APIRequestFactory().post(url, data, **kwargs)
|
||||
if middleware:
|
||||
middleware.process_request(request)
|
||||
if user:
|
||||
force_authenticate(request, user=user)
|
||||
response = view(request, *view_args, **view_kwargs)
|
||||
if middleware:
|
||||
middleware.process_response(request, response)
|
||||
return response
|
||||
return rf
|
||||
|
||||
@pytest.fixture
|
||||
def get():
|
||||
def rf(url, user=None, middleware=None, **kwargs):
|
||||
view, view_args, view_kwargs = resolve(urlparse(url)[2])
|
||||
if 'format' not in kwargs:
|
||||
kwargs['format'] = 'json'
|
||||
request = APIRequestFactory().get(url, **kwargs)
|
||||
if middleware:
|
||||
middleware.process_request(request)
|
||||
if user:
|
||||
force_authenticate(request, user=user)
|
||||
response = view(request, *view_args, **view_kwargs)
|
||||
if middleware:
|
||||
middleware.process_response(request, response)
|
||||
return response
|
||||
return rf
|
||||
|
||||
@pytest.fixture
|
||||
def put():
|
||||
def rf(url, data, user=None, middleware=None, **kwargs):
|
||||
view, view_args, view_kwargs = resolve(urlparse(url)[2])
|
||||
if 'format' not in kwargs:
|
||||
kwargs['format'] = 'json'
|
||||
request = APIRequestFactory().put(url, data, **kwargs)
|
||||
if middleware:
|
||||
middleware.process_request(request)
|
||||
if user:
|
||||
force_authenticate(request, user=user)
|
||||
response = view(request, *view_args, **view_kwargs)
|
||||
if middleware:
|
||||
middleware.process_response(request, response)
|
||||
return response
|
||||
return rf
|
||||
|
||||
@pytest.fixture
|
||||
def patch():
|
||||
def rf(url, data, user=None, middleware=None, **kwargs):
|
||||
view, view_args, view_kwargs = resolve(urlparse(url)[2])
|
||||
if 'format' not in kwargs:
|
||||
kwargs['format'] = 'json'
|
||||
request = APIRequestFactory().patch(url, data, **kwargs)
|
||||
if middleware:
|
||||
middleware.process_request(request)
|
||||
if user:
|
||||
force_authenticate(request, user=user)
|
||||
response = view(request, *view_args, **view_kwargs)
|
||||
if middleware:
|
||||
middleware.process_response(request, response)
|
||||
return response
|
||||
return rf
|
||||
|
||||
@pytest.fixture
|
||||
def delete():
|
||||
def rf(url, user=None, middleware=None, **kwargs):
|
||||
view, view_args, view_kwargs = resolve(urlparse(url)[2])
|
||||
if 'format' not in kwargs:
|
||||
kwargs['format'] = 'json'
|
||||
request = APIRequestFactory().delete(url, **kwargs)
|
||||
if middleware:
|
||||
middleware.process_request(request)
|
||||
if user:
|
||||
force_authenticate(request, user=user)
|
||||
response = view(request, *view_args, **view_kwargs)
|
||||
if middleware:
|
||||
middleware.process_response(request, response)
|
||||
return response
|
||||
return rf
|
||||
|
||||
@pytest.fixture
|
||||
def head():
|
||||
def rf(url, user=None, middleware=None, **kwargs):
|
||||
view, view_args, view_kwargs = resolve(urlparse(url)[2])
|
||||
if 'format' not in kwargs:
|
||||
kwargs['format'] = 'json'
|
||||
request = APIRequestFactory().head(url, **kwargs)
|
||||
if middleware:
|
||||
middleware.process_request(request)
|
||||
if user:
|
||||
force_authenticate(request, user=user)
|
||||
response = view(request, *view_args, **view_kwargs)
|
||||
if middleware:
|
||||
middleware.process_response(request, response)
|
||||
return response
|
||||
return rf
|
||||
|
||||
@pytest.fixture
|
||||
def options():
|
||||
def rf(url, data, user=None, middleware=None, **kwargs):
|
||||
view, view_args, view_kwargs = resolve(urlparse(url)[2])
|
||||
if 'format' not in kwargs:
|
||||
kwargs['format'] = 'json'
|
||||
request = APIRequestFactory().options(url, data, **kwargs)
|
||||
if middleware:
|
||||
middleware.process_request(request)
|
||||
if user:
|
||||
force_authenticate(request, user=user)
|
||||
response = view(request, *view_args, **view_kwargs)
|
||||
if middleware:
|
||||
middleware.process_response(request, response)
|
||||
return response
|
||||
return rf
|
||||
|
||||
@pytest.fixture
|
||||
def instance(settings):
|
||||
return Instance.objects.create(uuid=settings.SYSTEM_UUID, primary=True, hostname="instance.example.org")
|
||||
|
||||
@pytest.fixture
|
||||
def organization(instance):
|
||||
return Organization.objects.create(name="test-org", description="test-org-desc")
|
||||
|
||||
@pytest.fixture
|
||||
@mock.patch.object(Project, "update", lambda self, **kwargs: None)
|
||||
def project(instance):
|
||||
return Project.objects.create(name="test-proj",
|
||||
description="test-proj-desc",
|
||||
scm_type="git",
|
||||
scm_url="https://github.com/jlaska/ansible-playbooks")
|
||||
124
awx/main/tests/functional/test_notifications.py
Normal file
124
awx/main/tests/functional/test_notifications.py
Normal file
@@ -0,0 +1,124 @@
|
||||
import mock
|
||||
import pytest
|
||||
|
||||
from awx.main.models.notifications import Notification, Notifier
|
||||
from awx.main.models.inventory import Inventory, Group
|
||||
from awx.main.models.organization import Organization
|
||||
from awx.main.models.projects import Project
|
||||
from awx.main.models.jobs import JobTemplate
|
||||
|
||||
from django.core.urlresolvers import reverse
|
||||
from django.core.mail.message import EmailMessage
|
||||
|
||||
@pytest.fixture
|
||||
def notifier():
|
||||
return Notifier.objects.create(name="test-notification",
|
||||
notification_type="webhook",
|
||||
notification_configuration=dict(url="http://localhost",
|
||||
headers={"Test": "Header"}))
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_get_notifier_list(get, user, notifier):
|
||||
url = reverse('api:notifier_list')
|
||||
response = get(url, user('admin', True))
|
||||
assert response.status_code == 200
|
||||
assert len(response.data['results']) == 1
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_basic_parameterization(get, post, user, organization):
|
||||
u = user('admin-poster', True)
|
||||
url = reverse('api:notifier_list')
|
||||
response = post(url,
|
||||
dict(name="test-webhook",
|
||||
description="test webhook",
|
||||
organization=1,
|
||||
notification_type="webhook",
|
||||
notification_configuration=dict(url="http://localhost",
|
||||
headers={"Test": "Header"})),
|
||||
u)
|
||||
assert response.status_code == 201
|
||||
url = reverse('api:notifier_detail', args=(response.data['id'],))
|
||||
response = get(url, u)
|
||||
assert 'related' in response.data
|
||||
assert 'organization' in response.data['related']
|
||||
assert 'summary_fields' in response.data
|
||||
assert 'organization' in response.data['summary_fields']
|
||||
assert 'notifications' in response.data['related']
|
||||
assert 'notification_configuration' in response.data
|
||||
assert 'url' in response.data['notification_configuration']
|
||||
assert 'headers' in response.data['notification_configuration']
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_encrypted_subfields(get, post, user, organization):
|
||||
def assert_send(self, messages):
|
||||
assert self.account_token == "shouldhide"
|
||||
return 1
|
||||
u = user('admin-poster', True)
|
||||
url = reverse('api:notifier_list')
|
||||
response = post(url,
|
||||
dict(name="test-twilio",
|
||||
description="test twilio",
|
||||
organization=1,
|
||||
notification_type="twilio",
|
||||
notification_configuration=dict(account_sid="dummy",
|
||||
account_token="shouldhide",
|
||||
from_number="+19999999999",
|
||||
to_numbers=["9998887777"])),
|
||||
u)
|
||||
assert response.status_code == 201
|
||||
notifier_actual = Notifier.objects.get(id=response.data['id'])
|
||||
assert notifier_actual.notification_configuration['account_token'].startswith("$encrypted$")
|
||||
url = reverse('api:notifier_detail', args=(response.data['id'],))
|
||||
response = get(url, u)
|
||||
assert response.data['notification_configuration']['account_token'] == "$encrypted$"
|
||||
with mock.patch.object(notifier_actual.notification_class, "send_messages", assert_send):
|
||||
notifier_actual.send("Test", {'body': "Test"})
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_inherited_notifiers(get, post, user, organization, project):
|
||||
u = user('admin-poster', True)
|
||||
url = reverse('api:notifier_list')
|
||||
notifiers = []
|
||||
for nfiers in xrange(3):
|
||||
response = post(url,
|
||||
dict(name="test-webhook-{}".format(nfiers),
|
||||
description="test webhook {}".format(nfiers),
|
||||
organization=1,
|
||||
notification_type="webhook",
|
||||
notification_configuration=dict(url="http://localhost",
|
||||
headers={"Test": "Header"})),
|
||||
u)
|
||||
assert response.status_code == 201
|
||||
notifiers.append(response.data['id'])
|
||||
o = Organization.objects.get(id=1)
|
||||
p = Project.objects.get(id=1)
|
||||
o.projects.add(p)
|
||||
i = Inventory.objects.create(name='test', organization=o)
|
||||
i.save()
|
||||
g = Group.objects.create(name='test', inventory=i)
|
||||
g.save()
|
||||
jt = JobTemplate.objects.create(name='test', inventory=i, project=p, playbook='debug.yml')
|
||||
jt.save()
|
||||
url = reverse('api:organization_notifiers_any_list', args=(1,))
|
||||
response = post(url, dict(id=notifiers[0]), u)
|
||||
assert response.status_code == 204
|
||||
url = reverse('api:project_notifiers_any_list', args=(1,))
|
||||
response = post(url, dict(id=notifiers[1]), u)
|
||||
assert response.status_code == 204
|
||||
url = reverse('api:job_template_notifiers_any_list', args=(jt.id,))
|
||||
response = post(url, dict(id=notifiers[2]), u)
|
||||
assert response.status_code == 204
|
||||
assert len(jt.notifiers['any']) == 3
|
||||
assert len(p.notifiers['any']) == 2
|
||||
assert len(g.inventory_source.notifiers['any']) == 1
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_notifier_merging(get, post, user, organization, project, notifier):
|
||||
u = user('admin-poster', True)
|
||||
o = Organization.objects.get(id=1)
|
||||
p = Project.objects.get(id=1)
|
||||
n = Notifier.objects.get(id=1)
|
||||
o.projects.add(p)
|
||||
o.notifiers_any.add(n)
|
||||
p.notifiers_any.add(n)
|
||||
assert len(p.notifiers['any']) == 1
|
||||
Reference in New Issue
Block a user