Merge pull request #815 from wwitzel3/tests-activity-stream

refactor activity_stream tests

Makefile

@@ -363,7 +363,7 @@ test_unit:
 
 # Run all API unit tests with coverage enabled.
 test_coverage:
-	py.test --cov=awx --cov-report=xml --junitxml=./reports/junit.xml awx/main/tests awx/api/tests awx/fact/tests
+	py.test --create-db --cov=awx --cov-report=xml --junitxml=./reports/junit.xml awx/main/tests awx/api/tests awx/fact/tests
 
 # Output test coverage as HTML (into htmlcov directory).
 coverage_html:

awx/main/tests/functional/conftest.py

@@ -0,0 +1,56 @@
+import pytest
+
+from awx.main.models.organization import Organization
+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(_cls, _user, _url, pk=None, kwargs={}, middleware=None):
+        view = _cls.as_view()
+        request = APIRequestFactory().post(_url, kwargs, format='json')
+        if middleware:
+            middleware.process_request(request)
+        force_authenticate(request, user=_user)
+        response = view(request, pk=pk)
+        if middleware:
+            middleware.process_response(request, response)
+        return response
+    return rf
+
+@pytest.fixture
+def get():
+    def rf(_cls, _user, _url, pk=None, middleware=None):
+        view = _cls.as_view()
+        request = APIRequestFactory().get(_url, format='json')
+        if middleware:
+            middleware.process_request(request)
+        force_authenticate(request, user=_user)
+        response = view(request, pk=pk)
+        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")

awx/main/tests/functional/test_activity_streams.py

@@ -0,0 +1,61 @@
+import mock
+import pytest
+
+from awx.api.views import (
+    ActivityStreamList,
+    ActivityStreamDetail,
+    OrganizationList,
+)
+from awx.main.middleware import ActivityStreamMiddleware
+from awx.main.models.activity_stream import ActivityStream
+from django.core.urlresolvers import reverse
+
+def mock_feature_enabled(feature, bypass_database=None):
+    return True
+
+@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
+@pytest.mark.django_db
+def test_get_activity_stream_list(monkeypatch, organization, get, user):
+    url = reverse('api:activity_stream_list')
+    response = get(ActivityStreamList, user('admin', True), url)
+
+    assert response.status_code == 200
+
+@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
+@pytest.mark.django_db
+def test_basic_fields(monkeypatch, organization, get, user):
+    u = user('admin', True)
+    activity_stream = ActivityStream.objects.latest('pk')
+    activity_stream.actor = u
+    activity_stream.save()
+
+    aspk = activity_stream.pk
+    url = reverse('api:activity_stream_detail', args=(aspk,))
+    response = get(ActivityStreamDetail, user('admin', True), url, pk=aspk)
+
+    assert response.status_code == 200
+    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 response.data['summary_fields']['organization'][0]['name'] == 'test-org'
+
+@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
+@pytest.mark.django_db
+def test_middleware_actor_added(monkeypatch, post, get, user):
+    u = user('admin-poster', True)
+
+    url = reverse('api:organization_list')
+    response = post(OrganizationList, u, url,
+                    kwargs=dict(name='test-org', description='test-desc'),
+                    middleware=ActivityStreamMiddleware())
+    assert response.status_code == 201
+
+    org_id = response.data['id']
+    activity_stream = ActivityStream.objects.filter(organization__pk=org_id).first()
+
+    url = reverse('api:activity_stream_detail', args=(activity_stream.pk,))
+    response = get(ActivityStreamDetail, u, url, pk=activity_stream.pk)
+
+    assert response.status_code == 200
+    assert response.data['summary_fields']['actor']['username'] == 'admin-poster'

awx/main/tests/old/activity_stream.py

@@ -1,55 +0,0 @@
-# Copyright (c) 2015 Ansible, Inc.
-# All Rights Reserved.
-
-# Python
-from django.core.urlresolvers import reverse
-
-# AWX
-from awx.main.models import * # noqa
-from awx.main.tests.base import BaseTest
-
-class ActivityStreamTest(BaseTest):
-
-    def collection(self):
-        return reverse('api:activity_stream_list')
-
-    def item(self, item_id):
-        return reverse('api:activity_stream_detail', args=(item_id,))
-
-    def setUp(self):
-        super(ActivityStreamTest, self).setUp()
-        self.setup_instances()
-        self.create_test_license_file()
-        # TODO: Test non-enterprise license
-        self.setup_users()
-        self.org_created = self.post(reverse('api:organization_list'), dict(name='test org', description='test descr'), expect=201, auth=self.get_super_credentials())
-
-    def test_get_activity_stream_list(self):
-        url = self.collection()
-
-        with self.current_user(self.super_django_user):
-            self.options(url, expect=200)
-            self.head(url, expect=200)
-            response = self.get(url, expect=200)
-            self.check_pagination_and_size(response, 1, previous=None, next=None)
-
-    def test_basic_fields(self):
-        item_id = ActivityStream.objects.order_by('pk')[0].pk
-        org_item = self.item(item_id)
-
-        with self.current_user(self.super_django_user):
-            response = self.get(org_item, expect=200)
-
-            self.assertTrue("related" in response)
-            self.assertTrue("organization" in response['related'])
-            self.assertTrue("summary_fields" in response)
-            self.assertTrue("organization" in response['summary_fields'])
-            self.assertTrue(response['summary_fields']['organization'][0]['name'] == self.org_created['name'])
-
-    def test_changeby_user(self):
-        item_id = ActivityStream.objects.order_by('pk')[0].pk
-        org_item = self.item(item_id)
-
-        with self.current_user(self.super_django_user):
-            response = self.get(org_item, expect=200)
-            self.assertEqual(response['summary_fields']['actor']['username'], self.super_django_user.username)

pytest.ini

@@ -3,4 +3,4 @@ DJANGO_SETTINGS_MODULE = awx.settings.development
 python_paths = awx/lib/site-packages
 site_dirs = awx/lib/site-packages
 python_files = *.py
-addopts = --create-db
+addopts = --reuse-db