Merge pull request #196 from chrismeyersfsu/master

system tracking api merge conflicts resolved

awx/main/tests/__init__.py

@@ -16,3 +16,4 @@ from awx.main.tests.schedules import * # noqa
 from awx.main.tests.redact import * # noqa
 from awx.main.tests.views import * # noqa
 from awx.main.tests.commands import * # noqa
+from awx.main.tests.fact import * # noqa

awx/main/tests/base.py

@@ -25,9 +25,6 @@ from django.contrib.auth.models import User
 from django.test.client import Client
 from django.test.utils import override_settings
 
-# MongoEngine
-from mongoengine.connection import get_db, ConnectionError
-
 # AWX
 from awx.main.models import * # noqa
 from awx.main.backend import LDAPSettings
@@ -43,15 +40,6 @@ TEST_PLAYBOOK = '''- hosts: mygroup
     command: test 1 = 1
 '''
 
-class MongoDBRequired(django.test.TestCase):
-    def setUp(self):
-        # Drop mongo database
-        try:
-            self.db = get_db()
-            self.db.connection.drop_database(settings.MONGO_DB)
-        except ConnectionError:
-            self.skipTest('MongoDB connection failed')
-
 class QueueTestMixin(object):
     def start_queue(self):
         self.start_redis()

awx/main/tests/commands/cleanup_facts.py

@@ -10,7 +10,8 @@ import mock
 from django.core.management.base import CommandError
 
 # AWX
-from awx.main.tests.base import BaseTest, MongoDBRequired
+from awx.main.tests.base import BaseTest
+from awx.fact.tests.base import MongoDBRequired
 from awx.main.tests.commands.base import BaseCommandMixin
 from awx.main.management.commands.cleanup_facts import Command, CleanupFacts
 from awx.fact.models.fact import * # noqa

awx/main/tests/commands/run_fact_cache_receiver.py

@@ -10,7 +10,8 @@ from copy import deepcopy
 from mock import MagicMock
 
 # AWX
-from awx.main.tests.base import BaseTest, MongoDBRequired
+from awx.main.tests.base import BaseTest
+from awx.fact.tests.base import MongoDBRequired
 from awx.main.tests.commands.base import BaseCommandMixin
 from awx.main.management.commands.run_fact_cache_receiver import FactCacheReceiver
 from awx.fact.models.fact import * # noqa

awx/main/tests/fact/__init__.py

@@ -0,0 +1,6 @@
+# Copyright (c) 2015 Ansible, Inc.
+# All Rights Reserved
+
+from __future__ import absolute_import
+
+from .fact_api import * # noqa

awx/main/tests/fact/fact_api.py

@@ -0,0 +1,236 @@
+# Copyright (c) 2015 Ansible, Inc.
+# All Rights Reserved
+
+# Python
+
+# Django
+from django.core.urlresolvers import reverse
+
+# AWX
+from awx.main.utils import timestamp_apiformat
+from awx.main.models import * # noqa
+from awx.main.tests.base import BaseLiveServerTest
+from awx.fact.models import * # noqa
+from awx.fact.tests.base import BaseFactTestMixin, FactScanBuilder, TEST_FACT_ANSIBLE, TEST_FACT_PACKAGES, TEST_FACT_SERVICES
+from awx.main.utils import build_url
+
+__all__ = ['FactVersionApiTest', 'FactViewApiTest', 'SingleFactApiTest',]
+
+class FactApiBaseTest(BaseLiveServerTest, BaseFactTestMixin):
+    def setUp(self):
+        super(FactApiBaseTest, self).setUp()
+        self.setup_instances()
+        self.setup_users()
+        self.organization = self.make_organization(self.super_django_user)
+        self.organization.admins.add(self.normal_django_user)
+        self.inventory = self.organization.inventories.create(name='test-inventory', description='description for test-inventory')
+        self.host = self.inventory.hosts.create(name='host.example.com')
+        self.host2 = self.inventory.hosts.create(name='host2.example.com')
+        self.host3 = self.inventory.hosts.create(name='host3.example.com')
+
+    def setup_facts(self, scan_count):
+        self.builder = FactScanBuilder()
+        self.builder.add_fact('ansible', TEST_FACT_ANSIBLE)
+        self.builder.add_fact('packages', TEST_FACT_PACKAGES)
+        self.builder.add_fact('services', TEST_FACT_SERVICES)
+        self.builder.add_hostname('host.example.com')
+        self.builder.add_hostname('host2.example.com')
+        self.builder.add_hostname('host3.example.com')
+        self.builder.build(scan_count=scan_count, host_count=3)
+
+        self.fact_host = FactHost.objects.get(hostname=self.host.name)
+
+class FactVersionApiTest(FactApiBaseTest):
+    def check_equal(self, fact_versions, results):
+        def find(element, set1):
+            for e in set1:
+                if all([ e.get(field) == element.get(field) for field in element.keys()]):
+                    return e
+            return None
+
+        self.assertEqual(len(results), len(fact_versions))
+        for v in fact_versions:
+            v_dict = {
+                'timestamp': timestamp_apiformat(v.timestamp),
+                'module': v.module
+            }
+            e = find(v_dict, results)
+            self.assertIsNotNone(e, "%s not found in %s" % (v_dict, results))
+
+    def get_list(self, fact_versions, params=None):
+        url = build_url('api:host_fact_versions_list', args=(self.host.pk,), get=params)
+        with self.current_user(self.super_django_user):
+            response = self.get(url, expect=200)
+
+        self.check_equal(fact_versions, response['results'])
+        return response
+
+    def test_permission_list(self):
+        url = reverse('api:host_fact_versions_list', args=(self.host.pk,))
+        with self.current_user('admin'):
+            self.get(url, expect=200)
+        with self.current_user('normal'):
+            self.get(url, expect=200)
+        with self.current_user('other'):
+            self.get(url, expect=403)
+        with self.current_user('nobody'):
+            self.get(url, expect=403)
+        with self.current_user(None):
+            self.get(url, expect=401)
+
+    def test_list_empty(self):
+        url = reverse('api:host_fact_versions_list', args=(self.host.pk,))
+        with self.current_user(self.super_django_user):
+            response = self.get(url, expect=200)
+            self.assertIn('results', response)
+            self.assertIsInstance(response['results'], list)
+            self.assertEqual(len(response['results']), 0)
+
+    def test_list_related_fact_view(self):
+        self.setup_facts(2)
+        url = reverse('api:host_fact_versions_list', args=(self.host.pk,))
+        with self.current_user(self.super_django_user):
+            response = self.get(url, expect=200)
+            for entry in response['results']:
+                self.assertIn('fact_view', entry['related'])
+                r = self.get(entry['related']['fact_view'], expect=200)
+
+    def test_list(self):
+        self.setup_facts(2)
+        self.get_list(FactVersion.objects.filter(host=self.fact_host))
+
+    def test_list_module(self):
+        self.setup_facts(10)
+        self.get_list(FactVersion.objects.filter(host=self.fact_host, module='packages'), dict(module='packages'))
+
+    def test_list_time_from(self):
+        self.setup_facts(10)
+
+        params = {
+            'from': timestamp_apiformat(self.builder.get_timestamp(1)),
+        }
+        #    'to': timestamp_apiformat(self.builder.get_timestamp(3))
+        fact_versions = FactVersion.objects.filter(host=self.fact_host, timestamp__gt=params['from'])
+        self.get_list(fact_versions, params)
+
+    def test_list_time_to(self):
+        self.setup_facts(10)
+
+        params = {
+            'to': timestamp_apiformat(self.builder.get_timestamp(3))
+        }
+        fact_versions = FactVersion.objects.filter(host=self.fact_host, timestamp__lte=params['to'])
+        self.get_list(fact_versions, params)
+
+    def test_list_time_from_to(self):
+        self.setup_facts(10)
+
+        params = {
+            'from': timestamp_apiformat(self.builder.get_timestamp(1)),
+            'to': timestamp_apiformat(self.builder.get_timestamp(3))
+        }
+        fact_versions = FactVersion.objects.filter(host=self.fact_host, timestamp__gt=params['from'], timestamp__lte=params['to'])
+        self.get_list(fact_versions, params)
+
+
+class FactViewApiTest(FactApiBaseTest):
+    def check_equal(self, fact_obj, results):
+        fact_dict = {
+            'timestamp': timestamp_apiformat(fact_obj.timestamp),
+            'module': fact_obj.module,
+            'host': {
+                'hostname': fact_obj.host.hostname,
+                'id': str(fact_obj.host.id)
+            },
+            'fact': fact_obj.fact
+        }
+        self.assertEqual(fact_dict, results)
+
+    def test_permission_view(self):
+        url = reverse('api:host_fact_compare_view', args=(self.host.pk,))
+        with self.current_user('admin'):
+            self.get(url, expect=200)
+        with self.current_user('normal'):
+            self.get(url, expect=200)
+        with self.current_user('other'):
+            self.get(url, expect=403)
+        with self.current_user('nobody'):
+            self.get(url, expect=403)
+        with self.current_user(None):
+            self.get(url, expect=401)
+
+    def get_fact(self, fact_obj, params=None):
+        url = build_url('api:host_fact_compare_view', args=(self.host.pk,), get=params)
+        with self.current_user(self.super_django_user):
+            response = self.get(url, expect=200)
+
+        self.check_equal(fact_obj, response)
+
+    def test_view(self):
+        self.setup_facts(2)
+        self.get_fact(Fact.objects.filter(host=self.fact_host, module='ansible').order_by('-timestamp')[0])
+
+    def test_view_module_filter(self):
+        self.setup_facts(2)
+        self.get_fact(Fact.objects.filter(host=self.fact_host, module='services').order_by('-timestamp')[0], dict(module='services'))
+
+    def test_view_time_filter(self):
+        self.setup_facts(6)
+        ts = self.builder.get_timestamp(3)
+        self.get_fact(Fact.objects.filter(host=self.fact_host, module='ansible', timestamp__lte=ts).order_by('-timestamp')[0], 
+                      dict(datetime=ts))
+
+class SingleFactApiTest(FactApiBaseTest):
+    def setUp(self):
+        super(SingleFactApiTest, self).setUp()
+
+        self.group = self.inventory.groups.create(name='test-group')
+        self.group.hosts.add(self.host, self.host2, self.host3)
+
+    def test_permission_list(self):
+        url = reverse('api:host_fact_versions_list', args=(self.host.pk,))
+        with self.current_user('admin'):
+            self.get(url, expect=200)
+        with self.current_user('normal'):
+            self.get(url, expect=200)
+        with self.current_user('other'):
+            self.get(url, expect=403)
+        with self.current_user('nobody'):
+            self.get(url, expect=403)
+        with self.current_user(None):
+            self.get(url, expect=401)
+
+    def _test_related(self, url):
+        with self.current_user(self.super_django_user):
+            response = self.get(url, expect=200)
+            self.assertTrue(len(response['results']) > 0)
+            for entry in response['results']:
+                self.assertIn('single_fact', entry['related'])
+                # Requires fields
+                r = self.get(entry['related']['single_fact'], expect=400)
+
+    def test_related_host_list(self):
+        self.setup_facts(2)
+        self._test_related(reverse('api:host_list'))
+
+    def test_related_group_list(self):
+        self.setup_facts(2)
+        self._test_related(reverse('api:group_list'))
+
+    def test_related_inventory_list(self):
+        self.setup_facts(2)
+        self._test_related(reverse('api:inventory_list'))
+
+    def test_params(self):
+        self.setup_facts(2)
+        params = {
+            'module': 'packages',
+            'fact_key': 'name',
+            'fact_value': 'acpid',
+        }
+        url = build_url('api:inventory_single_fact_view', args=(self.inventory.pk,), get=params)
+        with self.current_user(self.super_django_user):
+            response = self.get(url, expect=200)
+            self.assertEqual(len(response['results']), 3)
+            for entry in response['results']:
+                self.assertEqual(entry['fact'][0]['name'], 'acpid')

awx/main/utils.py

@@ -19,7 +19,7 @@ import tempfile
 # Django REST Framework
 from rest_framework.exceptions import ParseError, PermissionDenied
 from django.utils.encoding import smart_str
-
+from django.core.urlresolvers import reverse
 
 # PyCrypto
 from Crypto.Cipher import AES
@@ -487,3 +487,16 @@ def get_pk_from_dict(_dict, key):
         return int(_dict[key])
     except (TypeError, KeyError, ValueError):
         return None
+
+def build_url(*args, **kwargs):
+    get = kwargs.pop('get', {})
+    url = reverse(*args, **kwargs)
+    if get:
+        url += '?' + urllib.urlencode(get)
+    return url
+
+def timestamp_apiformat(timestamp):
+    timestamp = timestamp.isoformat()
+    if timestamp.endswith('+00:00'):
+        timestamp = timestamp[:-6] + 'Z'
+    return timestamp