diff --git a/uw_myplan/__init__.py b/uw_myplan/__init__.py index a8a160a..74539f8 100644 --- a/uw_myplan/__init__.py +++ b/uw_myplan/__init__.py @@ -14,68 +14,70 @@ MyPlan, MyPlanTerm, MyPlanCourse, MyPlanCourseSection) logger = logging.getLogger(__name__) -dao = MyPlan_DAO() -def _get_plan_url(regid, year, quarter, terms): - return "/plan/v1/{year},{quarter},{terms},{uwregid}".format( - year=year, quarter=quarter, terms=terms, uwregid=regid) +class Plan(object): + def __init__(self, actas=None): + self.dao = MyPlan_DAO() -def _get_resource(regid, year, quarter, terms, clear_cached_token=False): - if clear_cached_token: - dao.clear_access_token() - return dao.getURL( - _get_plan_url(regid, year, quarter, terms), - {"Accept": "application/json"}) + def _get_plan_url(self, regid, year, quarter, terms): + return "/plan/v1/{year},{quarter},{terms},{uwregid}".format( + year=year, quarter=quarter, terms=terms, uwregid=regid) + def _get_resource(self, regid, year, quarter, terms, + clear_cached_token=False): + if clear_cached_token: + self.dao.clear_access_token() + return self.dao.getURL( + self._get_plan_url(regid, year, quarter, terms), + {"Accept": "application/json"}) -def get_plan(regid, year, quarter, terms=4): - response = _get_resource(regid, year, quarter, terms) - if response.status == 200: - return _process_data(json.loads(response.data)) - - if response.status == 401 or response.status == 403: - # clear cached access token, retry once - response = _get_resource( - regid, year, quarter, terms, clear_cached_token=True) + def get_plan(self, regid, year, quarter, terms=4): + response = self._get_resource(regid, year, quarter, terms) if response.status == 200: - return _process_data(json.loads(response.data)) - - raise DataFailureException( - _get_plan_url(regid, year, quarter, terms), - response.status, str(response.data)) - - -def _process_data(jdata): - plan = MyPlan() - for term_data in jdata: - term = MyPlanTerm() - term.year = term_data["Term"]["Year"] - term.quarter = term_data["Term"]["Quarter"] - - term.course_search_href = term_data["CourseSearchHref"] - term.degree_audit_href = term_data["DegreeAuditHref"] - term.myplan_href = term_data["MyPlanHref"] - term.registration_href = term_data["RegistrationHref"] - term.registered_courses_count = int( - term_data["RegisteredCoursesCount"]) - term.registered_sections_count = int( - term_data["RegisteredSectionsCount"]) - - for course_data in term_data["Courses"]: - course = MyPlanCourse() - course.curriculum_abbr = course_data["CurriculumAbbreviation"] - course.course_number = course_data["CourseNumber"] - - is_available = course_data["RegistrationAvailable"] - course.registrations_available = is_available - - for section_data in course_data["Sections"]: - section = MyPlanCourseSection() - section.section_id = section_data["SectionId"] - course.sections.append(section) - - term.courses.append(course) - plan.terms.append(term) - return plan + return self._process_data(json.loads(response.data)) + + if response.status == 401 or response.status == 403: + # clear cached access token, retry once + response = self._get_resource( + regid, year, quarter, terms, clear_cached_token=True) + if response.status == 200: + return self._process_data(json.loads(response.data)) + + raise DataFailureException( + self._get_plan_url(regid, year, quarter, terms), + response.status, str(response.data)) + + def _process_data(self, jdata): + plan = MyPlan() + for term_data in jdata: + term = MyPlanTerm() + term.year = term_data["Term"]["Year"] + term.quarter = term_data["Term"]["Quarter"] + + term.course_search_href = term_data["CourseSearchHref"] + term.degree_audit_href = term_data["DegreeAuditHref"] + term.myplan_href = term_data["MyPlanHref"] + term.registration_href = term_data["RegistrationHref"] + term.registered_courses_count = int( + term_data["RegisteredCoursesCount"]) + term.registered_sections_count = int( + term_data["RegisteredSectionsCount"]) + + for course_data in term_data["Courses"]: + course = MyPlanCourse() + course.curriculum_abbr = course_data["CurriculumAbbreviation"] + course.course_number = course_data["CourseNumber"] + + is_available = course_data["RegistrationAvailable"] + course.registrations_available = is_available + + for section_data in course_data["Sections"]: + section = MyPlanCourseSection() + section.section_id = section_data["SectionId"] + course.sections.append(section) + + term.courses.append(course) + plan.terms.append(term) + return plan diff --git a/uw_myplan/tests/test_dao.py b/uw_myplan/tests/test_dao.py index 07ed2a7..2641b3c 100644 --- a/uw_myplan/tests/test_dao.py +++ b/uw_myplan/tests/test_dao.py @@ -5,6 +5,7 @@ import mock from commonconf import override_settings from restclients_core.exceptions import DataFailureException +from restclients_core.models import MockHTTP from uw_myplan.dao import MyPlan_Auth_DAO, MyPlan_DAO from uw_myplan.utils import ( fdao_myplan_override, fdao_myplan_auth_override) @@ -22,6 +23,16 @@ def test_get_auth_token(self): self.assertIsNotNone( MyPlan_Auth_DAO().get_auth_token("test1")) + @mock.patch.object(MyPlan_Auth_DAO, "postURL") + def test_get_auth_token(self, mock): + response = MockHTTP() + response.status = 404 + response.data = "Not Found" + mock.return_value = response + self.assertRaises( + DataFailureException, + MyPlan_Auth_DAO().get_auth_token, "test1") + def test_no_auth_header(self): headers = MyPlan_DAO()._custom_headers("GET", "/", {}, "") self.assertFalse("Authorization" in headers) diff --git a/uw_myplan/tests/test_myplan.py b/uw_myplan/tests/test_myplan.py index 8fab8c6..9f071c1 100644 --- a/uw_myplan/tests/test_myplan.py +++ b/uw_myplan/tests/test_myplan.py @@ -2,22 +2,40 @@ # SPDX-License-Identifier: Apache-2.0 from unittest import TestCase -from uw_myplan import get_plan, _get_plan_url, _get_resource +import mock +from restclients_core.models import MockHTTP +from restclients_core.exceptions import DataFailureException +from uw_myplan import Plan -class MyPlanTestData(TestCase): +class PlanTest(TestCase): def test_plan_url(self): self.assertEquals( - _get_plan_url( + Plan()._get_plan_url( "9136CCB8F66711D5BE060004AC494FFE", 2013, "spring", 2), ( "/plan/v1/2013,spring,2," "9136CCB8F66711D5BE060004AC494FFE")) + @mock.patch.object(Plan, "_get_resource") + def test_error_401(self, mock): + response = MockHTTP() + response.status = 403 + response.data = "Not Authorized" + mock.return_value = response + self.assertRaises( + DataFailureException, + Plan().get_plan, + "9136CCB8F66711D5BE060004AC494FFE", + 2013, + "spring", + terms=4) + def test_javerage(self): - plan = get_plan(regid="9136CCB8F66711D5BE060004AC494FFE", - year=2013, - quarter="spring", - terms=4) + plan = Plan().get_plan( + regid="9136CCB8F66711D5BE060004AC494FFE", + year=2013, + quarter="spring", + terms=4) self.assertEquals(len(plan.terms), 4) self.assertEquals(plan.terms[0].year, 2013) @@ -61,15 +79,17 @@ def test_javerage(self): self.assertEquals(term_data.courses[0].sections[1].section_id, 'AA') self.assertEquals(term_data.courses[0].sections[2].section_id, 'AB') - resp = _get_resource( + resp = Plan()._get_resource( "9136CCB8F66711D5BE060004AC494FFE", 2013, "spring", 4, clear_cached_token=True) self.assertIsNotNone(resp) def test_json(self): - plan = get_plan(regid="9136CCB8F66711D5BE060004AC494FFE", - year=2013, quarter="spring", - terms=4) + plan = Plan().get_plan( + regid="9136CCB8F66711D5BE060004AC494FFE", + year=2013, + quarter="spring", + terms=4) json_data = plan.json_data() term_data = json_data["terms"][0] self.assertEquals(