Add tests for async and result row utils

Signed-off-by: Don Naegely <naegelyd@gmail.com>

.travis.yml

@@ -13,6 +13,7 @@ env:
 
 services:
     - memcached
+    - redis-server
 
 addons:
     - postgres
@@ -40,4 +41,4 @@ after_success:
 matrix:
     exclude:
         - python: "2.6"
-          env: DJANGO=1.7.6 POSTGRES_TEST_USER=postgres POSTGRES_TEST_NAME=avocado MYSQL_TEST_USER=root MYSQL_TEST_NAME=avocado
+          env: DJANGO=1.7.6 POSTGRES_TEST_USER=postgres POSTGRES_TEST_NAME=avocado MYSQL_TEST_USER=root MYSQL_TEST_NAME=avocado

avocado/query/utils.py

@@ -12,7 +12,7 @@
 
 logger = logging.getLogger(__name__)
 
-
+DEFAULT_LIMIT = 20
 TEMP_DB_ALIAS_PREFIX = '_db:{0}'
 
 
@@ -284,15 +284,15 @@ def get_result_rows(context, view, query_options, evaluate_rows=False):
     offset = None
 
     page = query_options.get('page')
-    limit = query_options.get('limit')
+    limit = query_options.get('limit') or 0
     stop_page = query_options.get('stop_page')
     query_name = query_options.get('query_name')
-    processor_name = query_options.get('processor')
+    processor_name = query_options.get('processor') or 'default'
     tree = query_options.get('tree')
-    export_type = query_options.get('export_type')
+    export_type = query_options.get('export_type') or 'html'
     reader = query_options.get('reader')
 
-    if page:
+    if page is not None:
         page = int(page)
 
         # Pages are 1-based.

tests/cases/query/tests/utils.py

@@ -1,11 +1,17 @@
 import time
 from threading import Thread
+
+from django.conf import settings
+from django.core import management
 from django.db import connections, DatabaseError
 from django.test import TransactionTestCase
-from django.core import management
-from django.conf import settings
-from tests.models import Employee
+from rq.job import JobStatus
+
+from avocado.async import utils as async_utils
+from avocado.models import DataContext, DataField, DataView
 from avocado.query import utils
+from tests.models import Employee
+from tests.processors import ManagerQueryProcessor
 
 
 class TempConnTest(TransactionTestCase):
@@ -135,3 +141,190 @@ def stopper(t, db, name):
                 t.assertTrue(canceled)
 
             self.run_cancel_test(runner, stopper)
+
+
+class AsyncResultRowTestCase(TransactionTestCase):
+    fixtures = ['tests/fixtures/employee_data.json']
+
+    def setUp(self):
+        management.call_command('avocado', 'init', 'tests', quiet=True)
+        # Don't start with any jobs in the queue.
+        async_utils.cancel_all_jobs()
+
+    def tearDown(self):
+        # Don't leave any jobs in the queue.
+        async_utils.cancel_all_jobs()
+
+    def test_create_and_cancel(self):
+        # Create 3 meaningless jobs. We're just testing job setup and
+        # cancellation here, not the execution.
+        utils.async_get_result_rows(None, None, {})
+        job_options = {
+            'name': 'Job X',
+        }
+        job_x_id = utils.async_get_result_rows(None, None, {}, job_options)
+        job_options = {
+            'name': 'Job Y',
+            'query_name': 'job_y_query',
+        }
+        job_y_id = utils.async_get_result_rows(None, None, {}, job_options)
+
+        self.assertEqual(async_utils.get_job_count(), 3)
+
+        jobs = async_utils.get_jobs()
+        self.assertEqual(len(jobs), 3)
+
+        job_x = async_utils.get_job(job_x_id)
+        self.assertTrue(job_x in jobs)
+        self.assertEqual(job_x.meta['name'], 'Job X')
+
+        self.assertEqual(async_utils.cancel_job(job_x_id), None)
+        self.assertEqual(async_utils.get_job_count(), 2)
+        async_utils.cancel_job('invalid_id')
+        self.assertEqual(async_utils.get_job_count(), 2)
+
+        self.assertTrue('canceled' in async_utils.cancel_job(job_y_id))
+        self.assertTrue(async_utils.get_job_count(), 1)
+
+        async_utils.cancel_all_jobs()
+        self.assertEqual(async_utils.get_job_count(), 0)
+
+    def test_job_result(self):
+        context = DataContext()
+        view = DataView()
+        limit = 3
+        query_options = {
+            'limit': limit,
+            'page': 1,
+        }
+
+        job_id = utils.async_get_result_rows(context, view, query_options)
+        self.assertTrue(async_utils.get_job_count(), 1)
+        async_utils.run_jobs()
+        time.sleep(1)
+        result = async_utils.get_job_result(job_id)
+        self.assertEqual(async_utils.get_job(job_id).status,
+                         JobStatus.FINISHED)
+        self.assertEqual(len(result['rows']), limit)
+        self.assertEqual(result['limit'], limit)
+
+    def test_invalid_job_result(self):
+        context = DataContext()
+        view = DataView()
+        query_options = {
+            'page': 0,
+        }
+
+        job_id = utils.async_get_result_rows(context, view, query_options)
+        self.assertTrue(async_utils.get_job_count(), 1)
+        async_utils.run_jobs()
+        time.sleep(1)
+        self.assertEqual(async_utils.get_job_result(job_id), None)
+        self.assertEqual(async_utils.get_job(job_id).status, JobStatus.FAILED)
+
+
+class ResultRowTestCase(TransactionTestCase):
+    fixtures = ['tests/fixtures/employee_data.json']
+
+    def setUp(self):
+        management.call_command('avocado', 'init', 'tests', quiet=True)
+
+    def test_invalid_options(self):
+        # Page numbers less than 1 should not be allowed.
+        query_options = {
+            'page': 0,
+        }
+        self.assertRaises(ValueError,
+                          utils.get_result_rows,
+                          None,
+                          None,
+                          query_options)
+
+        # Stop pages before start pages should not be allowed.
+        query_options = {
+            'page': 5,
+            'stop_page': 1,
+        }
+        self.assertRaises(ValueError,
+                          utils.get_result_rows,
+                          None,
+                          None,
+                          query_options)
+
+    def test_get_rows(self):
+        context = DataContext()
+        view = DataView()
+
+        # Unless we tell the function to evaluate the rows, it should return
+        # rows as a generator so we need to exclicitly evaluate it here.
+        result = utils.get_result_rows(context, view, {})
+        self.assertEqual(len(list(result['rows'])), Employee.objects.count())
+
+        # Now, have the method evaluate the rows.
+        result = utils.get_result_rows(context, view, {}, evaluate_rows=True)
+        self.assertEqual(len(result['rows']), Employee.objects.count())
+
+    def test_get_order_only(self):
+        field = DataField.objects.get(field_name='salary')
+        concept = field.concepts.all()[0]
+
+        context = DataContext()
+        view = DataView(json=[{
+            'concept': concept.pk,
+            'visible': False,
+            'sort': 'desc',
+        }])
+        result = utils.get_result_rows(context, view, {})
+        self.assertEqual(len(list(result['rows'])), Employee.objects.count())
+
+    def test_limit(self):
+        context = DataContext()
+        view = DataView()
+        limit = 2
+        query_options = {
+            'limit': limit,
+            'page': 1,
+        }
+        result = utils.get_result_rows(context, view, query_options)
+        self.assertEqual(len(list(result['rows'])), limit)
+        self.assertEqual(result['limit'], limit)
+
+    def test_processor(self):
+        context = DataContext()
+        view = DataView()
+        processor = 'manager'
+        query_options = {
+            'processor': processor,
+        }
+        result = utils.get_result_rows(context, view, query_options)
+        self.assertEqual(len(list(result['rows'])),
+                         Employee.objects.filter(is_manager=True).count())
+        self.assertTrue(isinstance(result['processor'], ManagerQueryProcessor))
+
+    def test_export_type(self):
+        context = DataContext()
+        view = DataView()
+        export_type = 'json'
+        query_options = {
+            'export_type': export_type,
+        }
+        result = utils.get_result_rows(context, view, query_options)
+        self.assertEqual(len(list(result['rows'])), Employee.objects.count())
+        self.assertEqual(result['export_type'], export_type)
+
+    def test_pages(self):
+        context = DataContext()
+        view = DataView()
+        query_options = {
+            'page': 1,
+            'stop_page': 10,
+        }
+        result = utils.get_result_rows(context, view, query_options)
+        self.assertEqual(len(list(result['rows'])), Employee.objects.count())
+
+        query_options = {
+            'page': 1,
+            'stop_page': 1,
+        }
+        result = utils.get_result_rows(context, view, query_options)
+        self.assertEqual(len(list(result['rows'])), Employee.objects.count())

tests/processors.py

@@ -0,0 +1,13 @@
+from avocado.models import DataContext
+from avocado.query.pipeline import QueryProcessor
+
+
+class ManagerQueryProcessor(QueryProcessor):
+    def __init__(self, *args, **kwargs):
+        kwargs['context'] = DataContext(json={
+            'field': 'tests.employee.is_manager',
+            'operator': 'exact',
+            'value': True,
+        })
+
+        super(ManagerQueryProcessor, self).__init__(*args, **kwargs)

tests/settings.py

@@ -143,17 +143,26 @@
             'handlers': ['null'],
             'level': 'DEBUG',
             'propagate': True,
-        }
+        },
+        'rq.worker': {
+            'handlers': ['null'],
+            'level': 'DEBUG',
+        },
     }
 }
 
 SOUTH_TESTS_MIGRATE = False
 
+AVOCADO_QUEUE_NAME = 'avocado_test_queue'
 AVOCADO = {
     'HISTORY_ENABLED': False,
     'HISTORY_MAX_SIZE': 50,
     'METADATA_MIGRATION_APP': 'core',
     'DATA_CACHE_ENABLED': False,
+    'QUERY_PROCESSORS': {
+        'manager': 'tests.processors.ManagerQueryProcessor',
+    },
+    'ASYNC_QUEUE': AVOCADO_QUEUE_NAME,
 }
 
 MODELTREES = {
@@ -171,3 +180,11 @@
 MIDDLEWARE_CLASSES = ()
 
 SECRET_KEY = 'acb123'
+
+RQ_QUEUES = {
+    AVOCADO_QUEUE_NAME: {
+        'HOST': 'localhost',
+        'PORT': 6379,
+        'DB': 0,
+    },
+}