[TNL-3940] Adding pagination in stub server and updating unit tests

common/djangoapps/terrain/stubs/edxnotes.py

@@ -7,11 +7,12 @@ import re
 from uuid import uuid4
 from datetime import datetime
 from copy import deepcopy
+from math import ceil
+from urllib import urlencode
 
 from .http import StubHttpRequestHandler, StubHttpService
 
 
-# pylint: disable=invalid-name
 class StubEdxNotesServiceHandler(StubHttpRequestHandler):
     """
     Handler for EdxNotes requests.
@@ -165,7 +166,7 @@ class StubEdxNotesServiceHandler(StubHttpRequestHandler):
         """
         Return the note by note id.
         """
-        notes = self.server.get_notes()
+        notes = self.server.get_all_notes()
         result = self.server.filter_by_id(notes, note_id)
         if result:
             self.respond(content=result[0])
@@ -191,6 +192,53 @@ class StubEdxNotesServiceHandler(StubHttpRequestHandler):
         else:
             self.respond(404, "404 Not Found")
 
+    @staticmethod
+    def _get_next_prev_url(url_path, query_params, page_num, page_size):
+        """
+        makes url with the query params including pagination params
+        for pagination next and previous urls
+        """
+        query_params = deepcopy(query_params)
+        query_params.update({
+            "page": page_num,
+            "page_size": page_size
+        })
+        return url_path + "?" + urlencode(query_params)
+
+    def _get_paginated_response(self, notes, page_num, page_size):
+        """
+        Returns a paginated response of notes.
+        """
+        start = (page_num - 1) * page_size
+        end = start + page_size
+        total_notes = len(notes)
+        url_path = "http://{server_address}:{port}{path}".format(
+            server_address=self.client_address[0],
+            port=self.server.port,
+            path=self.path_only
+        )
+
+        next_url = None if end >= total_notes else self._get_next_prev_url(
+            url_path, self.get_params, page_num + 1, page_size
+        )
+        prev_url = None if page_num == 1 else self._get_next_prev_url(
+            url_path, self.get_params, page_num - 1, page_size)
+
+        # Get notes from range
+        notes = deepcopy(notes[start:end])
+
+        paginated_response = {
+            'total': total_notes,
+            'num_pages': int(ceil(float(total_notes) / page_size)),
+            'current_page': page_num,
+            'rows': notes,
+            'next': next_url,
+            'start': start,
+            'previous': prev_url
+        }
+
+        return paginated_response
+
     def _search(self):
         """
         Search for a notes by user id, course_id and usage_id.
@@ -199,32 +247,35 @@ class StubEdxNotesServiceHandler(StubHttpRequestHandler):
         usage_id = self.get_params.get("usage_id", None)
         course_id = self.get_params.get("course_id", None)
         text = self.get_params.get("text", None)
+        page = int(self.get_params.get("page", 1))
+        page_size = int(self.get_params.get("page_size", 2))
 
         if user is None:
             self.respond(400, "Bad Request")
             return
 
-        notes = self.server.get_notes()
+        notes = self.server.get_all_notes()
         if course_id is not None:
             notes = self.server.filter_by_course_id(notes, course_id)
         if usage_id is not None:
             notes = self.server.filter_by_usage_id(notes, usage_id)
         if text:
             notes = self.server.search(notes, text)
-        self.respond(content={
-            "total": len(notes),
-            "rows": notes,
-        })
+        self.respond(content=self._get_paginated_response(notes, page, page_size))
 
     def _collection(self):
         """
         Return all notes for the user.
         """
         user = self.get_params.get("user", None)
+        page = int(self.get_params.get("page", 1))
+        page_size = int(self.get_params.get("page_size", 2))
+        notes = self.server.get_all_notes()
+
         if user is None:
             self.send_response(400, content="Bad Request")
             return
-        notes = self.server.get_notes()
+        notes = self._get_paginated_response(notes, page, page_size)
         self.respond(content=notes)
 
     def _cleanup(self):
@@ -245,9 +296,9 @@ class StubEdxNotesService(StubHttpService):
         super(StubEdxNotesService, self).__init__(*args, **kwargs)
         self.notes = list()
 
-    def get_notes(self):
+    def get_all_notes(self):
         """
-        Returns a list of all notes.
+        Returns a list of all notes without pagination
         """
         notes = deepcopy(self.notes)
         notes.reverse()

common/djangoapps/terrain/stubs/tests/test_edxnotes.py

 """
 Unit tests for stub EdxNotes implementation.
 """
-
+import urlparse
 import json
 import unittest
 import requests
@@ -19,7 +19,7 @@ class StubEdxNotesServiceTest(unittest.TestCase):
         """
         super(StubEdxNotesServiceTest, self).setUp()
         self.server = StubEdxNotesService()
-        dummy_notes = self._get_dummy_notes(count=2)
+        dummy_notes = self._get_dummy_notes(count=5)
         self.server.add_notes(dummy_notes)
         self.addCleanup(self.server.shutdown)
 
@@ -99,17 +99,48 @@ class StubEdxNotesServiceTest(unittest.TestCase):
         self.assertEqual(response.status_code, 404)
 
     def test_search(self):
+        # Without user
+        response = requests.get(self._get_url("api/v1/search"))
+        self.assertEqual(response.status_code, 400)
+
+        # get response with default page and page size
         response = requests.get(self._get_url("api/v1/search"), params={
             "user": "dummy-user-id",
             "usage_id": "dummy-usage-id",
             "course_id": "dummy-course-id",
         })
-        notes = self._get_notes()
+
         self.assertTrue(response.ok)
-        self.assertDictEqual({"total": 2, "rows": notes}, response.json())
+        self._verify_pagination_info(
+            response=response.json(),
+            total_notes=5,
+            num_pages=3,
+            notes_per_page=2,
+            start=0,
+            current_page=1,
+            next_page=2,
+            previous_page=None
+        )
 
-        response = requests.get(self._get_url("api/v1/search"))
-        self.assertEqual(response.status_code, 400)
+        # search notes with text that don't exist
+        response = requests.get(self._get_url("api/v1/search"), params={
+            "user": "dummy-user-id",
+            "usage_id": "dummy-usage-id",
+            "course_id": "dummy-course-id",
+            "text": "world war 2"
+        })
+
+        self.assertTrue(response.ok)
+        self._verify_pagination_info(
+            response=response.json(),
+            total_notes=0,
+            num_pages=0,
+            notes_per_page=0,
+            start=0,
+            current_page=1,
+            next_page=None,
+            previous_page=None
+        )
 
     def test_delete(self):
         notes = self._get_notes()
@@ -119,7 +150,7 @@ class StubEdxNotesServiceTest(unittest.TestCase):
         for note in notes:
             response = requests.delete(self._get_url("api/v1/annotations/" + note["id"]))
             self.assertEqual(response.status_code, 204)
-            remaining_notes = self.server.get_notes()
+            remaining_notes = self.server.get_all_notes()
             self.assertNotIn(note["id"], [note["id"] for note in remaining_notes])
 
         self.assertEqual(len(remaining_notes), 0)
@@ -139,24 +170,149 @@ class StubEdxNotesServiceTest(unittest.TestCase):
         response = requests.get(self._get_url("api/v1/annotations/does_not_exist"))
         self.assertEqual(response.status_code, 404)
 
+    # pylint: disable=too-many-arguments
+    def _verify_pagination_info(
+            self,
+            response,
+            total_notes,
+            num_pages,
+            notes_per_page,
+            current_page,
+            previous_page,
+            next_page,
+            start
+    ):
+        """
+        Verify the pagination information.
+
+        Argument:
+            response: response from api
+            total_notes: total notes in the response
+            num_pages: total number of pages in response
+            notes_per_page: number of notes in the response
+            current_page: current page number
+            previous_page: previous page number
+            next_page: next page number
+            start: start of the current page
+        """
+        def get_page_value(url):
+            """
+            Return page value extracted from url.
+            """
+            if url is None:
+                return None
+
+            parsed = urlparse.urlparse(url)
+            query_params = urlparse.parse_qs(parsed.query)
+
+            page = query_params["page"][0]
+            return page if page is None else int(page)
+
+        self.assertEqual(response["total"], total_notes)
+        self.assertEqual(response["num_pages"], num_pages)
+        self.assertEqual(len(response["rows"]), notes_per_page)
+        self.assertEqual(response["current_page"], current_page)
+        self.assertEqual(get_page_value(response["previous"]), previous_page)
+        self.assertEqual(get_page_value(response["next"]), next_page)
+        self.assertEqual(response["start"], start)
+
     def test_notes_collection(self):
-        response = requests.get(self._get_url("api/v1/annotations"), params={"user": "dummy-user-id"})
-        self.assertTrue(response.ok)
-        self.assertEqual(len(response.json()), 2)
+        """
+        Test paginated response of notes api
+        """
 
+        # Without user
         response = requests.get(self._get_url("api/v1/annotations"))
         self.assertEqual(response.status_code, 400)
 
+        # Without any pagination parameters
+        response = requests.get(self._get_url("api/v1/annotations"), params={"user": "dummy-user-id"})
+
+        self.assertTrue(response.ok)
+        self._verify_pagination_info(
+            response=response.json(),
+            total_notes=5,
+            num_pages=3,
+            notes_per_page=2,
+            start=0,
+            current_page=1,
+            next_page=2,
+            previous_page=None
+        )
+
+        # With pagination parameters
+        response = requests.get(self._get_url("api/v1/annotations"), params={
+            "user": "dummy-user-id",
+            "page": 2,
+            "page_size": 3
+        })
+
+        self.assertTrue(response.ok)
+        self._verify_pagination_info(
+            response=response.json(),
+            total_notes=5,
+            num_pages=2,
+            notes_per_page=2,
+            start=3,
+            current_page=2,
+            next_page=None,
+            previous_page=1
+        )
+
+    def test_notes_collection_next_previous_with_one_page(self):
+        """
+        Test next and previous urls of paginated response of notes api
+        when number of pages are 1
+        """
+        response = requests.get(self._get_url("api/v1/annotations"), params={
+            "user": "dummy-user-id",
+            "page_size": 10
+        })
+
+        self.assertTrue(response.ok)
+        self._verify_pagination_info(
+            response=response.json(),
+            total_notes=5,
+            num_pages=1,
+            notes_per_page=5,
+            start=0,
+            current_page=1,
+            next_page=None,
+            previous_page=None
+        )
+
+    def test_notes_collection_when_no_notes(self):
+        """
+        Test paginated response of notes api when there's no note present
+        """
+
+        # Delete all notes
+        self.test_cleanup()
+
+        # Get default page
+        response = requests.get(self._get_url("api/v1/annotations"), params={"user": "dummy-user-id"})
+        self.assertTrue(response.ok)
+        self._verify_pagination_info(
+            response=response.json(),
+            total_notes=0,
+            num_pages=0,
+            notes_per_page=0,
+            start=0,
+            current_page=1,
+            next_page=None,
+            previous_page=None
+        )
+
     def test_cleanup(self):
         response = requests.put(self._get_url("cleanup"))
         self.assertTrue(response.ok)
-        self.assertEqual(len(self.server.get_notes()), 0)
+        self.assertEqual(len(self.server.get_all_notes()), 0)
 
     def test_create_notes(self):
         dummy_notes = self._get_dummy_notes(count=2)
         response = requests.post(self._get_url("create_notes"), data=json.dumps(dummy_notes))
         self.assertTrue(response.ok)
-        self.assertEqual(len(self._get_notes()), 4)
+        self.assertEqual(len(self._get_notes()), 7)
 
         response = requests.post(self._get_url("create_notes"))
         self.assertEqual(response.status_code, 400)
@@ -177,7 +333,7 @@ class StubEdxNotesServiceTest(unittest.TestCase):
         """
         Return a list of notes from the stub EdxNotes service.
         """
-        notes = self.server.get_notes()
+        notes = self.server.get_all_notes()
         self.assertGreater(len(notes), 0, "Notes are empty.")
         return notes