[#28] Fix last page in pagination and add tests

ckanext/dcat/logic.py

@@ -1,4 +1,5 @@
 from __future__ import division
+import math
 
 from pylons import config
 from dateutil.parser import parse as dateutil_parse
@@ -177,7 +178,7 @@ def _page_url(page):
     pagination_info['current'] = _page_url(page)
     pagination_info['first'] = _page_url(1)
 
-    last_page = int(round(query['count'] / items_per_page)) or 1
+    last_page = int(math.ceil(query['count'] / items_per_page)) or 1
     pagination_info['last'] = _page_url(last_page)
 
     if page > 1:

ckanext/dcat/tests/test_controllers.py

@@ -1,13 +1,17 @@
 import nose
 
-from routes import url_for
+from ckan.lib.helpers import url_for
+
+from rdflib import Graph
 
 try:
     from ckan.tests import helpers, factories
 except ImportError:
     from ckan.new_tests import helpers, factories
 
 from ckanext.dcat.processors import RDFParser
+from ckanext.dcat.profiles import RDF, DCAT
+from ckanext.dcat.processors import HYDRA
 
 eq_ = nose.tools.eq_
 assert_true = nose.tools.assert_true
@@ -186,3 +190,42 @@ def test_catalog_ttl(self):
         dcat_datasets = [d for d in p.datasets()]
 
         eq_(len(dcat_datasets), 4)
+
+    def _object_value(self, graph, subject, predicate):
+
+        objects = [o for o in graph.objects(subject, predicate)]
+        return unicode(objects[0]) if objects else None
+
+    @helpers.change_config('ckanext.dcat.datasets_per_page', 10)
+    def test_catalog_pagination(self):
+
+        for i in xrange(12):
+            factories.Dataset()
+
+        app = self._get_test_app()
+
+        url = url_for('dcat_catalog', _format='rdf')
+
+        response = app.get(url)
+
+        content = response.body
+
+        g = Graph()
+        g.parse(data=content, format='xml')
+
+        eq_(len([d for d in g.subjects(RDF.type, DCAT.Dataset)]), 10)
+
+        pagination = [o for o in g.subjects(RDF.type, HYDRA.PagedCollection)][0]
+
+        eq_(self._object_value(g, pagination, HYDRA.totalItems), '12')
+
+        eq_(self._object_value(g, pagination, HYDRA.itemsPerPage), '10')
+
+        eq_(self._object_value(g, pagination, HYDRA.firstPage),
+            url_for('dcat_catalog', _format='rdf', page=1, host='localhost'))
+
+        eq_(self._object_value(g, pagination, HYDRA.nextPage),
+            url_for('dcat_catalog', _format='rdf', page=2, host='localhost'))
+
+        eq_(self._object_value(g, pagination, HYDRA.lastPage),
+            url_for('dcat_catalog', _format='rdf', page=2, host='localhost'))

ckanext/dcat/tests/test_logic.py

@@ -0,0 +1,224 @@
+import nose
+import mock
+
+from pylons import config
+
+from ckan.plugins import toolkit
+
+try:
+    from ckan.tests import helpers
+except ImportError:
+    from ckan.new_tests import helpers
+
+from ckanext.dcat.logic import _pagination_info
+
+eq_ = nose.tools.eq_
+assert_raises = nose.tools.assert_raises
+
+
+class TestPagination(object):
+
+    @helpers.change_config('ckanext.dcat.datasets_per_page', 10)
+    @mock.patch('ckan.plugins.toolkit.request')
+    def test_pagination(self, mock_request):
+
+        mock_request.params = {}
+        mock_request.path_url = 'http://example.com'
+
+        # No page defined (defaults to 1)
+        query = {
+            'count': 12,
+            'results': [x for x in xrange(10)],
+        }
+        data_dict = {
+            'page': None
+        }
+
+        pagination = _pagination_info(query, data_dict)
+
+        eq_(pagination['count'], 12)
+        eq_(pagination['items_per_page'],
+            config.get('ckanext.dcat.datasets_per_page'))
+        eq_(pagination['current'], 'http://example.com?page=1')
+        eq_(pagination['first'], 'http://example.com?page=1')
+        eq_(pagination['last'], 'http://example.com?page=2')
+        eq_(pagination['next'], 'http://example.com?page=2')
+        assert 'previous' not in pagination
+
+        # Page 1
+        query = {
+            'count': 12,
+            'results': [x for x in xrange(10)],
+        }
+        data_dict = {
+            'page': 1
+        }
+
+        pagination = _pagination_info(query, data_dict)
+
+        eq_(pagination['count'], 12)
+        eq_(pagination['items_per_page'],
+            config.get('ckanext.dcat.datasets_per_page'))
+        eq_(pagination['current'], 'http://example.com?page=1')
+        eq_(pagination['first'], 'http://example.com?page=1')
+        eq_(pagination['last'], 'http://example.com?page=2')
+        eq_(pagination['next'], 'http://example.com?page=2')
+        assert 'previous' not in pagination
+
+        # Page 2
+        query = {
+            'count': 12,
+            'results': [x for x in xrange(2)],
+        }
+        data_dict = {
+            'page': 2
+        }
+
+        pagination = _pagination_info(query, data_dict)
+
+        eq_(pagination['count'], 12)
+        eq_(pagination['items_per_page'],
+            config.get('ckanext.dcat.datasets_per_page'))
+        eq_(pagination['current'], 'http://example.com?page=2')
+        eq_(pagination['first'], 'http://example.com?page=1')
+        eq_(pagination['last'], 'http://example.com?page=2')
+        eq_(pagination['previous'], 'http://example.com?page=1')
+        assert 'next' not in pagination
+
+        # Page 3
+        query = {
+            'count': 12,
+            'results': [x for x in xrange(2)],
+        }
+        data_dict = {
+            'page': 3
+        }
+
+        pagination = _pagination_info(query, data_dict)
+
+        eq_(pagination['count'], 12)
+        eq_(pagination['items_per_page'],
+            config.get('ckanext.dcat.datasets_per_page'))
+        eq_(pagination['current'], 'http://example.com?page=3')
+        eq_(pagination['first'], 'http://example.com?page=1')
+        eq_(pagination['last'], 'http://example.com?page=2')
+        eq_(pagination['previous'], 'http://example.com?page=2')
+        assert 'next' not in pagination
+
+    @helpers.change_config('ckanext.dcat.datasets_per_page', 100)
+    @mock.patch('ckan.plugins.toolkit.request')
+    def test_pagination_less_results_than_page_size(self, mock_request):
+
+        mock_request.params = {}
+        mock_request.path_url = 'http://example.com'
+
+        # No page defined (defaults to 1)
+        query = {
+            'count': 12,
+            'results': [x for x in xrange(12)],
+        }
+        data_dict = {
+            'page': None
+        }
+
+        pagination = _pagination_info(query, data_dict)
+
+        eq_(pagination['count'], 12)
+        eq_(pagination['items_per_page'],
+            config.get('ckanext.dcat.datasets_per_page'))
+        eq_(pagination['current'], 'http://example.com?page=1')
+        eq_(pagination['first'], 'http://example.com?page=1')
+        eq_(pagination['last'], 'http://example.com?page=1')
+        assert 'next' not in pagination
+        assert 'previous' not in pagination
+
+    @helpers.change_config('ckanext.dcat.datasets_per_page', 10)
+    @mock.patch('ckan.plugins.toolkit.request')
+    def test_pagination_same_results_than_page_size(self, mock_request):
+
+        mock_request.params = {}
+        mock_request.path_url = 'http://example.com'
+
+        # No page defined (defaults to 1)
+        query = {
+            'count': 10,
+            'results': [x for x in xrange(10)],
+        }
+        data_dict = {
+            'page': None
+        }
+
+        pagination = _pagination_info(query, data_dict)
+
+        eq_(pagination['count'], 10)
+        eq_(pagination['items_per_page'],
+            config.get('ckanext.dcat.datasets_per_page'))
+        eq_(pagination['current'], 'http://example.com?page=1')
+        eq_(pagination['first'], 'http://example.com?page=1')
+        eq_(pagination['last'], 'http://example.com?page=1')
+        assert 'next' not in pagination
+        assert 'previous' not in pagination
+
+    @helpers.change_config('ckanext.dcat.datasets_per_page', 10)
+    @mock.patch('ckan.plugins.toolkit.request')
+    def test_pagination_keeps_params(self, mock_request):
+
+        mock_request.params = {'a': 1, 'b': 2}
+        mock_request.path_url = 'http://example.com'
+
+        # No page defined (defaults to 1)
+        query = {
+            'count': 12,
+            'results': [x for x in xrange(10)],
+        }
+        data_dict = {
+            'page': None
+        }
+
+        pagination = _pagination_info(query, data_dict)
+
+        eq_(pagination['count'], 12)
+        eq_(pagination['items_per_page'],
+            config.get('ckanext.dcat.datasets_per_page'))
+        eq_(pagination['current'], 'http://example.com?a=1&b=2&page=1')
+        eq_(pagination['first'], 'http://example.com?a=1&b=2&page=1')
+        eq_(pagination['last'], 'http://example.com?a=1&b=2&page=2')
+        eq_(pagination['next'], 'http://example.com?a=1&b=2&page=2')
+        assert 'previous' not in pagination
+
+    def test_pagination_no_results_empty_dict(self):
+        query = {
+            'count': 0,
+            'results': [],
+        }
+        data_dict = {
+            'page': None
+        }
+
+        pagination = _pagination_info(query, data_dict)
+
+        eq_(pagination, {})
+
+    def test_pagination_wrong_page(self):
+        query = {
+            'count': 10,
+            'results': [x for x in xrange(10)],
+        }
+        data_dict = {
+            'page': 'a'
+        }
+
+        assert_raises(toolkit.ValidationError,
+                      _pagination_info, query, data_dict)
+
+    def test_pagination_wrong_page_number(self):
+        query = {
+            'count': 10,
+            'results': [x for x in xrange(10)],
+        }
+        data_dict = {
+            'page': '-1'
+        }
+
+        assert_raises(toolkit.ValidationError,
+                      _pagination_info, query, data_dict)