From 840f5017f4d7fd1f77d8e1b630a6982730d172c2 Mon Sep 17 00:00:00 2001 From: Ehtesham Date: Wed, 30 Dec 2015 13:58:19 +0500 Subject: [PATCH] [TNL-3940] Adding pagination in stub server and updating unit tests --- common/djangoapps/terrain/stubs/edxnotes.py | 71 ++++++- .../terrain/stubs/tests/test_edxnotes.py | 186 ++++++++++++++++-- 2 files changed, 232 insertions(+), 25 deletions(-) diff --git a/common/djangoapps/terrain/stubs/edxnotes.py b/common/djangoapps/terrain/stubs/edxnotes.py index 3e697f51e9..c5b4510558 100644 --- a/common/djangoapps/terrain/stubs/edxnotes.py +++ b/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() diff --git a/common/djangoapps/terrain/stubs/tests/test_edxnotes.py b/common/djangoapps/terrain/stubs/tests/test_edxnotes.py index 6225a559a7..19aa0969eb 100644 --- a/common/djangoapps/terrain/stubs/tests/test_edxnotes.py +++ b/common/djangoapps/terrain/stubs/tests/test_edxnotes.py @@ -1,7 +1,7 @@ """ 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()) - response = requests.get(self._get_url("api/v1/search")) - self.assertEqual(response.status_code, 400) + 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 + ) + + # 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) - 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) + # 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): + """ + 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