test_pagination.py

import pytest
from django.core.paginator import Paginator as DjangoPaginator
from django.db import models
from django.test import TestCase, override_settings

from rest_framework import (
    exceptions, filters, generics, pagination, serializers, status
)
from rest_framework.pagination import PAGE_BREAK, PageLink
from rest_framework.request import Request
from rest_framework.test import APIRequestFactory

factory = APIRequestFactory()


class TestPaginationIntegration:
    """
    Integration tests.
    """

    def setup_method(self):
        class PassThroughSerializer(serializers.BaseSerializer):
            def to_representation(self, item):
                return item

        class EvenItemsOnly(filters.BaseFilterBackend):
            def filter_queryset(self, request, queryset, view):
                return [item for item in queryset if item % 2 == 0]

        class BasicPagination(pagination.PageNumberPagination):
            page_size = 5
            page_size_query_param = 'page_size'
            max_page_size = 20

        self.view = generics.ListAPIView.as_view(
            serializer_class=PassThroughSerializer,
            queryset=range(1, 101),
            filter_backends=[EvenItemsOnly],
            pagination_class=BasicPagination
        )

    def test_filtered_items_are_paginated(self):
        request = factory.get('/', {'page': 2})
        response = self.view(request)
        assert response.status_code == status.HTTP_200_OK
        assert response.data == {
            'results': [12, 14, 16, 18, 20],
            'previous': 'http://testserver/',
            'next': 'http://testserver/?page=3',
            'count': 50
        }

    def test_setting_page_size(self):
        """
        When 'paginate_by_param' is set, the client may choose a page size.
        """
        request = factory.get('/', {'page_size': 10})
        response = self.view(request)
        assert response.status_code == status.HTTP_200_OK
        assert response.data == {
            'results': [2, 4, 6, 8, 10, 12, 14, 16, 18, 20],
            'previous': None,
            'next': 'http://testserver/?page=2&page_size=10',
            'count': 50
        }

    def test_setting_page_size_over_maximum(self):
        """
        When page_size parameter exceeds maximum allowable,
        then it should be capped to the maximum.
        """
        request = factory.get('/', {'page_size': 1000})
        response = self.view(request)
        assert response.status_code == status.HTTP_200_OK
        assert response.data == {
            'results': [
                2, 4, 6, 8, 10, 12, 14, 16, 18, 20,
                22, 24, 26, 28, 30, 32, 34, 36, 38, 40
            ],
            'previous': None,
            'next': 'http://testserver/?page=2&page_size=1000',
            'count': 50
        }

    def test_setting_page_size_to_zero(self):
        """
        When page_size parameter is invalid it should return to the default.
        """
        request = factory.get('/', {'page_size': 0})
        response = self.view(request)
        assert response.status_code == status.HTTP_200_OK
        assert response.data == {
            'results': [2, 4, 6, 8, 10],
            'previous': None,
            'next': 'http://testserver/?page=2&page_size=0',
            'count': 50
        }

    def test_additional_query_params_are_preserved(self):
        request = factory.get('/', {'page': 2, 'filter': 'even'})
        response = self.view(request)
        assert response.status_code == status.HTTP_200_OK
        assert response.data == {
            'results': [12, 14, 16, 18, 20],
            'previous': 'http://testserver/?filter=even',
            'next': 'http://testserver/?filter=even&page=3',
            'count': 50
        }

    def test_empty_query_params_are_preserved(self):
        request = factory.get('/', {'page': 2, 'filter': ''})
        response = self.view(request)
        assert response.status_code == status.HTTP_200_OK
        assert response.data == {
            'results': [12, 14, 16, 18, 20],
            'previous': 'http://testserver/?filter=',
            'next': 'http://testserver/?filter=&page=3',
            'count': 50
        }

    def test_404_not_found_for_zero_page(self):
        request = factory.get('/', {'page': '0'})
        response = self.view(request)
        assert response.status_code == status.HTTP_404_NOT_FOUND
        assert response.data == {
            'detail': 'Invalid page.'
        }

    def test_404_not_found_for_invalid_page(self):
        request = factory.get('/', {'page': 'invalid'})
        response = self.view(request)
        assert response.status_code == status.HTTP_404_NOT_FOUND
        assert response.data == {
            'detail': 'Invalid page.'
        }


class TestPaginationSettingsIntegration:
    """
    Integration tests for pagination settings.
    """

    def setup_method(self):
        class PassThroughSerializer(serializers.BaseSerializer):
            def to_representation(self, item):
                return item

        class EvenItemsOnly(filters.BaseFilterBackend):
            def filter_queryset(self, request, queryset, view):
                return [item for item in queryset if item % 2 == 0]

        class BasicPagination(pagination.PageNumberPagination):
            page_size_query_param = 'page_size'

        self.view = generics.ListAPIView.as_view(
            serializer_class=PassThroughSerializer,
            queryset=range(1, 101),
            filter_backends=[EvenItemsOnly],
            pagination_class=BasicPagination
        )

    @override_settings(
        REST_FRAMEWORK={
            "MAX_PAGE_SIZE": 20,
            "PAGE_SIZE": 5,
        }
    )
    def test_setting_page_size_over_maximum(self):
        """
        When page_size parameter exceeds maximum allowable,
        then it should be capped to the maximum.
        """
        request = factory.get('/', {'page_size': 1000})
        response = self.view(request)
        assert response.status_code == status.HTTP_200_OK
        assert len(response.data["results"]) == 20, response.data
        assert response.data == {
            'results': [
                2, 4, 6, 8, 10, 12, 14, 16, 18, 20,
                22, 24, 26, 28, 30, 32, 34, 36, 38, 40
            ],
            'previous': None,
            'next': 'http://testserver/?page=2&page_size=1000',
            'count': 50
        }

    @override_settings(
        REST_FRAMEWORK={
            "MAX_PAGE_SIZE": 20,
            "PAGE_SIZE": 5,
        }
    )
    def test_setting_page_size_to_zero(self):
        """
        When page_size parameter is invalid it should return to the default.
        """
        request = factory.get('/', {'page_size': 0})
        response = self.view(request)
        assert response.status_code == status.HTTP_200_OK
        assert len(response.data["results"]) == 5, response.data
        assert response.data == {
            'results': [2, 4, 6, 8, 10],
            'previous': None,
            'next': 'http://testserver/?page=2&page_size=0',
            'count': 50
        }


class TestPaginationDisabledIntegration:
    """
    Integration tests for disabled pagination.
    """

    def setup_method(self):
        class PassThroughSerializer(serializers.BaseSerializer):
            def to_representation(self, item):
                return item

        self.view = generics.ListAPIView.as_view(
            serializer_class=PassThroughSerializer,
            queryset=range(1, 101),
            pagination_class=None
        )

    def test_unpaginated_list(self):
        request = factory.get('/', {'page': 2})
        response = self.view(request)
        assert response.status_code == status.HTTP_200_OK
        assert response.data == list(range(1, 101))


class TestPageNumberPagination:
    """
    Unit tests for `pagination.PageNumberPagination`.
    """

    def setup_method(self):
        class ExamplePagination(pagination.PageNumberPagination):
            page_size = 5

        self.pagination = ExamplePagination()
        self.queryset = range(1, 101)

    def paginate_queryset(self, request):
        return list(self.pagination.paginate_queryset(self.queryset, request))

    def get_paginated_content(self, queryset):
        response = self.pagination.get_paginated_response(queryset)
        return response.data

    def get_html_context(self):
        return self.pagination.get_html_context()

    @pytest.mark.parametrize('url', ['/', '/?page='])
    def test_no_page_number(self, url):
        request = Request(factory.get(url))
        queryset = self.paginate_queryset(request)
        content = self.get_paginated_content(queryset)
        context = self.get_html_context()
        assert queryset == [1, 2, 3, 4, 5]
        assert content == {
            'results': [1, 2, 3, 4, 5],
            'previous': None,
            'next': 'http://testserver/?page=2',
            'count': 100
        }
        assert context == {
            'previous_url': None,
            'next_url': 'http://testserver/?page=2',
            'page_links': [
                PageLink('http://testserver/', 1, True, False),
                PageLink('http://testserver/?page=2', 2, False, False),
                PageLink('http://testserver/?page=3', 3, False, False),
                PAGE_BREAK,
                PageLink('http://testserver/?page=20', 20, False, False),
            ]
        }
        assert self.pagination.display_page_controls
        assert isinstance(self.pagination.to_html(), str)

    def test_second_page(self):
        request = Request(factory.get('/', {'page': 2}))
        queryset = self.paginate_queryset(request)
        content = self.get_paginated_content(queryset)
        context = self.get_html_context()
        assert queryset == [6, 7, 8, 9, 10]
        assert content == {
            'results': [6, 7, 8, 9, 10],
            'previous': 'http://testserver/',
            'next': 'http://testserver/?page=3',
            'count': 100
        }
        assert context == {
            'previous_url': 'http://testserver/',
            'next_url': 'http://testserver/?page=3',
            'page_links': [
                PageLink('http://testserver/', 1, False, False),
                PageLink('http://testserver/?page=2', 2, True, False),
                PageLink('http://testserver/?page=3', 3, False, False),
                PAGE_BREAK,
                PageLink('http://testserver/?page=20', 20, False, False),
            ]
        }

    def test_last_page(self):
        request = Request(factory.get('/', {'page': 'last'}))
        queryset = self.paginate_queryset(request)
        content = self.get_paginated_content(queryset)
        context = self.get_html_context()
        assert queryset == [96, 97, 98, 99, 100]
        assert content == {
            'results': [96, 97, 98, 99, 100],
            'previous': 'http://testserver/?page=19',
            'next': None,
            'count': 100
        }
        assert context == {
            'previous_url': 'http://testserver/?page=19',
            'next_url': None,
            'page_links': [
                PageLink('http://testserver/', 1, False, False),
                PAGE_BREAK,
                PageLink('http://testserver/?page=18', 18, False, False),
                PageLink('http://testserver/?page=19', 19, False, False),
                PageLink('http://testserver/?page=20', 20, True, False),
            ]
        }

    def test_invalid_page(self):
        request = Request(factory.get('/', {'page': 'invalid'}))
        with pytest.raises(exceptions.NotFound):
            self.paginate_queryset(request)

    def test_get_paginated_response_schema(self):
        unpaginated_schema = {
            'type': 'object',
            'item': {
                'properties': {
                    'test-property': {
                        'type': 'integer',
                    },
                },
            },
        }

        assert self.pagination.get_paginated_response_schema(unpaginated_schema) == {
            'type': 'object',
            'required': ['count', 'results'],
            'properties': {
                'count': {
                    'type': 'integer',
                    'example': 123,
                },
                'next': {
                    'type': 'string',
                    'nullable': True,
                    'format': 'uri',
                    'example': 'http://api.example.org/accounts/?page=4',
                },
                'previous': {
                    'type': 'string',
                    'nullable': True,
                    'format': 'uri',
                    'example': 'http://api.example.org/accounts/?page=2',
                },
                'results': unpaginated_schema,
            },
        }


class TestPageNumberPaginationOverride:
    """
    Unit tests for `pagination.PageNumberPagination`.

    the Django Paginator Class is overridden.
    """

    def setup_method(self):
        class OverriddenDjangoPaginator(DjangoPaginator):
            # override the count in our overridden Django Paginator
            # we will only return one page, with one item
            count = 1

        class ExamplePagination(pagination.PageNumberPagination):
            django_paginator_class = OverriddenDjangoPaginator
            page_size = 5

        self.pagination = ExamplePagination()
        self.queryset = range(1, 101)

    def paginate_queryset(self, request):
        return list(self.pagination.paginate_queryset(self.queryset, request))

    def get_paginated_content(self, queryset):
        response = self.pagination.get_paginated_response(queryset)
        return response.data

    def get_html_context(self):
        return self.pagination.get_html_context()

    def test_no_page_number(self):
        request = Request(factory.get('/'))
        queryset = self.paginate_queryset(request)
        content = self.get_paginated_content(queryset)
        context = self.get_html_context()
        assert queryset == [1]
        assert content == {
            'results': [1, ],
            'previous': None,
            'next': None,
            'count': 1
        }
        assert context == {
            'previous_url': None,
            'next_url': None,
            'page_links': [
                PageLink('http://testserver/', 1, True, False),
            ]
        }
        assert not self.pagination.display_page_controls
        assert isinstance(self.pagination.to_html(), str)

    def test_invalid_page(self):
        request = Request(factory.get('/', {'page': 'invalid'}))
        with pytest.raises(exceptions.NotFound):
            self.paginate_queryset(request)


class TestLimitOffset:
    """
    Unit tests for `pagination.LimitOffsetPagination`.
    """

    def setup_method(self):
        class ExamplePagination(pagination.LimitOffsetPagination):
            default_limit = 10
            max_limit = 15

        self.pagination = ExamplePagination()
        self.queryset = range(1, 101)

    def paginate_queryset(self, request):
        return list(self.pagination.paginate_queryset(self.queryset, request))

    def get_paginated_content(self, queryset):
        response = self.pagination.get_paginated_response(queryset)
        return response.data

    def get_html_context(self):
        return self.pagination.get_html_context()

    def test_no_offset(self):
        request = Request(factory.get('/', {'limit': 5}))
        queryset = self.paginate_queryset(request)
        content = self.get_paginated_content(queryset)
        context = self.get_html_context()
        assert queryset == [1, 2, 3, 4, 5]
        assert content == {
            'results': [1, 2, 3, 4, 5],
            'previous': None,
            'next': 'http://testserver/?limit=5&offset=5',
            'count': 100
        }
        assert context == {
            'previous_url': None,
            'next_url': 'http://testserver/?limit=5&offset=5',
            'page_links': [
                PageLink('http://testserver/?limit=5', 1, True, False),
                PageLink('http://testserver/?limit=5&offset=5', 2, False, False),
                PageLink('http://testserver/?limit=5&offset=10', 3, False, False),
                PAGE_BREAK,
                PageLink('http://testserver/?limit=5&offset=95', 20, False, False),
            ]
        }
        assert self.pagination.display_page_controls
        assert isinstance(self.pagination.to_html(), str)

    def test_pagination_not_applied_if_limit_or_default_limit_not_set(self):
        class MockPagination(pagination.LimitOffsetPagination):
            default_limit = None
        request = Request(factory.get('/'))
        queryset = MockPagination().paginate_queryset(self.queryset, request)
        assert queryset is None

    def test_single_offset(self):
        """
        When the offset is not a multiple of the limit we get some edge cases:
        * The first page should still be offset zero.
        * We may end up displaying an extra page in the pagination control.
        """
        request = Request(factory.get('/', {'limit': 5, 'offset': 1}))
        queryset = self.paginate_queryset(request)
        content = self.get_paginated_content(queryset)
        context = self.get_html_context()
        assert queryset == [2, 3, 4, 5, 6]
        assert content == {
            'results': [2, 3, 4, 5, 6],
            'previous': 'http://testserver/?limit=5',
            'next': 'http://testserver/?limit=5&offset=6',
            'count': 100
        }
        assert context == {
            'previous_url': 'http://testserver/?limit=5',
            'next_url': 'http://testserver/?limit=5&offset=6',
            'page_links': [
                PageLink('http://testserver/?limit=5', 1, False, False),
                PageLink('http://testserver/?limit=5&offset=1', 2, True, False),
                PageLink('http://testserver/?limit=5&offset=6', 3, False, False),
                PAGE_BREAK,
                PageLink('http://testserver/?limit=5&offset=96', 21, False, False),
            ]
        }

    def test_first_offset(self):
        request = Request(factory.get('/', {'limit': 5, 'offset': 5}))
        queryset = self.paginate_queryset(request)
        content = self.get_paginated_content(queryset)
        context = self.get_html_context()
        assert queryset == [6, 7, 8, 9, 10]
        assert content == {
            'results': [6, 7, 8, 9, 10],
            'previous': 'http://testserver/?limit=5',
            'next': 'http://testserver/?limit=5&offset=10',
            'count': 100
        }
        assert context == {
            'previous_url': 'http://testserver/?limit=5',
            'next_url': 'http://testserver/?limit=5&offset=10',
            'page_links': [
                PageLink('http://testserver/?limit=5', 1, False, False),
                PageLink('http://testserver/?limit=5&offset=5', 2, True, False),
                PageLink('http://testserver/?limit=5&offset=10', 3, False, False),
                PAGE_BREAK,
                PageLink('http://testserver/?limit=5&offset=95', 20, False, False),
            ]
        }

    def test_middle_offset(self):
        request = Request(factory.get('/', {'limit': 5, 'offset': 10}))
        queryset = self.paginate_queryset(request)
        content = self.get_paginated_content(queryset)
        context = self.get_html_context()
        assert queryset == [11, 12, 13, 14, 15]
        assert content == {
            'results': [11, 12, 13, 14, 15],
            'previous': 'http://testserver/?limit=5&offset=5',
            'next': 'http://testserver/?limit=5&offset=15',
            'count': 100
        }
        assert context == {
            'previous_url': 'http://testserver/?limit=5&offset=5',
            'next_url': 'http://testserver/?limit=5&offset=15',
            'page_links': [
                PageLink('http://testserver/?limit=5', 1, False, False),
                PageLink('http://testserver/?limit=5&offset=5', 2, False, False),
                PageLink('http://testserver/?limit=5&offset=10', 3, True, False),
                PageLink('http://testserver/?limit=5&offset=15', 4, False, False),
                PAGE_BREAK,
                PageLink('http://testserver/?limit=5&offset=95', 20, False, False),
            ]
        }

    def test_ending_offset(self):
        request = Request(factory.get('/', {'limit': 5, 'offset': 95}))
        queryset = self.paginate_queryset(request)
        content = self.get_paginated_content(queryset)
        context = self.get_html_context()
        assert queryset == [96, 97, 98, 99, 100]
        assert content == {
            'results': [96, 97, 98, 99, 100],
            'previous': 'http://testserver/?limit=5&offset=90',
            'next': None,
            'count': 100
        }
        assert context == {
            'previous_url': 'http://testserver/?limit=5&offset=90',
            'next_url': None,
            'page_links': [
                PageLink('http://testserver/?limit=5', 1, False, False),
                PAGE_BREAK,
                PageLink('http://testserver/?limit=5&offset=85', 18, False, False),
                PageLink('http://testserver/?limit=5&offset=90', 19, False, False),
                PageLink('http://testserver/?limit=5&offset=95', 20, True, False),
            ]
        }

    def test_erronous_offset(self):
        request = Request(factory.get('/', {'limit': 5, 'offset': 1000}))
        queryset = self.paginate_queryset(request)
        self.get_paginated_content(queryset)
        self.get_html_context()

    def test_invalid_offset(self):
        """
        An invalid offset query param should be treated as 0.
        """
        request = Request(factory.get('/', {'limit': 5, 'offset': 'invalid'}))
        queryset = self.paginate_queryset(request)
        assert queryset == [1, 2, 3, 4, 5]

    def test_invalid_limit(self):
        """
        An invalid limit query param should be ignored in favor of the default.
        """
        request = Request(factory.get('/', {'limit': 'invalid', 'offset': 0}))
        queryset = self.paginate_queryset(request)
        content = self.get_paginated_content(queryset)
        next_limit = self.pagination.default_limit
        next_offset = self.pagination.default_limit
        next_url = 'http://testserver/?limit={}&offset={}'.format(next_limit, next_offset)
        assert queryset == [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
        assert content.get('next') == next_url

    def test_zero_limit(self):
        """
        An zero limit query param should be ignored in favor of the default.
        """
        request = Request(factory.get('/', {'limit': 0, 'offset': 0}))
        queryset = self.paginate_queryset(request)
        content = self.get_paginated_content(queryset)
        next_limit = self.pagination.default_limit
        next_offset = self.pagination.default_limit
        next_url = 'http://testserver/?limit={}&offset={}'.format(next_limit, next_offset)
        assert queryset == [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
        assert content.get('next') == next_url

    def test_max_limit(self):
        """
        The limit defaults to the max_limit when there is a max_limit and the
        requested limit is greater than the max_limit
        """
        offset = 50
        request = Request(factory.get('/', {'limit': '11235', 'offset': offset}))
        queryset = self.paginate_queryset(request)
        content = self.get_paginated_content(queryset)
        max_limit = self.pagination.max_limit
        next_offset = offset + max_limit
        prev_offset = offset - max_limit
        base_url = 'http://testserver/?limit={}'.format(max_limit)
        next_url = base_url + '&offset={}'.format(next_offset)
        prev_url = base_url + '&offset={}'.format(prev_offset)
        assert queryset == list(range(51, 66))
        assert content.get('next') == next_url
        assert content.get('previous') == prev_url

    def test_get_paginated_response_schema(self):
        unpaginated_schema = {
            'type': 'object',
            'item': {
                'properties': {
                    'test-property': {
                        'type': 'integer',
                    },
                },
            },
        }

        assert self.pagination.get_paginated_response_schema(unpaginated_schema) == {
            'type': 'object',
            'required': ['count', 'results'],
            'properties': {
                'count': {
                    'type': 'integer',
                    'example': 123,
                },
                'next': {
                    'type': 'string',
                    'nullable': True,
                    'format': 'uri',
                    'example': 'http://api.example.org/accounts/?offset=400&limit=100',
                },
                'previous': {
                    'type': 'string',
                    'nullable': True,
                    'format': 'uri',
                    'example': 'http://api.example.org/accounts/?offset=200&limit=100',
                },
                'results': unpaginated_schema,
            },
        }


class CursorPaginationTestsMixin:

    def test_invalid_cursor(self):
        request = Request(factory.get('/', {'cursor': '123'}))
        with pytest.raises(exceptions.NotFound):
            self.pagination.paginate_queryset(self.queryset, request)

    def test_use_with_ordering_filter(self):
        class MockView:
            filter_backends = (filters.OrderingFilter,)
            ordering_fields = ['username', 'created']
            ordering = 'created'

        request = Request(factory.get('/', {'ordering': 'username'}))
        ordering = self.pagination.get_ordering(request, [], MockView())
        assert ordering == ('username',)

        request = Request(factory.get('/', {'ordering': '-username'}))
        ordering = self.pagination.get_ordering(request, [], MockView())
        assert ordering == ('-username',)

        request = Request(factory.get('/', {'ordering': 'invalid'}))
        ordering = self.pagination.get_ordering(request, [], MockView())
        assert ordering == ('created',)

    def test_use_with_ordering_filter_without_ordering_default_value(self):
        class MockView:
            filter_backends = (filters.OrderingFilter,)
            ordering_fields = ['username', 'created']

        request = Request(factory.get('/'))
        ordering = self.pagination.get_ordering(request, [], MockView())
        # it gets the value of `ordering` provided by CursorPagination
        assert ordering == ('created',)

        request = Request(factory.get('/', {'ordering': 'username'}))
        ordering = self.pagination.get_ordering(request, [], MockView())
        assert ordering == ('username',)

        request = Request(factory.get('/', {'ordering': 'invalid'}))
        ordering = self.pagination.get_ordering(request, [], MockView())
        assert ordering == ('created',)

    def test_cursor_pagination(self):
        (previous, current, next, previous_url, next_url) = self.get_pages('/')

        assert previous is None
        assert current == [1, 1, 1, 1, 1]
        assert next == [1, 2, 3, 4, 4]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [1, 1, 1, 1, 1]
        assert current == [1, 2, 3, 4, 4]
        assert next == [4, 4, 5, 6, 7]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [1, 2, 3, 4, 4]
        assert current == [4, 4, 5, 6, 7]
        assert next == [7, 7, 7, 7, 7]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [4, 4, 4, 5, 6]  # Paging artifact
        assert current == [7, 7, 7, 7, 7]
        assert next == [7, 7, 7, 8, 9]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [7, 7, 7, 7, 7]
        assert current == [7, 7, 7, 8, 9]
        assert next == [9, 9, 9, 9, 9]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [7, 7, 7, 8, 9]
        assert current == [9, 9, 9, 9, 9]
        assert next is None

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous == [7, 7, 7, 7, 7]
        assert current == [7, 7, 7, 8, 9]
        assert next == [9, 9, 9, 9, 9]

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous == [4, 4, 5, 6, 7]
        assert current == [7, 7, 7, 7, 7]
        assert next == [8, 9, 9, 9, 9]  # Paging artifact

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous == [1, 2, 3, 4, 4]
        assert current == [4, 4, 5, 6, 7]
        assert next == [7, 7, 7, 7, 7]

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous == [1, 1, 1, 1, 1]
        assert current == [1, 2, 3, 4, 4]
        assert next == [4, 4, 5, 6, 7]

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous is None
        assert current == [1, 1, 1, 1, 1]
        assert next == [1, 2, 3, 4, 4]

        assert isinstance(self.pagination.to_html(), str)

    def test_cursor_pagination_current_page_empty_forward(self):
        # Regression test for #6504
        self.pagination.base_url = "/"

        # We have a cursor on the element at position 100, but this element doesn't exist
        # anymore.
        cursor = pagination.Cursor(reverse=False, offset=0, position=100)
        url = self.pagination.encode_cursor(cursor)
        self.pagination.base_url = "/"

        # Loading the page with this cursor doesn't crash
        (previous, current, next, previous_url, next_url) = self.get_pages(url)

        # The previous url doesn't crash either
        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        # And point to things that are not completely off.
        assert previous == [7, 7, 7, 8, 9]
        assert current == [9, 9, 9, 9, 9]
        assert next == []
        assert previous_url is not None
        assert next_url is not None

    def test_cursor_pagination_current_page_empty_reverse(self):
        # Regression test for #6504
        self.pagination.base_url = "/"

        # We have a cursor on the element at position 100, but this element doesn't exist
        # anymore.
        cursor = pagination.Cursor(reverse=True, offset=0, position=100)
        url = self.pagination.encode_cursor(cursor)
        self.pagination.base_url = "/"

        # Loading the page with this cursor doesn't crash
        (previous, current, next, previous_url, next_url) = self.get_pages(url)

        # The previous url doesn't crash either
        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        # And point to things that are not completely off.
        assert previous == [7, 7, 7, 7, 8]
        assert current == []
        assert next is None
        assert previous_url is not None
        assert next_url is None

    def test_cursor_pagination_with_page_size(self):
        (previous, current, next, previous_url, next_url) = self.get_pages('/?page_size=20')

        assert previous is None
        assert current == [1, 1, 1, 1, 1, 1, 2, 3, 4, 4, 4, 4, 5, 6, 7, 7, 7, 7, 7, 7]
        assert next == [7, 7, 7, 8, 9, 9, 9, 9, 9, 9]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)
        assert previous == [1, 1, 1, 1, 1, 1, 2, 3, 4, 4, 4, 4, 5, 6, 7, 7, 7, 7, 7, 7]
        assert current == [7, 7, 7, 8, 9, 9, 9, 9, 9, 9]
        assert next is None

    def test_cursor_pagination_with_page_size_over_limit(self):
        (previous, current, next, previous_url, next_url) = self.get_pages('/?page_size=30')

        assert previous is None
        assert current == [1, 1, 1, 1, 1, 1, 2, 3, 4, 4, 4, 4, 5, 6, 7, 7, 7, 7, 7, 7]
        assert next == [7, 7, 7, 8, 9, 9, 9, 9, 9, 9]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)
        assert previous == [1, 1, 1, 1, 1, 1, 2, 3, 4, 4, 4, 4, 5, 6, 7, 7, 7, 7, 7, 7]
        assert current == [7, 7, 7, 8, 9, 9, 9, 9, 9, 9]
        assert next is None

    def test_cursor_pagination_with_page_size_zero(self):
        (previous, current, next, previous_url, next_url) = self.get_pages('/?page_size=0')

        assert previous is None
        assert current == [1, 1, 1, 1, 1]
        assert next == [1, 2, 3, 4, 4]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [1, 1, 1, 1, 1]
        assert current == [1, 2, 3, 4, 4]
        assert next == [4, 4, 5, 6, 7]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [1, 2, 3, 4, 4]
        assert current == [4, 4, 5, 6, 7]
        assert next == [7, 7, 7, 7, 7]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [4, 4, 4, 5, 6]  # Paging artifact
        assert current == [7, 7, 7, 7, 7]
        assert next == [7, 7, 7, 8, 9]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [7, 7, 7, 7, 7]
        assert current == [7, 7, 7, 8, 9]
        assert next == [9, 9, 9, 9, 9]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [7, 7, 7, 8, 9]
        assert current == [9, 9, 9, 9, 9]
        assert next is None

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous == [7, 7, 7, 7, 7]
        assert current == [7, 7, 7, 8, 9]
        assert next == [9, 9, 9, 9, 9]

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous == [4, 4, 5, 6, 7]
        assert current == [7, 7, 7, 7, 7]
        assert next == [8, 9, 9, 9, 9]  # Paging artifact

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous == [1, 2, 3, 4, 4]
        assert current == [4, 4, 5, 6, 7]
        assert next == [7, 7, 7, 7, 7]

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous == [1, 1, 1, 1, 1]
        assert current == [1, 2, 3, 4, 4]
        assert next == [4, 4, 5, 6, 7]

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous is None
        assert current == [1, 1, 1, 1, 1]
        assert next == [1, 2, 3, 4, 4]

    def test_cursor_pagination_with_page_size_negative(self):
        (previous, current, next, previous_url, next_url) = self.get_pages('/?page_size=-5')

        assert previous is None
        assert current == [1, 1, 1, 1, 1]
        assert next == [1, 2, 3, 4, 4]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [1, 1, 1, 1, 1]
        assert current == [1, 2, 3, 4, 4]
        assert next == [4, 4, 5, 6, 7]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [1, 2, 3, 4, 4]
        assert current == [4, 4, 5, 6, 7]
        assert next == [7, 7, 7, 7, 7]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [4, 4, 4, 5, 6]  # Paging artifact
        assert current == [7, 7, 7, 7, 7]
        assert next == [7, 7, 7, 8, 9]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [7, 7, 7, 7, 7]
        assert current == [7, 7, 7, 8, 9]
        assert next == [9, 9, 9, 9, 9]

        (previous, current, next, previous_url, next_url) = self.get_pages(next_url)

        assert previous == [7, 7, 7, 8, 9]
        assert current == [9, 9, 9, 9, 9]
        assert next is None

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous == [7, 7, 7, 7, 7]
        assert current == [7, 7, 7, 8, 9]
        assert next == [9, 9, 9, 9, 9]

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous == [4, 4, 5, 6, 7]
        assert current == [7, 7, 7, 7, 7]
        assert next == [8, 9, 9, 9, 9]  # Paging artifact

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous == [1, 2, 3, 4, 4]
        assert current == [4, 4, 5, 6, 7]
        assert next == [7, 7, 7, 7, 7]

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous == [1, 1, 1, 1, 1]
        assert current == [1, 2, 3, 4, 4]
        assert next == [4, 4, 5, 6, 7]

        (previous, current, next, previous_url, next_url) = self.get_pages(previous_url)

        assert previous is None
        assert current == [1, 1, 1, 1, 1]
        assert next == [1, 2, 3, 4, 4]

    def test_get_paginated_response_schema(self):
        unpaginated_schema = {
            'type': 'object',
            'item': {
                'properties': {
                    'test-property': {
                        'type': 'integer',
                    },
                },
            },
        }

        assert self.pagination.get_paginated_response_schema(unpaginated_schema) == {
            'type': 'object',
            'required': ['results'],
            'properties': {
                'next': {
                    'type': 'string',
                    'nullable': True,
                    'format': 'uri',
                    'example': 'http://api.example.org/accounts/?cursor=cD00ODY%3D"'
                },
                'previous': {
                    'type': 'string',
                    'nullable': True,
                    'format': 'uri',
                    'example': 'http://api.example.org/accounts/?cursor=cj0xJnA9NDg3'
                },
                'results': unpaginated_schema,
            },
        }


class TestCursorPagination(CursorPaginationTestsMixin):
    """
    Unit tests for `pagination.CursorPagination`.
    """

    def setup_method(self):
        class MockObject:
            def __init__(self, idx):
                self.created = idx

        class MockQuerySet:
            def __init__(self, items):
                self.items = items

            def filter(self, created__gt=None, created__lt=None):
                if created__gt is not None:
                    return MockQuerySet([
                        item for item in self.items
                        if item.created > int(created__gt)
                    ])

                assert created__lt is not None
                return MockQuerySet([
                    item for item in self.items
                    if item.created < int(created__lt)
                ])

            def order_by(self, *ordering):
                if ordering[0].startswith('-'):
                    return MockQuerySet(list(reversed(self.items)))
                return self

            def __getitem__(self, sliced):
                return self.items[sliced]

        class ExamplePagination(pagination.CursorPagination):
            page_size = 5
            page_size_query_param = 'page_size'
            max_page_size = 20
            ordering = 'created'

        self.pagination = ExamplePagination()
        self.queryset = MockQuerySet([
            MockObject(idx) for idx in [
                1, 1, 1, 1, 1,
                1, 2, 3, 4, 4,
                4, 4, 5, 6, 7,
                7, 7, 7, 7, 7,
                7, 7, 7, 8, 9,
                9, 9, 9, 9, 9
            ]
        ])

    def get_pages(self, url):
        """
        Given a URL return a tuple of:

        (previous page, current page, next page, previous url, next url)
        """
        request = Request(factory.get(url))
        queryset = self.pagination.paginate_queryset(self.queryset, request)
        current = [item.created for item in queryset]

        next_url = self.pagination.get_next_link()
        previous_url = self.pagination.get_previous_link()

        if next_url is not None:
            request = Request(factory.get(next_url))
            queryset = self.pagination.paginate_queryset(self.queryset, request)
            next = [item.created for item in queryset]
        else:
            next = None

        if previous_url is not None:
            request = Request(factory.get(previous_url))
            queryset = self.pagination.paginate_queryset(self.queryset, request)
            previous = [item.created for item in queryset]
        else:
            previous = None

        return (previous, current, next, previous_url, next_url)


class CursorPaginationModel(models.Model):
    created = models.IntegerField()


class TestCursorPaginationWithValueQueryset(CursorPaginationTestsMixin, TestCase):
    """
    Unit tests for `pagination.CursorPagination` for value querysets.
    """

    def setUp(self):
        class ExamplePagination(pagination.CursorPagination):
            page_size = 5
            page_size_query_param = 'page_size'
            max_page_size = 20
            ordering = 'created'

        self.pagination = ExamplePagination()
        data = [
            1, 1, 1, 1, 1,
            1, 2, 3, 4, 4,
            4, 4, 5, 6, 7,
            7, 7, 7, 7, 7,
            7, 7, 7, 8, 9,
            9, 9, 9, 9, 9
        ]
        for idx in data:
            CursorPaginationModel.objects.create(created=idx)

        self.queryset = CursorPaginationModel.objects.values()

    def get_pages(self, url):
        """
        Given a URL return a tuple of:

        (previous page, current page, next page, previous url, next url)
        """
        request = Request(factory.get(url))
        queryset = self.pagination.paginate_queryset(self.queryset, request)
        current = [item['created'] for item in queryset]

        next_url = self.pagination.get_next_link()
        previous_url = self.pagination.get_previous_link()

        if next_url is not None:
            request = Request(factory.get(next_url))
            queryset = self.pagination.paginate_queryset(self.queryset, request)
            next = [item['created'] for item in queryset]
        else:
            next = None

        if previous_url is not None:
            request = Request(factory.get(previous_url))
            queryset = self.pagination.paginate_queryset(self.queryset, request)
            previous = [item['created'] for item in queryset]
        else:
            previous = None

        return (previous, current, next, previous_url, next_url)


def test_get_displayed_page_numbers():
    """
    Test our contextual page display function.

    This determines which pages to display in a pagination control,
    given the current page and the last page.
    """
    displayed_page_numbers = pagination._get_displayed_page_numbers

    # At five pages or less, all pages are displayed, always.
    assert displayed_page_numbers(1, 5) == [1, 2, 3, 4, 5]
    assert displayed_page_numbers(2, 5) == [1, 2, 3, 4, 5]
    assert displayed_page_numbers(3, 5) == [1, 2, 3, 4, 5]
    assert displayed_page_numbers(4, 5) == [1, 2, 3, 4, 5]
    assert displayed_page_numbers(5, 5) == [1, 2, 3, 4, 5]

    # Between six and either pages we may have a single page break.
    assert displayed_page_numbers(1, 6) == [1, 2, 3, None, 6]
    assert displayed_page_numbers(2, 6) == [1, 2, 3, None, 6]
    assert displayed_page_numbers(3, 6) == [1, 2, 3, 4, 5, 6]
    assert displayed_page_numbers(4, 6) == [1, 2, 3, 4, 5, 6]
    assert displayed_page_numbers(5, 6) == [1, None, 4, 5, 6]
    assert displayed_page_numbers(6, 6) == [1, None, 4, 5, 6]

    assert displayed_page_numbers(1, 7) == [1, 2, 3, None, 7]
    assert displayed_page_numbers(2, 7) == [1, 2, 3, None, 7]
    assert displayed_page_numbers(3, 7) == [1, 2, 3, 4, None, 7]
    assert displayed_page_numbers(4, 7) == [1, 2, 3, 4, 5, 6, 7]
    assert displayed_page_numbers(5, 7) == [1, None, 4, 5, 6, 7]
    assert displayed_page_numbers(6, 7) == [1, None, 5, 6, 7]
    assert displayed_page_numbers(7, 7) == [1, None, 5, 6, 7]

    assert displayed_page_numbers(1, 8) == [1, 2, 3, None, 8]
    assert displayed_page_numbers(2, 8) == [1, 2, 3, None, 8]
    assert displayed_page_numbers(3, 8) == [1, 2, 3, 4, None, 8]
    assert displayed_page_numbers(4, 8) == [1, 2, 3, 4, 5, None, 8]
    assert displayed_page_numbers(5, 8) == [1, None, 4, 5, 6, 7, 8]
    assert displayed_page_numbers(6, 8) == [1, None, 5, 6, 7, 8]
    assert displayed_page_numbers(7, 8) == [1, None, 6, 7, 8]
    assert displayed_page_numbers(8, 8) == [1, None, 6, 7, 8]

    # At nine or more pages we may have two page breaks, one on each side.
    assert displayed_page_numbers(1, 9) == [1, 2, 3, None, 9]
    assert displayed_page_numbers(2, 9) == [1, 2, 3, None, 9]
    assert displayed_page_numbers(3, 9) == [1, 2, 3, 4, None, 9]
    assert displayed_page_numbers(4, 9) == [1, 2, 3, 4, 5, None, 9]
    assert displayed_page_numbers(5, 9) == [1, None, 4, 5, 6, None, 9]
    assert displayed_page_numbers(6, 9) == [1, None, 5, 6, 7, 8, 9]
    assert displayed_page_numbers(7, 9) == [1, None, 6, 7, 8, 9]
    assert displayed_page_numbers(8, 9) == [1, None, 7, 8, 9]
    assert displayed_page_numbers(9, 9) == [1, None, 7, 8, 9]