diff --git a/docs/api-reference/TEXTBOOK-API.md b/docs/api-reference/TEXTBOOK-API.md index 248aeae..d05af5d 100644 --- a/docs/api-reference/TEXTBOOK-API.md +++ b/docs/api-reference/TEXTBOOK-API.md @@ -1,104 +1,25 @@ > [Home](README.md) > Textbook API ---- # Textbook API -### **get_textbook(term, department, course, instructor, section)** +Textbook terms are discovered from the bookstore instead of being embedded in PittAPI. Select a term before looking +up a course: -#### **Parameters**: - - `term`: Term number | Example: `2671` - - `department`: Department code | Example: `CS` - - `course`: Course number | Example: `0401`, `411` - - `instructor`: Instructor name | Example: `GARRISON III`, `YANG`, `HOFFMAN` - - `section`: Section number | Example: `1030`, `1060` - -#### **Returns**: -Returns a list of dictionaries containing Author, ISBN, Edition, Title, and Citation - -#### **Example**: - -###### **Code**: -```python -get_textbook( - term='2671, - department='CS', - course='445', - instructor='GARRISON III' -) - -get_textbook( - term='2671', - department='CS', - course='401', - section='1010' -) -``` - -###### **Sample Output**: ```python - [ - { - 'author': 'Carrano', - 'citation': 'Data Struct.+Abstract.W/Java-W/Access by Carrano. ' - 'Pearson Education, 4th Edition, 2014. (ISBN: 9780133744057).', - 'edition': '4', - 'isbn': '9780133744057', - 'title': 'Data Struct.+Abstract.W/Java-W/Access' - } - ] - - [ - { - 'author': 'Gaddis', - 'citation': 'Starting Out W/Java:From..-W/Access by Gaddis. Pearson ' - 'Education, 6th Edition, 2015. (ISBN: 9780133957051).', - 'edition': '6', - 'isbn': '9780133957051', - 'title': 'Starting Out W/Java:From...-W/Access' - } - ] +from pittapi import TextbookClient +from pittapi.textbook import CourseInfo + +with TextbookClient() as textbooks: + terms = textbooks.get_terms() + fall = next(term for term in terms if term.name == "Fall 26") + textbooks.select_term(fall) + books = textbooks.get_textbooks_for_course( + CourseInfo("CS", "0441", instructor="GARRISON III") + ) ``` -### **get_textbooks(term, courses)** +A term can also be supplied to `TextbookClient(term=...)`. Changing it with `select_term()` clears cached +term-specific subjects. `get_textbooks_for_courses()` accepts a list or tuple and preserves course order. -#### **Parameters**: - - `term`: Term number | Example: `2671` - - `courses`: List of dictionaries of class info | Example: `[{'department': 'CS', 'course': '0401', 'instructor': 'HOFFMAN'}]` - -#### **Returns**: -Returns a list of dictionaries containing Author, ISBN, Edition, Title, and Citation - -#### **Example**: - -###### **Code**: -```python -get_textbooks( - term='2671', - courses=[ - {'department': 'CS', 'course': '445', 'section': '1010'}, - {'department': 'STAT', 'course': '1000', 'instructor': 'REGISTER'} - ] -) -``` - -###### **Sample Output**: -```python - [ - { - 'author': 'Moore', - 'citation': 'Intro.To Practice Of Stat.-W/Access by Moore. Freeman - & Company, W. H., 8th Edition, 2014. (ISBN: 9781464158933).', - 'edition': '8', - 'isbn': '9781464158933', - 'title': 'Intro.To Practice Of Stat.-W/Access' - }, - { - 'author': 'Carrano', - 'citation': 'Data Struct.+Abstract.W/Java-W/Access by Carrano. - Pearson Education, 4th Edition, 2014. (ISBN: 9780133744057).', - 'edition': '4', - 'isbn': '9780133744057', - 'title': 'Data Struct.+Abstract.W/Java-W/Access' - } - ] -``` +`CourseInfo` normalizes the subject and instructor to uppercase and pads course numbers to four digits. Specify an +instructor or four-digit section number when a course has multiple distinguishable sections. diff --git a/pittapi/course.py b/pittapi/course.py index 5e34969..27d835c 100644 --- a/pittapi/course.py +++ b/pittapi/course.py @@ -15,17 +15,31 @@ You should have received a copy of the GNU General Public License along with this program; if not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +Course catalog and section information from Pitt PeopleSoft. """ from __future__ import annotations +from dataclasses import dataclass import re -import requests -from typing import NamedTuple, Any - -JSON = dict[str, Any] +from typing import Any + +from pittapi.base_client import BaseClient + +__all__ = [ + "Attribute", + "Component", + "Course", + "CourseClient", + "CourseDetails", + "Instructor", + "Meeting", + "Section", + "SectionDetails", + "Subject", +] -# https://pitcsprd.csps.pitt.edu/psc/pitcsprd/EMPLOYEE/SA/s/WEBLIB_HCX_CM.H_CLASS_SEARCH.FieldFormula.IScript_ClassSearch?institution=UPITT&term=2244&date_from=&date_thru=&subject=CS&subject_like=&catalog_nbr=&time_range=&days=&campus=PIT&location=&x_acad_career=UGRD&acad_group=&rqmnt_designtn=&instruction_mode=&keyword=&class_nbr=&acad_org=&enrl_stat=O&crse_attr=&crse_attr_value=&instructor_name=&instr_first_name=&session_code=&units=&trigger_search=&page=1 SUBJECTS_API = ( "https://pitcsprd.csps.pitt.edu/psc/pitcsprd/EMPLOYEE/SA/s/" "WEBLIB_HCX_CM.H_COURSE_CATALOG.FieldFormula.IScript_CatalogSubjects?institution=UPITT" @@ -37,380 +51,325 @@ COURSE_DETAIL_API = ( "https://pitcsprd.csps.pitt.edu/psc/pitcsprd/EMPLOYEE/SA/s/" "WEBLIB_HCX_CM.H_COURSE_CATALOG.FieldFormula.IScript_CatalogCourseDetails?institution=UPITT&course_id={id}" - "&effdt=2018-06-30&crse_offer_nbr=1&use_catalog_print=Y" + "&crse_offer_nbr=1&use_catalog_print=Y" ) COURSE_SECTIONS_API = ( "https://pitcsprd.csps.pitt.edu/psc/pitcsprd/EMPLOYEE/SA/s/" - "WEBLIB_HCX_CM.H_BROWSE_CLASSES.FieldFormula.IScript_BrowseSections?institution=UPITT&campus=&location=&course_id={id}" - "&institution=UPITT&term={term}&crse_offer_nbr=1" + "WEBLIB_HCX_CM.H_BROWSE_CLASSES.FieldFormula.IScript_BrowseSections?institution=UPITT&campus=&location=" + "&course_id={id}&institution=UPITT&term={term}&crse_offer_nbr=1" ) SECTION_DETAILS_API = ( "https://pitcsprd.csps.pitt.edu/psc/pitcsprd/EMPLOYEE/SA/s/" "WEBLIB_HCX_CM.H_CLASS_SEARCH.FieldFormula.IScript_ClassDetails?institution=UPITT&term={term}&class_nbr={id}" ) -# id -> unique course ID, not to be confused with course code (for instance, CS 0007 has code 105611) -# career -> for example, UGRD (undergraduate) +VALID_TERM = re.compile(r"2\d\d[147]\Z") -TERM_REGEX = r"2\d\d[147]" -VALID_TERMS = re.compile(TERM_REGEX) - -class Instructor(NamedTuple): +@dataclass(frozen=True, slots=True) +class Instructor: name: str email: str | None = None -class Meeting(NamedTuple): +@dataclass(frozen=True, slots=True) +class Meeting: days: str start_time: str end_time: str start_date: str end_date: str - instructors: list[Instructor] | None = None + instructors: tuple[Instructor, ...] = () -class Attribute(NamedTuple): +@dataclass(frozen=True, slots=True) +class Attribute: attribute: str attribute_description: str value: str value_description: str -class Component(NamedTuple): +@dataclass(frozen=True, slots=True) +class Component: component: str required: bool -class SectionDetails(NamedTuple): +@dataclass(frozen=True, slots=True) +class SectionDetails: units: str - class_capacity: str enrollment_total: str enrollment_available: str wait_list_capacity: str wait_list_total: str valid_to_enroll: str + combined_section_numbers: tuple[str, ...] = () - combined_section_numbers: list[str] | None = None - -class Section(NamedTuple): +@dataclass(frozen=True, slots=True) +class Section: term: str session: str section_number: str class_number: str section_type: str status: str - instructors: list[Instructor] | None = None - meetings: list[Meeting] | None = None + instructors: tuple[Instructor, ...] = () + meetings: tuple[Meeting, ...] = () details: SectionDetails | None = None -class Course(NamedTuple): +@dataclass(frozen=True, slots=True) +class Course: subject_code: str course_number: str course_id: str course_title: str -class CourseDetails(NamedTuple): +@dataclass(frozen=True, slots=True) +class CourseDetails: course: Course course_description: str | None = None credit_range: tuple[int, int] | None = None requisites: str | None = None - components: list[Component] | None = None - attributes: list[Attribute] | None = None - sections: list[Section] | None = None + components: tuple[Component, ...] = () + attributes: tuple[Attribute, ...] = () + sections: tuple[Section, ...] = () -class Subject(NamedTuple): +@dataclass(frozen=True, slots=True) +class Subject: subject_code: str - courses: dict[str, Course] - - -def get_subject_courses(subject: str) -> Subject: - subject = _validate_subject(subject) - - json_response = _get_subject_courses(subject) - - courses = {} - for course in json_response["courses"]: - course_number = course["catalog_nbr"] - course_id = course["crse_id"] - course_title = course["descr"] - - course_obj = Course( - subject_code=subject, - course_number=course_number, - course_id=course_id, - course_title=course_title, - ) - - courses[course_number] = course_obj - - return Subject(subject_code=subject, courses=courses) - - -def get_course_details(term: str | int, subject: str, course: str | int) -> CourseDetails: - term = _validate_term(term) - subject = _validate_subject(subject) - course = _validate_course(course) - - internal_course_id = _get_course_id(subject, course) - json_response = _get_course_info(internal_course_id)["course_details"] - json_response_details = _get_course_sections(internal_course_id, term) - - course_title = json_response_details["sections"][0]["descr"] - course_description = json_response["descrlong"] - credit_range = (json_response["units_minimum"], json_response["units_maximum"]) + courses: tuple[Course, ...] + + +class CourseClient(BaseClient): + """Fetch courses and sections from Pitt's PeopleSoft catalog.""" + + def get_subject_courses(self, subject: str) -> Subject: + normalized_subject = self.validate_subject(subject) + data = self.get_subject_course_data(normalized_subject) + try: + courses = tuple(parse_course(item, normalized_subject) for item in data["courses"]) + except (KeyError, TypeError) as error: + raise ValueError("subject course response is missing required data") from error + return Subject(subject_code=normalized_subject, courses=courses) + + def get_course_details(self, term: str | int, subject: str, course: str | int) -> CourseDetails: + normalized_term = validate_term(term) + normalized_subject = self.validate_subject(subject) + normalized_course = validate_course_number(course) + course_id = self.find_course_id(normalized_subject, normalized_course) + + catalog_data = self.get_course_data(course_id) + section_data = self.get_course_section_data(course_id, normalized_term) + try: + return parse_course_details( + catalog_data["course_details"], + section_data["sections"], + normalized_term, + normalized_subject, + normalized_course, + course_id, + ) + except (KeyError, IndexError, TypeError) as error: + raise ValueError("course response is missing required data") from error + + def get_section_details(self, term: str | int, class_number: str | int) -> Section: + normalized_term = validate_term(term) + data = self.get_section_data(normalized_term, class_number) + try: + return parse_section_details(data["section_info"], normalized_term, str(class_number)) + except (KeyError, IndexError, TypeError, ValueError) as error: + raise ValueError("section response is missing required data") from error + + def validate_subject(self, subject: str) -> str: + normalized_subject = subject.upper() + data = self.request("GET", SUBJECTS_API).json() + try: + subject_codes = {item["subject"] for item in data["subjects"]} + except (KeyError, TypeError) as error: + raise ValueError("subject response is missing required data") from error + if normalized_subject not in subject_codes: + raise ValueError(f"invalid subject code: {subject}") + return normalized_subject + + def find_course_id(self, subject: str, course_number: str) -> str: + data = self.get_subject_course_data(subject) + try: + for course in data["courses"]: + if course["catalog_nbr"] == course_number: + return course["crse_id"] + except (KeyError, TypeError) as error: + raise ValueError("subject course response is missing required data") from error + raise LookupError(f"course not found: {subject} {course_number}") + + def get_subject_course_data(self, subject: str) -> dict[str, Any]: + return self.request("GET", SUBJECT_COURSES_API.format(subject=subject)).json() + + def get_course_data(self, course_id: str) -> dict[str, Any]: + data = self.request("GET", COURSE_DETAIL_API.format(id=course_id)).json() + if not data.get("course_details"): + raise LookupError(f"course ID not found: {course_id}") + return data + + def get_course_section_data(self, course_id: str, term: str) -> dict[str, Any]: + data = self.request("GET", COURSE_SECTIONS_API.format(id=course_id, term=term)).json() + if not data.get("sections"): + raise LookupError(f"no sections found for course ID {course_id} in term {term}") + return data + + def get_section_data(self, term: str, class_number: str | int) -> dict[str, Any]: + data = self.request("GET", SECTION_DETAILS_API.format(term=term, id=class_number)).json() + if "error" in data: + raise LookupError(f"section not found: {class_number}") + return data + + +def validate_term(term: str | int) -> str: + term_text = str(term) + if not VALID_TERM.fullmatch(term_text): + raise ValueError("term must be a four-digit Pitt term ending in 1, 4, or 7") + return term_text + + +def validate_course_number(course: str | int) -> str: + course_text = str(course) + if not course_text.isdigit() or int(course_text) <= 0: + raise ValueError("course number must be a positive number") + if len(course_text) > 4: + raise ValueError("course number cannot exceed four digits") + return course_text.zfill(4) + + +def parse_course(data: dict[str, Any], subject: str) -> Course: + return Course( + subject_code=subject, + course_number=data["catalog_nbr"], + course_id=data["crse_id"], + course_title=data["descr"], + ) - requisites = None - if "offerings" in json_response and len(json_response["offerings"]) != 0 and "req_group" in json_response["offerings"][0]: - requisites = json_response["offerings"][0]["req_group"] - components = None - if "components" in json_response and len(json_response["components"]) != 0: - components = [ - Component( - component=component["descr"], - required=True if component["optional"] == "N" else False, - ) - for component in json_response["components"] - ] - - attributes = None - if "attributes" in json_response and len(json_response["attributes"]) != 0: - attributes = [ - Attribute( - attribute=attribute["crse_attribute"], - attribute_description=attribute["crse_attribute_descr"], - value=attribute["crse_attribute_value"], - value_description=attribute["crse_attribute_value_descr"], - ) - for attribute in json_response["attributes"] - ] - - sections = [] - for section in json_response_details["sections"]: - session = section["session"] - section_number = section["class_section"] - class_number = str(section["class_nbr"]) - section_type = section["section_type"] - status = section["enrl_stat_descr"] - - instructors = None - if len(section["instructors"]) != 0 and section["instructors"][0] != "To be Announced": - instructors = [ - Instructor(name=instructor["name"], email=instructor["email"]) for instructor in section["instructors"] - ] - - meetings = None - if len(section["meetings"]) != 0: - meetings = [ - Meeting( - days=meeting["days"], - start_time=meeting["start_time"], - end_time=meeting["end_time"], - start_date=meeting["start_dt"], - end_date=meeting["end_dt"], - instructors=[Instructor(name=meeting["instructor"])], - ) - for meeting in section["meetings"] - ] - - sections.append( - Section( - term=term, - session=session, - section_number=section_number, - class_number=class_number, - section_type=section_type, - status=status, - instructors=instructors, - meetings=meetings, - ) +def parse_course_details( + catalog: dict[str, Any], + sections: list[dict[str, Any]], + term: str, + subject: str, + course_number: str, + course_id: str, +) -> CourseDetails: + course = Course( + subject_code=subject, + course_number=course_number, + course_id=course_id, + course_title=sections[0]["descr"], + ) + offerings = catalog.get("offerings", ()) + requisites = offerings[0].get("req_group") if offerings else None + components = tuple( + Component(component=item["descr"], required=item["optional"] == "N") for item in catalog.get("components", ()) + ) + attributes = tuple( + Attribute( + attribute=item["crse_attribute"], + attribute_description=item["crse_attribute_descr"], + value=item["crse_attribute_value"], + value_description=item["crse_attribute_value_descr"], ) - + for item in catalog.get("attributes", ()) + ) + parsed_sections = tuple(parse_catalog_section(item, term) for item in sections) return CourseDetails( - course=Course( - subject_code=subject, - course_number=course, - course_id=internal_course_id, - course_title=course_title, - ), - course_description=course_description, - credit_range=credit_range, + course=course, + course_description=catalog.get("descrlong"), + credit_range=(catalog["units_minimum"], catalog["units_maximum"]), requisites=requisites, components=components, attributes=attributes, - sections=sections, + sections=parsed_sections, ) -def get_section_details(term: str | int, class_number: str | int) -> Section: - term = _validate_term(term) - - json_response = _get_section_details(term, class_number) - details = json_response["section_info"]["class_details"] - meetings = json_response["section_info"]["meetings"] - enrollment = json_response["section_info"]["class_availability"] - - session = details["session"] - section_num = details["class_section"] - section_type = details["component"] - status = details["status"] - - meeting_objs = None - if len(meetings) != 0: - meeting_objs = [] - for meeting in meetings: - days = meeting["days"] - start_time = meeting["meeting_time_start"] - end_time = meeting["meeting_time_end"] - # start_date = meeting["start_date"] - # end_date = meeting["end_date"] - date_range = meeting["date_range"].split(" - ") - - instructors = None - if len(meeting["instructors"]) != 0 and meeting["instructors"][0]["name"] not in ["To be Announced", "-"]: - instructors = [] - for instructor in meeting["instructors"]: - name = instructor["name"] - email = instructor["email"] - - instructors.append(Instructor(name=name, email=email)) - - meeting_objs.append( - Meeting( - days=days, - start_time=start_time, - end_time=end_time, - start_date=date_range[0], - end_date=date_range[1], - instructors=instructors, - ) +def parse_catalog_section(data: dict[str, Any], term: str) -> Section: + instructors = parse_instructors(data.get("instructors", ())) + meetings = [] + for item in data.get("meetings", ()): + meeting_instructors = () + if item.get("instructor"): + meeting_instructors = (Instructor(name=item["instructor"]),) + meetings.append( + Meeting( + days=item["days"], + start_time=item["start_time"], + end_time=item["end_time"], + start_date=item["start_dt"], + end_date=item["end_dt"], + instructors=meeting_instructors, ) + ) + return Section( + term=term, + session=data["session"], + section_number=data["class_section"], + class_number=str(data["class_nbr"]), + section_type=data["section_type"], + status=data["enrl_stat_descr"], + instructors=instructors, + meetings=tuple(meetings), + ) - units = details["units"] - class_capacity = enrollment["class_capacity"] - enrollment_total = enrollment["enrollment_total"] - enrollment_available = str(enrollment["enrollment_available"]) - wait_list_capacity = enrollment["wait_list_capacity"] - wait_list_total = enrollment["wait_list_total"] - valid_to_enroll = json_response["section_info"]["valid_to_enroll"] - combined_section_numbers = None - if json_response["section_info"]["is_combined"]: - combined_section_numbers = [] - for section in json_response["section_info"]["combined_sections"]: - combined_section_numbers.append(section["class_nbr"]) + +def parse_section_details(section_info: dict[str, Any], term: str, class_number: str) -> Section: + details_data = section_info["class_details"] + enrollment = section_info["class_availability"] + combined_numbers = () + if section_info["is_combined"]: + combined_numbers = tuple(str(item["class_nbr"]) for item in section_info["combined_sections"]) details = SectionDetails( - units=units, - class_capacity=class_capacity, - enrollment_total=enrollment_total, - enrollment_available=enrollment_available, - wait_list_capacity=wait_list_capacity, - wait_list_total=wait_list_total, - valid_to_enroll=valid_to_enroll, - combined_section_numbers=combined_section_numbers, + units=details_data["units"], + class_capacity=enrollment["class_capacity"], + enrollment_total=enrollment["enrollment_total"], + enrollment_available=str(enrollment["enrollment_available"]), + wait_list_capacity=enrollment["wait_list_capacity"], + wait_list_total=enrollment["wait_list_total"], + valid_to_enroll=section_info["valid_to_enroll"], + combined_section_numbers=combined_numbers, ) - + meetings = tuple(parse_detailed_meeting(item) for item in section_info["meetings"]) return Section( term=term, - session=session, - section_number=section_num, - class_number=str(class_number), - section_type=section_type, - status=status, - instructors=None, - meetings=meeting_objs, + session=details_data["session"], + section_number=details_data["class_section"], + class_number=class_number, + section_type=details_data["component"], + status=details_data["status"], + meetings=meetings, details=details, ) -# validation for method inputs -def _validate_term(term: str | int) -> str: - """Validates that the term entered follows the pattern that Pitt does for term codes.""" - if VALID_TERMS.match(str(term)): - return str(term) - raise ValueError("Term entered isn't a valid Pitt term, must match regex " + TERM_REGEX) - - -def _validate_subject(subject: str) -> str: - """Validates that the subject code entered is present in the API request.""" - if subject in _get_subject_codes(): - return subject - raise ValueError("Subject code entered isn't a valid Pitt subject code.") - - -def _validate_course(course: str | int) -> str: - """Validates that the course name entered is 4 characters long and in string form.""" - if course == "": - raise ValueError("Invalid course number, please enter a non-empty string.") - if (type(course) is str) and (not course.isdigit()): - raise ValueError("Invalid course number, must be a number") - if (type(course) is int) and (course <= 0): - raise ValueError("Invalid course number, must be positive") - course_length = len(str(course)) - if course_length < 4: - return ("0" * (4 - course_length)) + str(course) - elif course_length > 4: - raise ValueError("Invalid course number, must be 4 characters long") - return str(course) - - -# peoplesoft api calls -def _get_subjects() -> JSON: - response: JSON = requests.get(SUBJECTS_API).json() - return response - - -def _get_subject_courses(subject: str) -> JSON: - response: JSON = requests.get(SUBJECT_COURSES_API.format(subject=subject)).json() - return response - - -def _get_course_info(course_id: str) -> JSON: - response: JSON = requests.get(COURSE_DETAIL_API.format(id=course_id)).json() - if response["course_details"] == {}: - raise ValueError("Invalid course ID; course with that ID does not exist") - return response - - -def _get_course_sections(course_id: str, term: str) -> JSON: - response: JSON = requests.get(COURSE_SECTIONS_API.format(id=course_id, term=term)).json() - if len(response["sections"]) == 0: - raise ValueError("Invalid course ID; course with that ID does not exist") - return response - - -def _get_section_details(term: str | int, section_id: str | int) -> JSON: - response: JSON = requests.get(SECTION_DETAILS_API.format(term=term, id=section_id)).json() - if "error" in response: - raise ValueError("Invalid section ID; section with that ID does not exist") - return response - - -# operations from api calls -def _get_subject_codes() -> list[str]: - response = _get_subjects() - codes = [] - for subject in response["subjects"]: - codes.append(subject["subject"]) - return codes - - -def _get_internal_id_dict(subject: str) -> dict[str, str]: - response = _get_subject_courses(subject) - internal_id_dict = {} - for course in response["courses"]: - if course["catalog_nbr"] not in internal_id_dict: - internal_id_dict[course["catalog_nbr"]] = course["crse_id"] - return internal_id_dict +def parse_detailed_meeting(data: dict[str, Any]) -> Meeting: + start_date, end_date = data["date_range"].split(" - ") + return Meeting( + days=data["days"], + start_time=data["meeting_time_start"], + end_time=data["meeting_time_end"], + start_date=start_date, + end_date=end_date, + instructors=parse_instructors(data["instructors"]), + ) -def _get_course_id(subject: str, course: str) -> str: - subject_dict = _get_internal_id_dict(subject) - if str(course) not in subject_dict: - raise ValueError("No course with that number within listed subject") - return subject_dict[str(course)] +def parse_instructors(instructors: list[dict[str, Any]] | tuple[Any, ...]) -> tuple[Instructor, ...]: + if not instructors or instructors[0] == "To be Announced": + return () + parsed = [] + for instructor in instructors: + if instructor["name"] in ("To be Announced", "-"): + continue + parsed.append(Instructor(name=instructor["name"], email=instructor.get("email"))) + return tuple(parsed) diff --git a/pittapi/sports.py b/pittapi/sports.py index 8ab5c39..70d4b64 100644 --- a/pittapi/sports.py +++ b/pittapi/sports.py @@ -15,146 +15,135 @@ You should have received a copy of the GNU General Public License along with this program; if not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +Pitt football and men's basketball information from ESPN. """ -from __future__ import annotations +from dataclasses import dataclass +from typing import Any -import requests +from pittapi.base_client import BaseClient -from typing import Any, NamedTuple +__all__ = ["Address", "GameInfo", "SportsClient", "Team", "Venue"] -JSON = dict[str, Any] +FOOTBALL_URL = "https://site.api.espn.com/apis/site/v2/sports/football/college-football/teams/pitt" +MENS_BASKETBALL_URL = "https://site.api.espn.com/apis/site/v2/sports/basketball/mens-college-basketball/teams/pittsburgh" +NO_RECORD = "There's no record right now." -FOOTBALL_URL = "http://site.api.espn.com/apis/site/v2/sports/football/college-football/teams/pitt" -MENS_BASKETBALL_URL = "http://site.api.espn.com/apis/site/v2/sports/basketball/mens-college-basketball/teams/pittsburgh" +@dataclass(frozen=True, slots=True) +class Team: + id: str + school: str + name: str -class GameInfo(NamedTuple): - timestamp: str | None = None - opponent: dict[str, str] | None = None - home_away: str | None = None - location: dict[str, str] | None = None - status: str | None = None +@dataclass(frozen=True, slots=True) +class Address: + city: str + state: str -def get_mens_basketball_record() -> str: - """returns the current record of the men's basketball team""" - basketball_data = _get_mens_basketball_data() - try: - record_summary: str = basketball_data["team"]["record"]["items"][0]["summary"] - except KeyError: - record_summary = "There's no record right now." +@dataclass(frozen=True, slots=True) +class Venue: + full_name: str + address: Address - return record_summary +@dataclass(frozen=True, slots=True) +class GameInfo: + timestamp: str | None = None + opponent: Team | None = None + home_away: str | None = None + location: Venue | None = None + status: str = "SCHEDULED" -def get_next_mens_basketball_game() -> GameInfo: - """returns a dict containing details of the next scheduled men's basketball game.""" - basketball_data = _get_mens_basketball_data() - next_game = None - try: - next_game = basketball_data["team"]["nextEvent"][0] - opponent = None - homeaway = None - status = None - if next_game["competitions"][0]["status"]["type"]["name"] == "STATUS_FINAL": - status = "GAME_COMPLETE" - elif next_game["competitions"][0]["status"]["type"]["name"] == "STATUS_IN_PROGRESS": - status = "IN_PROGRESS" - if str(next_game["competitions"][0]["competitors"][0]["id"]) == "221": - opponent = next_game["competitions"][0]["competitors"][1] - homeaway = next_game["competitions"][0]["competitors"][0]["homeAway"] - else: - opponent = next_game["competitions"][0]["competitors"][0] - homeaway = next_game["competitions"][0]["competitors"][1]["homeAway"] - return GameInfo( - timestamp=next_game["date"], - opponent={ - "id": opponent["team"]["id"], - "school": opponent["team"]["nickname"], - "name": opponent["team"]["displayName"], - }, - home_away=homeaway, - location={ - "full_name": next_game["competitions"][0]["venue"]["fullName"], - "address": next_game["competitions"][0]["venue"]["address"], - }, - status=status, - ) - except IndexError: - # IndexError occurs when a next game on the schedule is not present - return GameInfo(status="NO_GAME_SCHEDULED") +class SportsClient(BaseClient): + """Fetch Pitt football and men's basketball summaries.""" -def get_mens_basketball_standings() -> str: - """returns a string describing the placement of the men's basketball team. eg: '14th in ACC'""" - basketball_data = _get_mens_basketball_data() - return_value: str = basketball_data["team"]["standingSummary"] - return return_value + def get_mens_basketball_record(self) -> str: + return parse_record(self.get_team_data(MENS_BASKETBALL_URL)) + def get_next_mens_basketball_game(self) -> GameInfo: + return parse_next_game(self.get_team_data(MENS_BASKETBALL_URL)) -def get_football_record() -> str: - """returns the current record of the men's football team""" - football_data = _get_football_data() + def get_mens_basketball_standings(self) -> str: + return parse_standings(self.get_team_data(MENS_BASKETBALL_URL)) - try: - record_summary: str = football_data["team"]["record"]["items"][0]["summary"] - except KeyError: - record_summary = "There's no record right now." + def get_football_record(self) -> str: + return parse_record(self.get_team_data(FOOTBALL_URL)) - return record_summary + def get_next_football_game(self) -> GameInfo: + return parse_next_game(self.get_team_data(FOOTBALL_URL)) + def get_football_standings(self) -> str: + return parse_standings(self.get_team_data(FOOTBALL_URL)) -def get_next_football_game() -> GameInfo: - football_data = _get_football_data() - next_game = None - try: - next_game = football_data["team"]["nextEvent"][0] - opponent = None - homeaway = None - status = None - if next_game["competitions"][0]["status"]["type"]["name"] == "STATUS_FINAL": - status = "GAME_COMPLETE" - elif next_game["competitions"][0]["status"]["type"]["name"] == "STATUS_IN_PROGRESS": - status = "IN_PROGRESS" - if str(next_game["competitions"][0]["competitors"][0]["id"]) == "221": - opponent = next_game["competitions"][0]["competitors"][1] - homeaway = next_game["competitions"][0]["competitors"][0]["homeAway"] - else: - opponent = next_game["competitions"][0]["competitors"][0] - homeaway = next_game["competitions"][0]["competitors"][1]["homeAway"] - return GameInfo( - timestamp=next_game["date"], - opponent={ - "id": opponent["team"]["id"], - "school": opponent["team"]["nickname"], - "name": opponent["team"]["displayName"], - }, - home_away=homeaway, - location={ - "full_name": next_game["competitions"][0]["venue"]["fullName"], - "address": next_game["competitions"][0]["venue"]["address"], - }, - status=status, - ) - except IndexError: - # IndexError occurs when a next game on the schedule is not present - return GameInfo(status="NO_GAME_SCHEDULED") + def get_team_data(self, url: str) -> dict[str, Any]: + data = self.request("GET", url).json() + if not isinstance(data, dict): + raise ValueError("sports response must be an object") + return data -def get_football_standings() -> str: - """returns a string describing the placement of the football team. eg: '14th in ACC'""" - football_data = _get_football_data() - return_value: str = football_data["team"]["standingSummary"] - return return_value +def parse_record(data: dict[str, Any]) -> str: + try: + return data["team"]["record"]["items"][0]["summary"] + except (KeyError, IndexError, TypeError): + return NO_RECORD -def _get_mens_basketball_data() -> JSON: - json_data: JSON = requests.get(MENS_BASKETBALL_URL).json() - return json_data +def parse_standings(data: dict[str, Any]) -> str: + try: + return data["team"]["standingSummary"] + except (KeyError, TypeError) as error: + raise ValueError("sports response is missing standings") from error -def _get_football_data() -> JSON: - json_data: JSON = requests.get(FOOTBALL_URL).json() - return json_data +def parse_next_game(data: dict[str, Any]) -> GameInfo: + try: + events = data["team"]["nextEvent"] + if not events: + return GameInfo(status="NO_GAME_SCHEDULED") + event = events[0] + competition = event["competitions"][0] + pitt_team_id = str(data["team"]["id"]) + pitt, opponent = find_competitors(competition["competitors"], pitt_team_id) + opponent_team = opponent["team"] + venue = competition["venue"] + status_name = competition["status"]["type"]["name"] + statuses = { + "STATUS_FINAL": "GAME_COMPLETE", + "STATUS_IN_PROGRESS": "IN_PROGRESS", + } + return GameInfo( + timestamp=event["date"], + opponent=Team( + id=str(opponent_team["id"]), + school=opponent_team["nickname"], + name=opponent_team["displayName"], + ), + home_away=pitt["homeAway"], + location=Venue( + full_name=venue["fullName"], + address=Address(city=venue["address"]["city"], state=venue["address"]["state"]), + ), + status=statuses.get(status_name, "SCHEDULED"), + ) + except (KeyError, IndexError, TypeError) as error: + raise ValueError("sports response is missing game data") from error + + +def find_competitors( + competitors: list[dict[str, Any]], + pitt_team_id: str, +) -> tuple[dict[str, Any], dict[str, Any]]: + if len(competitors) != 2: + raise ValueError("a sports competition must contain two teams") + first, second = competitors + if str(first["id"]) == pitt_team_id: + return first, second + if str(second["id"]) == pitt_team_id: + return second, first + raise ValueError("Pitt is not present in the competition") diff --git a/pittapi/textbook.py b/pittapi/textbook.py index b2c2897..710a82c 100644 --- a/pittapi/textbook.py +++ b/pittapi/textbook.py @@ -1,235 +1,286 @@ -from __future__ import annotations +""" +The Pitt API, to access workable data of the University of Pittsburgh +Copyright (C) 2015 Ritwik Gupta -import warnings +This program is free software; you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation; either version 2 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License along +with this program; if not, write to the Free Software Foundation, Inc., +51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +Textbook information from Pitt's bookstore comparison service. +""" + +from __future__ import annotations -from bs4 import BeautifulSoup from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass +import json +import re +from typing import Any + +from bs4 import BeautifulSoup import requests -from requests import ConnectionError -from typing import Any, NamedTuple -BASE_URL = "https://pitt.verbacompare.com/" +from pittapi.base_client import BaseClient + +__all__ = ["CourseInfo", "Textbook", "TextbookClient", "TextbookTerm"] +BASE_URL = "https://pitt.verbacompare.com/" SUBJECTS_URL = BASE_URL + "compare/departments/?term={term_id}" -COURSES_URL = BASE_URL + "compare/courses/?id={dept_id}&term_id={term_id}" +COURSES_URL = BASE_URL + "compare/courses/?id={department_id}&term_id={term_id}" BOOKS_URL = BASE_URL + "compare/books?id={section_id}" - -CURRENT_TERM_ID = 78104 # Term ID for fall 2024, TODO: figure out how this ID is generated MAX_REQUEST_ATTEMPTS = 3 - -sess = requests.Session() -request_headers: dict[str, str] | None = None -subject_map: dict[str, str] | None = None +TERMS_PATTERN = re.compile(r"Collections\.Terms\((\[.*?\])\)", re.DOTALL) -@dataclass # No dataclass slots because they're not supported in Python 3.9 +@dataclass(frozen=True, slots=True) class CourseInfo: + """A normalized course lookup without any network side effects.""" + subject: str course_num: str instructor: str | None = None section_num: str | None = None def __post_init__(self) -> None: - if not subject_map: - _update_subject_map() - assert subject_map - - self.subject = self.subject.upper() - if self.subject not in subject_map: - raise LookupError(f"{self.subject} is not a valid subject") - if len(self.course_num) > 4 or not self.course_num.isdigit(): - raise ValueError("Invalid course number") - self.course_num = "0" * (4 - len(self.course_num)) + self.course_num - if self.instructor: - self.instructor = self.instructor.upper() + subject = self.subject.upper() + course_num = self.course_num + instructor = self.instructor.upper() if self.instructor else None + + if not course_num.isdigit() or len(course_num) > 4: + raise ValueError("invalid course number") if self.section_num and (len(self.section_num) != 4 or not self.section_num.isdigit()): - raise ValueError("Invalid section number") + raise ValueError("invalid section number") + object.__setattr__(self, "subject", subject) + object.__setattr__(self, "course_num", course_num.zfill(4)) + object.__setattr__(self, "instructor", instructor) -class Textbook(NamedTuple): + +@dataclass(frozen=True, slots=True) +class Textbook: title: str | None author: str | None edition: str | None isbn: str | None citation: str | None - @classmethod - def from_json(cls, json: dict[str, Any]) -> Textbook | None: - parsed_textbook = cls( - title=json.get("title"), - author=json.get("author"), - edition=json.get("edition"), - isbn=json.get("isbn"), - citation=json.get("citation"), - ) - # If all fields are None, then don't bother returning a Textbook object - return parsed_textbook if any(field for field in parsed_textbook) else None - - -def _update_headers() -> None: - for i in range(MAX_REQUEST_ATTEMPTS): - base_response = sess.get(BASE_URL) - if base_response.status_code == 200: - break - warnings.warn(f"Attempt {i + 1} to connect to textbook site failed, trying again") - if base_response.status_code != 200: # Request failed too many times - raise ConnectionError(f"Failed to connect to textbook site after {MAX_REQUEST_ATTEMPTS} attempts") - - soup = BeautifulSoup(base_response.text, "html.parser") - csrf_element = soup.find("meta", attrs={"name": "csrf-token"}) - if csrf_element: - csrf_token = csrf_element.get("content") - if isinstance(csrf_token, str): - global request_headers - request_headers = {"X-CSRF-Token": csrf_token} - return - raise ConnectionError("Unable to find valid request credentials, cannot connect to textbook site") - - -def _update_subject_map() -> None: - if not request_headers: - _update_headers() - - for i in range(MAX_REQUEST_ATTEMPTS): - subject_response = sess.get(SUBJECTS_URL.format(term_id=CURRENT_TERM_ID), headers=request_headers) - if subject_response.status_code == 200: - break - warnings.warn(f"Attempt {i + 1} to retrieve list of subjects failed, trying again") - _update_headers() # Try again with new CSRF token - if subject_response.status_code != 200: # Request failed too many times - raise ConnectionError(f"Failed to retrieve list of subjects after {MAX_REQUEST_ATTEMPTS} attempts") - - subject_json: list[dict[str, str]] = subject_response.json() - global subject_map - subject_map = {entry["name"]: entry["id"] for entry in subject_json} - -def _find_section_from_json(sections: list[dict[str, str]], instructor: str | None, section_num: str | None) -> str: +@dataclass(frozen=True, slots=True) +class TextbookTerm: + """A bookstore term available for textbook inquiries.""" + + id: str + name: str + inquiry_enabled: bool + ordering_enabled: bool + + +class TextbookClient(BaseClient): + """Fetch textbooks while keeping credentials and subjects per client.""" + + def __init__( + self, + term: TextbookTerm | None = None, + session: requests.Session | None = None, + timeout: float = 10.0, + ) -> None: + super().__init__(session=session, timeout=timeout) + self.term = term + self.terms: tuple[TextbookTerm, ...] | None = None + self.headers: dict[str, str] | None = None + self.subject_ids: dict[str, str] | None = None + + def initialize_headers(self) -> None: + last_error = None + for _ in range(MAX_REQUEST_ATTEMPTS): + try: + response = self.request("GET", BASE_URL) + except requests.HTTPError as error: + last_error = error + continue + + soup = BeautifulSoup(response.text, "html.parser") + csrf_element = soup.find("meta", attrs={"name": "csrf-token"}) + csrf_token = csrf_element.get("content") if csrf_element else None + if not isinstance(csrf_token, str): + raise requests.ConnectionError("textbook site did not provide valid request credentials") + + terms_match = TERMS_PATTERN.search(response.text) + if terms_match is None: + raise ValueError("textbook site did not provide available terms") + try: + terms_data = json.loads(terms_match.group(1)) + self.terms = tuple( + TextbookTerm( + id=str(item["id"]), + name=item["name"], + inquiry_enabled=item["inquiry"], + ordering_enabled=item["ordering"], + ) + for item in terms_data + ) + except (json.JSONDecodeError, KeyError, TypeError) as error: + raise ValueError("textbook term response is missing required data") from error + + self.headers = {"X-CSRF-Token": csrf_token} + return + raise requests.ConnectionError( + f"failed to connect to textbook site after {MAX_REQUEST_ATTEMPTS} attempts" + ) from last_error + + def get_terms(self) -> tuple[TextbookTerm, ...]: + """Return the terms currently published by the bookstore.""" + if self.terms is None: + self.initialize_headers() + return self.terms or () + + def select_term(self, term: TextbookTerm) -> None: + """Select a discovered term and clear term-specific subject state.""" + if term != self.term: + self.term = term + self.subject_ids = None + + def initialize_subjects(self) -> None: + if self.term is None: + raise ValueError("select a textbook term before requesting subjects") + if self.headers is None: + self.initialize_headers() + + url = SUBJECTS_URL.format(term_id=self.term.id) + last_error = None + for attempt in range(MAX_REQUEST_ATTEMPTS): + try: + response = self.request("GET", url, headers=self.headers) + except requests.HTTPError as error: + last_error = error + if attempt < MAX_REQUEST_ATTEMPTS - 1: + self.initialize_headers() + continue + + try: + self.subject_ids = {item["name"]: item["id"] for item in response.json()} + except (KeyError, TypeError) as error: + raise ValueError("textbook subject response is missing required data") from error + return + raise requests.ConnectionError(f"failed to retrieve subjects after {MAX_REQUEST_ATTEMPTS} attempts") from last_error + + def get_textbooks_for_course(self, course: CourseInfo) -> tuple[Textbook, ...]: + return self.get_textbooks_for_courses((course,)) + + def get_textbooks_for_courses(self, courses: tuple[CourseInfo, ...] | list[CourseInfo]) -> tuple[Textbook, ...]: + if self.term is None: + raise ValueError("select a textbook term before requesting textbooks") + if self.subject_ids is None: + self.initialize_subjects() + + courses_by_subject = {} + for course in courses: + if course.subject not in self.subject_ids: + raise LookupError(f"invalid textbook subject: {course.subject}") + if course.subject not in courses_by_subject: + courses_by_subject[course.subject] = self.get_courses(course.subject) + + section_ids = [] + for course in courses: + section_ids.append(find_section_id(courses_by_subject[course.subject], course)) + + with ThreadPoolExecutor() as executor: + textbook_groups = executor.map(self.get_textbooks_for_section, section_ids) + + textbooks = [] + for group in textbook_groups: + textbooks.extend(group) + return tuple(textbooks) + + def get_courses(self, subject: str) -> list[dict[str, Any]]: + if self.term is None: + raise ValueError("select a textbook term before requesting courses") + if self.headers is None: + self.initialize_headers() + if self.subject_ids is None or subject not in self.subject_ids: + raise LookupError(f"invalid textbook subject: {subject}") + + url = COURSES_URL.format(department_id=self.subject_ids[subject], term_id=self.term.id) + last_error = None + for attempt in range(MAX_REQUEST_ATTEMPTS): + try: + response = self.request("GET", url, headers=self.headers) + except requests.HTTPError as error: + last_error = error + if attempt < MAX_REQUEST_ATTEMPTS - 1: + self.initialize_headers() + continue + + data = response.json() + if not isinstance(data, list): + raise ValueError("textbook course response must contain a list") + return data + raise requests.ConnectionError(f"failed to retrieve {subject} courses") from last_error + + def get_textbooks_for_section(self, section_id: str) -> tuple[Textbook, ...]: + if self.term is None: + raise ValueError("select a textbook term before requesting textbooks") + if self.headers is None: + self.initialize_headers() + data = self.request("GET", BOOKS_URL.format(section_id=section_id), headers=self.headers).json() + if not isinstance(data, list): + raise ValueError("textbook response must contain a list") + + textbooks = [] + for item in data: + textbook = parse_textbook(item) + if textbook is not None: + textbooks.append(textbook) + return tuple(textbooks) + + +def find_section_id(courses: list[dict[str, Any]], course: CourseInfo) -> str: + for course_data in courses: + if course_data["id"] == course.subject + course.course_num: + return find_section(course_data["sections"], course.instructor, course.section_num) + raise LookupError(f"invalid textbook course: {course.subject} {course.course_num}") + + +def find_section( + sections: list[dict[str, str]], + instructor: str | None, + section_num: str | None, +) -> str: if section_num: for section in sections: if section["name"] == section_num: return section["id"] - raise LookupError(f"No section found with given {section_num=}") + raise LookupError(f"section not found: {section_num}") + if instructor: for section in sections: if section["instructor"] == instructor: return section["id"] - raise LookupError(f"No section found with given {instructor=}") + raise LookupError(f"instructor not found: {instructor}") - # Not enough info provided, so try to deduce the correct section: - # - If there's only 1 section of the course, then the sole section must be the correct one - # - If all sections of the course are taught by the same instructor, then we can assume that all sections will have the - # same textbook, meaning that the exact section doesn't matter instructors = {section["instructor"] for section in sections} if len(sections) == 1 or len(instructors) == 1: return sections[0]["id"] - raise LookupError( - "Cannot determine section ID from given arguments, please provide the instructor's name and/or the section number" - ) - - -def _get_textbooks_for_ids(ids: list[str]) -> list[Textbook]: - """Fetches a course's textbook information and returns a list - of textbooks for the given course. - """ - if not request_headers: - _update_headers() - - def fetch(section_id: str) -> requests.Response: - response = requests.get(BOOKS_URL.format(section_id=section_id), headers=request_headers) - response.raise_for_status() - return response - - with ThreadPoolExecutor() as executor: - responses = executor.map(fetch, ids) + raise LookupError("provide an instructor or section number to identify the textbook section") - books = [] - for response in responses: - for book_json in response.json(): - book = Textbook.from_json(book_json) - if book: - books.append(book) - else: - warnings.warn(f"No textbook info found for {response}") - return [book for book in books if book] # Drop all None values - -def _find_section_id_from_json( - course_json: list[dict[str, Any]], subject: str, course_num: str, instructor: str | None, section_num: str | None -) -> str: - for course in course_json: - if course["id"] == subject + course_num: - return _find_section_from_json(course["sections"], instructor, section_num) - raise LookupError(f"{subject} {course_num} is not a valid course") - - -def _get_textbooks_from_json( - course_json: list[dict[str, Any]], subject: str, course_num: str, instructor: str | None, section_num: str | None -) -> list[Textbook]: - section_id = _find_section_id_from_json(course_json, subject, course_num, instructor, section_num) - return _get_textbooks_for_ids([section_id]) - - -def get_textbooks_for_course(course: CourseInfo) -> list[Textbook]: - if not request_headers: - _update_headers() - if not subject_map: - _update_subject_map() - assert subject_map - - for i in range(MAX_REQUEST_ATTEMPTS): - course_response = sess.get( - COURSES_URL.format(dept_id=subject_map[course.subject], term_id=CURRENT_TERM_ID), headers=request_headers - ) - if course_response.status_code == 200: - break - warnings.warn(f"Attempt {i} to retrieve list of {course.subject} courses failed, trying again") - _update_headers() # Try again with new CSRF token - if course_response.status_code != 200: # Request failed too many times - raise ConnectionError(f"Failed to retrieve list of {course.subject} courses from textbook site") - - return _get_textbooks_from_json( - course_json=course_response.json(), - subject=course.subject, - course_num=course.course_num, - instructor=course.instructor, - section_num=course.section_num, +def parse_textbook(data: dict[str, Any]) -> Textbook | None: + textbook = Textbook( + title=data.get("title"), + author=data.get("author"), + edition=data.get("edition"), + isbn=data.get("isbn"), + citation=data.get("citation"), ) - - -def get_textbooks_for_courses(courses_info: list[CourseInfo]) -> list[Textbook]: - if not request_headers: - _update_headers() - if not subject_map: - _update_subject_map() - assert subject_map - - # Precompute list of unique subjects to avoid unnecessary API requests - subjects = {course_info.subject for course_info in courses_info} - courses_for_subjects: dict[str, list[dict[str, Any]]] = {} - for subject in subjects: - for i in range(MAX_REQUEST_ATTEMPTS): - course_response = sess.get( - COURSES_URL.format(dept_id=subject_map[subject], term_id=CURRENT_TERM_ID), headers=request_headers - ) - if course_response.status_code == 200: - break - warnings.warn(f"Attempt {i} to retrieve list {subject} courses failed, trying again") - _update_headers() # Try again with new CSRF token - if course_response.status_code != 200: # Request failed too many times - raise ConnectionError(f"Failed to retrieve list of {subject} courses from textbook site") - - courses_for_subjects[subject] = course_response.json() - - section_ids = [ - _find_section_id_from_json( - course_json=courses_for_subjects[course_info.subject], - subject=course_info.subject, - course_num=course_info.course_num, - instructor=course_info.instructor, - section_num=course_info.section_num, - ) - for course_info in courses_info - ] - return _get_textbooks_for_ids(section_ids) + values = (textbook.title, textbook.author, textbook.edition, textbook.isbn, textbook.citation) + return textbook if any(values) else None diff --git a/tests/course_test.py b/tests/course_test.py index 5615992..bf7991d 100644 --- a/tests/course_test.py +++ b/tests/course_test.py @@ -1,303 +1,267 @@ -""" -The Pitt API, to access workable data of the University of Pittsburgh -Copyright (C) 2015 Ritwik Gupta - -This program is free software; you can redistribute it and/or modify -it under the terms of the GNU General Public License as published by -the Free Software Foundation; either version 2 of the License, or -(at your option) any later version. - -This program is distributed in the hope that it will be useful, -but WITHOUT ANY WARRANTY; without even the implied warranty of -MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -GNU General Public License for more details. - -You should have received a copy of the GNU General Public License along -with this program; if not, write to the Free Software Foundation, Inc., -51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. -""" - from copy import deepcopy -import unittest -from unittest.mock import patch +import json +import pytest import responses -from pittapi import course - -from pittapi.course import Attribute, Course, CourseDetails, Instructor, Meeting, Section, SectionDetails, Subject +from pittapi.course import ( + COURSE_DETAIL_API, + COURSE_SECTIONS_API, + SECTION_DETAILS_API, + SUBJECT_COURSES_API, + SUBJECTS_API, + Course, + CourseClient, + CourseDetails, + Section, + Subject, + parse_catalog_section, + parse_course_details, + parse_detailed_meeting, + parse_instructors, + parse_section_details, + validate_course_number, + validate_term, +) from tests.mocks.course_mocks import ( - mocked_subject_data, - mocked_courses_data, - mocked_courses_data_invalid, mocked_course_info_data, mocked_course_sections_data, + mocked_courses_data, mocked_section_details_data, + mocked_subject_data, ) +TERM = "2231" +SUBJECT = "CS" +COURSE_NUMBER = "0007" +COURSE_ID = "105611" + + +def add_subject_response(data=mocked_subject_data): + responses.add(responses.GET, SUBJECTS_API, json=data) + + +@responses.activate +def test_get_subject_courses_returns_models(): + add_subject_response() + responses.add(responses.GET, SUBJECT_COURSES_API.format(subject=SUBJECT), json=mocked_courses_data) + + result = CourseClient().get_subject_courses("cs") + + assert isinstance(result, Subject) + assert isinstance(result.courses[0], Course) + assert result.courses[0].course_number == COURSE_NUMBER + + +@responses.activate +def test_people_soft_redirect_preserves_session_cookie(): + url = SUBJECT_COURSES_API.format(subject=SUBJECT) + responses.add( + responses.GET, + url, + status=302, + headers={"Location": f"{url}&", "Set-Cookie": "pscheck=accepted; Path=/"}, + ) + + def return_courses(request): + assert "pscheck=accepted" in request.headers["Cookie"] + return 200, {"Content-Type": "application/json"}, json.dumps(mocked_courses_data) + + responses.add_callback(responses.GET, f"{url}&", callback=return_courses) + + assert CourseClient().get_subject_course_data(SUBJECT) == mocked_courses_data + assert responses.calls[0].request.url == url + assert responses.calls[1].request.url == f"{url}&" + + +@responses.activate +def test_get_course_details_returns_nested_models(): + add_subject_response() + responses.add(responses.GET, SUBJECT_COURSES_API.format(subject=SUBJECT), json=mocked_courses_data) + responses.add(responses.GET, COURSE_DETAIL_API.format(id=COURSE_ID), json=mocked_course_info_data) + responses.add( + responses.GET, + COURSE_SECTIONS_API.format(id=COURSE_ID, term=TERM), + json=mocked_course_sections_data, + ) + + result = CourseClient().get_course_details(TERM, SUBJECT, 7) + + assert "effdt" not in responses.calls[2].request.url + assert isinstance(result, CourseDetails) + assert result.course.course_id == COURSE_ID + assert result.components[0].required + assert result.attributes[0].value == "ALG" + assert result.sections[0].instructors[0].email == "rmf105@pitt.edu" + assert result.sections[0].meetings[0].instructors[0].name == "Robert Fishel" + + +@responses.activate +def test_get_section_details_returns_nested_models(): + responses.add( + responses.GET, + SECTION_DETAILS_API.format(term=TERM, id=27815), + json=mocked_section_details_data, + ) + result = CourseClient().get_section_details(TERM, 27815) + assert isinstance(result, Section) + assert result.details.enrollment_available == "4" + assert result.meetings[0].instructors[0].name == "Robert Fishel" + + +@pytest.mark.parametrize("term", [2191, "2194", "2197"]) +def test_validate_term(term): + assert validate_term(term) == str(term) + + +@pytest.mark.parametrize("term", ["214", "1111", "12345"]) +def test_reject_invalid_term(term): + with pytest.raises(ValueError, match="four-digit Pitt term"): + validate_term(term) + + +@pytest.mark.parametrize(("value", "expected"), [(7, "0007"), ("449", "0449"), (1501, "1501")]) +def test_validate_course_number(value, expected): + assert validate_course_number(value) == expected + + +@pytest.mark.parametrize("value", ["", "abc", 0, -1]) +def test_reject_nonpositive_or_nonnumeric_course(value): + with pytest.raises(ValueError, match="positive number"): + validate_course_number(value) + + +def test_reject_long_course_number(): + with pytest.raises(ValueError, match="four digits"): + validate_course_number("12345") + + +@responses.activate +def test_subject_validation_errors(): + add_subject_response() + with pytest.raises(ValueError, match="invalid subject code"): + CourseClient().validate_subject("MISSING") + + add_subject_response({}) + with pytest.raises(ValueError, match="subject response"): + CourseClient().validate_subject("CS") + + +@responses.activate +def test_subject_course_response_validation(): + add_subject_response() + responses.add(responses.GET, SUBJECT_COURSES_API.format(subject=SUBJECT), json={}) + with pytest.raises(ValueError, match="subject course response"): + CourseClient().get_subject_courses(SUBJECT) + + +@responses.activate +def test_find_course_id_errors_and_duplicate_selection(): + client = CourseClient() + duplicate = { + "courses": [ + {"catalog_nbr": "0001", "crse_id": "other"}, + {"catalog_nbr": COURSE_NUMBER, "crse_id": "first"}, + {"catalog_nbr": COURSE_NUMBER, "crse_id": "second"}, + ] + } + responses.add(responses.GET, SUBJECT_COURSES_API.format(subject=SUBJECT), json=duplicate) + assert client.find_course_id(SUBJECT, COURSE_NUMBER) == "first" + + responses.add(responses.GET, SUBJECT_COURSES_API.format(subject=SUBJECT), json={"courses": []}) + with pytest.raises(LookupError, match="course not found"): + client.find_course_id(SUBJECT, "9999") + + responses.add(responses.GET, SUBJECT_COURSES_API.format(subject=SUBJECT), json={}) + with pytest.raises(ValueError, match="subject course response"): + client.find_course_id(SUBJECT, COURSE_NUMBER) + + +@responses.activate +def test_low_level_lookup_errors(): + client = CourseClient() + responses.add(responses.GET, COURSE_DETAIL_API.format(id="bad"), json={"course_details": {}}) + with pytest.raises(LookupError, match="course ID not found"): + client.get_course_data("bad") + + responses.add(responses.GET, COURSE_SECTIONS_API.format(id="bad", term=TERM), json={"sections": []}) + with pytest.raises(LookupError, match="no sections"): + client.get_course_section_data("bad", TERM) + + responses.add(responses.GET, SECTION_DETAILS_API.format(term=TERM, id="bad"), json={"error": "missing"}) + with pytest.raises(LookupError, match="section not found"): + client.get_section_data(TERM, "bad") + + +@responses.activate +def test_public_parsing_errors_are_value_errors(): + add_subject_response() + responses.add(responses.GET, SUBJECT_COURSES_API.format(subject=SUBJECT), json=mocked_courses_data) + responses.add(responses.GET, COURSE_DETAIL_API.format(id=COURSE_ID), json={"course_details": {"bad": True}}) + responses.add( + responses.GET, + COURSE_SECTIONS_API.format(id=COURSE_ID, term=TERM), + json={"sections": [{"bad": True}]}, + ) + with pytest.raises(ValueError, match="course response"): + CourseClient().get_course_details(TERM, SUBJECT, COURSE_NUMBER) -class CourseTest(unittest.TestCase): - def setUp(self): - subjects_patcher = patch.object(course, "_get_subjects", return_value=mocked_subject_data) - section_details_patcher = patch.object(course, "_get_section_details", return_value=mocked_section_details_data) - self.mock_get_subjects = subjects_patcher.start() - self.mock_get_section_details = section_details_patcher.start() - self.addCleanup(subjects_patcher.stop) - self.addCleanup(section_details_patcher.stop) - - def test_validate_term(self): - # If convert to string - self.assertTrue(isinstance(course._validate_term(2191), str)) - - self.assertEqual(course._validate_term(2191), "2191") - self.assertEqual(course._validate_term("2191"), "2191") - - self.assertRaises(ValueError, course._validate_term, "214") - self.assertRaises(ValueError, course._validate_term, "1111") - self.assertRaises(ValueError, course._validate_term, "12345") - - def test_validate_subject(self): - self.assertEqual(course._validate_subject("CS"), "CS") - - self.assertRaises(ValueError, course._validate_subject, "foobar") - - def test_validate_course(self): - self.assertEqual(course._validate_course(7), "0007") - self.assertEqual(course._validate_course(449), "0449") - self.assertEqual(course._validate_course(1501), "1501") - - self.assertEqual(course._validate_course("7"), "0007") - self.assertEqual(course._validate_course("0007"), "0007") - self.assertEqual(course._validate_course("449"), "0449") - self.assertEqual(course._validate_course("1501"), "1501") - - self.assertRaises(ValueError, course._validate_course, -1) - self.assertRaises(ValueError, course._validate_course, 0) - self.assertRaises(ValueError, course._validate_course, "") - self.assertRaises(ValueError, course._validate_course, "A00") - self.assertRaises(ValueError, course._validate_course, "Hello") - self.assertRaises(ValueError, course._validate_course, "10000") - - def test_get_subject_courses(self): - with patch.object(course, "_get_subject_courses", return_value=mocked_courses_data) as get_subject_courses: - subject_courses = course.get_subject_courses("CS") - - get_subject_courses.assert_called_once_with("CS") - self.assertTrue(isinstance(subject_courses, Subject)) - - self.assertEqual(len(subject_courses.courses), 1) - self.assertTrue("0007" in subject_courses.courses) - test_course = subject_courses.courses["0007"] - self.assertTrue(isinstance(test_course, Course)) - - def test_get_subject_courses_invalid(self): - with patch.object(course, "_get_subject_courses", return_value=mocked_courses_data_invalid) as get_subject_courses: - self.assertRaises(ValueError, course.get_subject_courses, "nonsense") - - get_subject_courses.assert_not_called() - - def test_get_course_details(self): - with ( - patch.object(course, "_get_course_id", return_value="105611"), - patch.object(course, "_get_course_info", return_value=mocked_course_info_data), - patch.object(course, "_get_course_sections", return_value=mocked_course_sections_data), - ): - course_sections = course.get_course_details("2231", "CS", "0007") - - self.assertTrue(isinstance(course_sections, CourseDetails)) - - self.assertTrue(isinstance(course_sections.course, Course)) - course_obj = course_sections.course - self.assertEqual(course_obj.subject_code, "CS") - self.assertEqual(course_obj.course_number, "0007") - self.assertEqual(course_obj.course_id, "105611") - self.assertEqual(course_obj.course_title, "INTRO TO COMPUTER PROGRAMMING") - - self.assertEqual(len(course_sections.sections), 1) - test_attribute = course_sections.attributes[0] - self.assertTrue(isinstance(test_attribute, Attribute)) - self.assertEqual(test_attribute.attribute, "DSGE") - self.assertEqual(test_attribute.attribute_description, "*DSAS General Ed. Requirements") - self.assertEqual(test_attribute.value, "ALG") - self.assertEqual(test_attribute.value_description, "Algebra") - test_section = course_sections.sections[0] - self.assertTrue(isinstance(test_section, Section)) - self.assertEqual(test_section.term, "2231") - self.assertEqual(test_section.session, "Academic Term") - self.assertEqual(test_section.section_number, "1000") - self.assertEqual(test_section.class_number, "27815") - self.assertEqual(test_section.section_type, "REC") - self.assertEqual(test_section.status, "Open") - - self.assertEqual(len(test_section.instructors), 1) - self.assertEqual(len(test_section.meetings), 1) - test_instructor = test_section.instructors[0] - test_meeting = test_section.meetings[0] - self.assertTrue(isinstance(test_instructor, Instructor)) - self.assertEqual(test_instructor.name, "Robert Fishel") - self.assertEqual(test_instructor.email, "rmf105@pitt.edu") - self.assertTrue(isinstance(test_meeting, Meeting)) - self.assertEqual(test_meeting.days, "Fr") - self.assertEqual(test_meeting.start_time, "10.00.00.000000-05:00") - self.assertEqual(test_meeting.end_time, "10.50.00.000000-05:00") - self.assertEqual(test_meeting.start_date, "08/29/2022") - self.assertEqual(test_meeting.end_date, "12/09/2022") - - test_instructor = test_section.instructors[0] - self.assertTrue(isinstance(test_instructor, Instructor)) - self.assertEqual(test_instructor.name, "Robert Fishel") - - def test_get_section_details(self): - section_details = course.get_section_details("2231", "27815") - - self.assertTrue(isinstance(section_details, Section)) - self.assertEqual(section_details.term, "2231") - self.assertEqual(section_details.session, "Academic Term") - self.assertEqual(section_details.class_number, "27815") - self.assertEqual(section_details.section_type, "REC") - self.assertEqual(section_details.status, "Open") - self.assertIsNone(section_details.instructors) - test_meeting = section_details.meetings[0] - - self.assertTrue(isinstance(test_meeting, Meeting)) - self.assertEqual(test_meeting.days, "Fr") - self.assertEqual(test_meeting.start_time, "10:00AM") - self.assertEqual(test_meeting.end_time, "10:50AM") - self.assertEqual(test_meeting.start_date, "08/29/2022") - self.assertEqual(test_meeting.end_date, "12/09/2022") - test_instructor = test_meeting.instructors[0] - - self.assertTrue(isinstance(test_instructor, Instructor)) - self.assertEqual(test_instructor.name, "Robert Fishel") - self.assertEqual(test_instructor.email, "rmf105@pitt.edu") - - test_details = section_details.details - self.assertTrue(isinstance(test_details, SectionDetails)) - self.assertEqual(test_details.units, "0 units") - self.assertEqual(test_details.class_capacity, "28") - self.assertEqual(test_details.enrollment_total, "24") - self.assertEqual(test_details.enrollment_available, "4") - self.assertEqual(test_details.wait_list_capacity, "50") - self.assertEqual(test_details.wait_list_total, "7") - self.assertEqual(test_details.valid_to_enroll, "T") - self.assertIsNone(test_details.combined_section_numbers) - - def test_get_course_details_without_optional_data(self): - course_info = deepcopy(mocked_course_info_data) - course_info["course_details"].pop("offerings") - course_info["course_details"]["components"] = [] - course_info["course_details"]["attributes"] = [] - course_sections = deepcopy(mocked_course_sections_data) - course_sections["sections"][0]["instructors"] = [] - course_sections["sections"][0]["meetings"] = [] - - with ( - patch.object(course, "_get_course_id", return_value="105611"), - patch.object(course, "_get_course_info", return_value=course_info), - patch.object(course, "_get_course_sections", return_value=course_sections), - ): - details = course.get_course_details("2231", "CS", "0007") - - self.assertIsNone(details.requisites) - self.assertIsNone(details.components) - self.assertIsNone(details.attributes) - self.assertIsNone(details.sections[0].instructors) - self.assertIsNone(details.sections[0].meetings) - - def test_get_section_details_without_meetings_and_with_combined_sections(self): - section_data = deepcopy(mocked_section_details_data) - section_data["section_info"]["meetings"] = [] - section_data["section_info"]["is_combined"] = True - section_data["section_info"]["combined_sections"] = [{"class_nbr": "27815"}, {"class_nbr": "27816"}] - - with patch.object(course, "_get_section_details", return_value=section_data): - section = course.get_section_details("2231", "27815") - - self.assertIsNone(section.meetings) - self.assertEqual(section.details.combined_section_numbers, ["27815", "27816"]) - - def test_get_section_details_without_meeting_instructors(self): - section_data = deepcopy(mocked_section_details_data) - section_data["section_info"]["meetings"][0]["instructors"] = [] - - with patch.object(course, "_get_section_details", return_value=section_data): - section = course.get_section_details("2231", "27815") - - self.assertIsNone(section.meetings[0].instructors) - - def test_internal_course_id_helpers(self): - duplicate_courses = { - "courses": [ - {"catalog_nbr": "0007", "crse_id": "first"}, - {"catalog_nbr": "0007", "crse_id": "duplicate"}, - ] - } - with patch.object(course, "_get_subject_courses", return_value=duplicate_courses): - self.assertEqual(course._get_internal_id_dict("CS"), {"0007": "first"}) - self.assertEqual(course._get_course_id("CS", "0007"), "first") - with self.assertRaisesRegex(ValueError, "No course with that number"): - course._get_course_id("CS", "9999") - - -class CourseHttpHelperTest(unittest.TestCase): - @responses.activate - def test_subject_and_course_http_helpers(self): - responses.add(responses.GET, course.SUBJECTS_API, json=mocked_subject_data, status=200) - responses.add( - responses.GET, - course.SUBJECT_COURSES_API.format(subject="CS"), - json=mocked_courses_data, - status=200, - ) - responses.add( - responses.GET, - course.COURSE_DETAIL_API.format(id="105611"), - json=mocked_course_info_data, - status=200, - ) - responses.add( - responses.GET, - course.COURSE_SECTIONS_API.format(id="105611", term="2231"), - json=mocked_course_sections_data, - status=200, - ) - responses.add( - responses.GET, - course.SECTION_DETAILS_API.format(term="2231", id="27815"), - json=mocked_section_details_data, - status=200, - ) - - self.assertEqual(course._get_subjects(), mocked_subject_data) - self.assertEqual(course._get_subject_courses("CS"), mocked_courses_data) - self.assertEqual(course._get_course_info("105611"), mocked_course_info_data) - self.assertEqual(course._get_course_sections("105611", "2231"), mocked_course_sections_data) - self.assertEqual(course._get_section_details("2231", "27815"), mocked_section_details_data) - - @responses.activate - def test_course_http_helper_errors(self): - responses.add( - responses.GET, - course.COURSE_DETAIL_API.format(id="invalid"), - json={"course_details": {}}, - status=200, - ) - responses.add( - responses.GET, - course.COURSE_SECTIONS_API.format(id="invalid", term="2231"), - json={"sections": []}, - status=200, - ) - responses.add( - responses.GET, - course.SECTION_DETAILS_API.format(term="2231", id="invalid"), - json={"error": "invalid"}, - status=200, - ) - - with self.assertRaisesRegex(ValueError, "Invalid course ID"): - course._get_course_info("invalid") - with self.assertRaisesRegex(ValueError, "Invalid course ID"): - course._get_course_sections("invalid", "2231") - with self.assertRaisesRegex(ValueError, "Invalid section ID"): - course._get_section_details("2231", "invalid") + responses.add( + responses.GET, + SECTION_DETAILS_API.format(term=TERM, id="bad"), + json={"section_info": {}}, + ) + with pytest.raises(ValueError, match="section response"): + CourseClient().get_section_details(TERM, "bad") + + +def test_optional_course_data_and_unnamed_meetings(): + catalog = { + "descrlong": None, + "units_minimum": 1, + "units_maximum": 2, + } + raw_section = deepcopy(mocked_course_sections_data["sections"][0]) + raw_section["instructors"] = ["To be Announced"] + raw_section["meetings"][0]["instructor"] = "" + + result = parse_course_details(catalog, [raw_section], TERM, SUBJECT, COURSE_NUMBER, COURSE_ID) + + assert result.requisites is None + assert result.components == () + assert result.attributes == () + assert result.sections[0].instructors == () + assert result.sections[0].meetings[0].instructors == () + + +def test_catalog_section_without_meetings(): + raw_section = deepcopy(mocked_course_sections_data["sections"][0]) + raw_section.pop("meetings") + assert parse_catalog_section(raw_section, TERM).meetings == () + + +def test_combined_section_and_unnamed_detailed_instructors(): + section_info = deepcopy(mocked_section_details_data["section_info"]) + section_info["is_combined"] = True + section_info["combined_sections"] = [{"class_nbr": 1}, {"class_nbr": "2"}] + section_info["meetings"][0]["instructors"] = [ + {"name": "To be Announced", "email": None}, + {"name": "-", "email": None}, + ] + + result = parse_section_details(section_info, TERM, "27815") + + assert result.details.combined_section_numbers == ("1", "2") + assert result.meetings[0].instructors == () + + +def test_parse_instructors_empty_and_email_optional(): + assert parse_instructors(()) == () + assert parse_instructors([{"name": "Teacher"}])[0].email is None + + +def test_detailed_meeting_requires_date_range(): + meeting = deepcopy(mocked_section_details_data["section_info"]["meetings"][0]) + meeting["date_range"] = "invalid" + with pytest.raises(ValueError): + parse_detailed_meeting(meeting) diff --git a/tests/sports_test.py b/tests/sports_test.py index 0be506b..b4fc5d6 100644 --- a/tests/sports_test.py +++ b/tests/sports_test.py @@ -1,269 +1,121 @@ -""" -The Pitt API, to access workable data of the University of Pittsburgh -Copyright (C) 2015 Ritwik Gupta -This program is free software; you can redistribute it and/or modify -it under the terms of the GNU General Public License as published by -the Free Software Foundation; either version 2 of the License, or -(at your option) any later version. -This program is distributed in the hope that it will be useful, -but WITHOUT ANY WARRANTY; without even the implied warranty of -MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -GNU General Public License for more details. -You should have received a copy of the GNU General Public License along -with this program; if not, write to the Free Software Foundation, Inc., -51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. -""" - -from copy import deepcopy -import unittest -from unittest.mock import patch - +import pytest import responses -from pittapi import sports - -class LibraryTest(unittest.TestCase): - def setUp(self): - self.mocked_basketball_data = { - "team": { - "id": "221", - "record": { - "items": [ +from pittapi.sports import ( + FOOTBALL_URL, + MENS_BASKETBALL_URL, + GameInfo, + SportsClient, + Team, + find_competitors, + parse_next_game, + parse_record, + parse_standings, +) + + +def team_data(status="STATUS_FINAL"): + return { + "team": { + "id": "221", + "record": {"items": [{"summary": "11-21"}]}, + "standingSummary": "12th in ACC", + "nextEvent": [ + { + "date": "2022-03-08T19:00Z", + "competitions": [ { - "description": "Overall Record", - "type": "total", - "summary": "11-21", - }, - { - "description": "Home Record", - "type": "home", - "summary": "9-11", - }, - ] - }, - "nextEvent": [ - { - "date": "2022-03-08T19:00Z", - "competitions": [ - { - "venue": { - "fullName": "Barclays Center", - "address": {"city": "Brooklyn", "state": "NY"}, - }, - "competitors": [ - { - "id": "221", - "homeAway": "home", - "team": { - "id": "221", - "location": "Pittsburgh", - "nickname": "Pittsburgh", - "abbreviation": "PITT", - "displayName": "Pittsburgh Panthers", - }, + "venue": { + "fullName": "Barclays Center", + "address": {"city": "Brooklyn", "state": "NY"}, + }, + "competitors": [ + { + "id": 221, + "homeAway": "home", + "team": { + "id": 221, + "nickname": "Pittsburgh", + "displayName": "Pittsburgh Panthers", }, - { + }, + { + "id": "103", + "homeAway": "away", + "team": { "id": "103", - "homeAway": "away", - "team": { - "id": "103", - "location": "Boston College", - "nickname": "Boston College", - "abbreviation": "BC", - "displayName": "Boston College Eagles", - }, + "nickname": "Boston College", + "displayName": "Boston College Eagles", }, - ], - "status": {"type": {"name": "STATUS_FINAL"}}, - } - ], - } - ], - "standingSummary": "12th in ACC", - } - } - self.mocked_football_data = { - "team": { - "id": "221", - "name": "Pittsburgh", - "record": { - "items": [ - { - "description": "Overall Record", - "type": "total", - "summary": "10-2", - }, - { - "description": "Home Record", - "type": "home", - "summary": "5-2", - }, - ] - }, - "nextEvent": [ - { - "id": "401405793", - "date": "2022-03-08T19:00Z", - "name": "Pittsburgh Panthers at Wake Forest Deamon Deacons", - "competitions": [ - { - "venue": { - "fullName": "Bank of America Stadium", - "address": {"city": "Charlotte", "state": "NC"}, }, - "competitors": [ - { - "id": "221", - "homeAway": "away", - "team": { - "id": "221", - "location": "Pittsburgh", - "nickname": "Pittsburgh", - "abbreviation": "PITT", - "displayName": "Pittsburgh Panthers", - }, - }, - { - "id": "104", - "homeAway": "away", - "team": { - "id": "103", - "location": "Wake Forest", - "nickname": "Wake Forest", - "abbreviation": "WAKE", - "displayName": "Wake Forest Deamon Deacons", - }, - }, - ], - "status": {"type": {"name": "STATUS_IN_PROGRESS"}}, - } - ], - } - ], - "standingSummary": "1st in ACC - Coastal", - } + ], + "status": {"type": {"name": status}}, + } + ], + } + ], } - basketball_patcher = patch.object( - sports, - "_get_mens_basketball_data", - return_value=self.mocked_basketball_data, - ) - football_patcher = patch.object( - sports, - "_get_football_data", - return_value=self.mocked_football_data, - ) - self.mock_get_basketball_data = basketball_patcher.start() - self.mock_get_football_data = football_patcher.start() - self.addCleanup(basketball_patcher.stop) - self.addCleanup(football_patcher.stop) - - def test_get_mens_basketball_record(self): - self.assertEqual("11-21", sports.get_mens_basketball_record()) - - def test_get_mens_basketball_record_offseason(self): - offseason_data = {"team": {"id": "221", "record": {}}} - self.mock_get_basketball_data.return_value = offseason_data - - self.assertEqual("There's no record right now.", sports.get_mens_basketball_record()) - - def test_get_football_record(self): - self.assertEqual("10-2", sports.get_football_record()) - - def test_get_football_record_offseason(self): - offseason_data = {"team": {"id": "221", "record": {}}} - self.mock_get_football_data.return_value = offseason_data - - self.assertEqual("There's no record right now.", sports.get_football_record()) - - def test_get_mens_basketball_standings(self): - self.assertEqual("12th in ACC", sports.get_mens_basketball_standings()) - - def test_get_football_standings(self): - self.assertEqual("1st in ACC - Coastal", sports.get_football_standings()) - - def test_get_next_mens_basketball_game(self): - next_game_details = sports.get_next_mens_basketball_game() - self.assertEqual("GAME_COMPLETE", next_game_details.status) - self.assertEqual("103", next_game_details.opponent["id"]) - self.assertEqual("home", next_game_details.home_away) - - def test_get_next_football_game(self): - next_game_details = sports.get_next_football_game() - self.assertEqual("IN_PROGRESS", next_game_details.status) - self.assertEqual("103", next_game_details.opponent["id"]) - self.assertEqual("away", next_game_details.home_away) + } - def test_get_next_mens_basketball_game_offseason(self): - offseason_data = {"team": {"nextEvent": []}} - self.mock_get_basketball_data.return_value = offseason_data - next_game_details = sports.get_next_mens_basketball_game() - self.assertIsNone(next_game_details.timestamp) - self.assertIsNone(next_game_details.opponent) - self.assertIsNone(next_game_details.home_away) - self.assertIsNone(next_game_details.location) - self.assertEqual("NO_GAME_SCHEDULED", next_game_details.status) +@responses.activate +def test_all_sports_client_methods(): + basketball = team_data() + football = team_data("STATUS_IN_PROGRESS") + for _ in range(3): + responses.add(responses.GET, MENS_BASKETBALL_URL, json=basketball) + responses.add(responses.GET, FOOTBALL_URL, json=football) - def test_get_next_football_game_offseason(self): - offseason_data = {"team": {"nextEvent": []}} - self.mock_get_football_data.return_value = offseason_data + client = SportsClient() + assert client.get_mens_basketball_record() == "11-21" + assert client.get_next_mens_basketball_game().status == "GAME_COMPLETE" + assert client.get_mens_basketball_standings() == "12th in ACC" + assert client.get_football_record() == "11-21" + assert client.get_next_football_game().status == "IN_PROGRESS" + assert client.get_football_standings() == "12th in ACC" - next_game_details = sports.get_next_football_game() - self.assertIsNone(next_game_details.timestamp) - self.assertIsNone(next_game_details.opponent) - self.assertIsNone(next_game_details.home_away) - self.assertIsNone(next_game_details.location) - self.assertEqual("NO_GAME_SCHEDULED", next_game_details.status) - def test_basketball_in_progress_with_pitt_second(self): - basketball_data = deepcopy(self.mocked_basketball_data) - competition = basketball_data["team"]["nextEvent"][0]["competitions"][0] - competition["status"]["type"]["name"] = "STATUS_IN_PROGRESS" - competition["competitors"].reverse() - self.mock_get_basketball_data.return_value = basketball_data +def test_scheduled_game_with_pitt_second(): + data = team_data("STATUS_SCHEDULED") + data["team"]["nextEvent"][0]["competitions"][0]["competitors"].reverse() - game = sports.get_next_mens_basketball_game() + game = parse_next_game(data) - self.assertEqual(game.status, "IN_PROGRESS") - self.assertEqual(game.opponent["id"], "103") - self.assertEqual(game.home_away, "home") + assert game.status == "SCHEDULED" + assert game.home_away == "home" + assert game.opponent == Team(id="103", school="Boston College", name="Boston College Eagles") + assert game.location.address.city == "Brooklyn" - def test_basketball_scheduled_game(self): - basketball_data = deepcopy(self.mocked_basketball_data) - basketball_data["team"]["nextEvent"][0]["competitions"][0]["status"]["type"]["name"] = "STATUS_SCHEDULED" - self.mock_get_basketball_data.return_value = basketball_data - self.assertIsNone(sports.get_next_mens_basketball_game().status) +def test_no_scheduled_game(): + assert parse_next_game({"team": {"nextEvent": []}}) == GameInfo(status="NO_GAME_SCHEDULED") - def test_football_final_game(self): - football_data = deepcopy(self.mocked_football_data) - football_data["team"]["nextEvent"][0]["competitions"][0]["status"]["type"]["name"] = "STATUS_FINAL" - self.mock_get_football_data.return_value = football_data - self.assertEqual(sports.get_next_football_game().status, "GAME_COMPLETE") +def test_record_fallback_and_standings_error(): + assert parse_record({"team": {"record": {}}}) == "There's no record right now." + with pytest.raises(ValueError, match="missing standings"): + parse_standings({}) - def test_football_scheduled_with_pitt_second(self): - football_data = deepcopy(self.mocked_football_data) - competition = football_data["team"]["nextEvent"][0]["competitions"][0] - competition["status"]["type"]["name"] = "STATUS_SCHEDULED" - competition["competitors"].reverse() - self.mock_get_football_data.return_value = football_data - game = sports.get_next_football_game() +@pytest.mark.parametrize( + "competitors", + [ + [], + [{"id": "1"}, {"id": "2"}], + ], +) +def test_invalid_competitors(competitors): + with pytest.raises(ValueError): + find_competitors(competitors, "221") - self.assertIsNone(game.status) - self.assertEqual(game.opponent["id"], "103") - self.assertEqual(game.home_away, "away") +def test_malformed_game_data(): + with pytest.raises(ValueError, match="missing game data"): + parse_next_game({"team": {"nextEvent": [{}]}}) -class SportsHttpHelperTest(unittest.TestCase): - @responses.activate - def test_sports_http_helpers(self): - basketball_data = {"team": {"id": "221"}} - football_data = {"team": {"id": "221"}} - responses.add(responses.GET, sports.MENS_BASKETBALL_URL, json=basketball_data, status=200) - responses.add(responses.GET, sports.FOOTBALL_URL, json=football_data, status=200) - self.assertEqual(sports._get_mens_basketball_data(), basketball_data) - self.assertEqual(sports._get_football_data(), football_data) +@responses.activate +def test_sports_response_must_be_object(): + responses.add(responses.GET, FOOTBALL_URL, json=[]) + with pytest.raises(ValueError, match="must be an object"): + SportsClient().get_team_data(FOOTBALL_URL) diff --git a/tests/textbook_test.py b/tests/textbook_test.py index 22fde3b..9d6f2c3 100644 --- a/tests/textbook_test.py +++ b/tests/textbook_test.py @@ -1,499 +1,310 @@ -""" -The Pitt API, to access workable data of the University of Pittsburgh -Copyright (C) 2015 Ritwik Gupta - -This program is free software; you can redistribute it and/or modify -it under the terms of the GNU General Public License as published by -the Free Software Foundation; either version 2 of the License, or -(at your option) any later version. - -This program is distributed in the hope that it will be useful, -but WITHOUT ANY WARRANTY; without even the implied warranty of -MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -GNU General Public License for more details. - -You should have received a copy of the GNU General Public License along -with this program; if not, write to the Free Software Foundation, Inc., -51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. -""" - -from pittapi import textbook +import json +from pathlib import Path +import pytest +import requests import responses -import json -import unittest -from pathlib import Path -from pytest import mark -from requests import ConnectionError, HTTPError -from typing import Any - -SAMPLE_PATH = Path() / "tests" / "samples" -CSRF_TOKEN = "1MTtTVOcQCCXDjKNKTqkfiwp0lmLWz1RvFy2ed65XeyGO4on-8zWsQpEAt4cjiH0glx9CIyjhAOKpXhIqDK_vg" -CS_SUBJECT_ID = "22457" -MATH_SUBJECT_ID = "22528" -CS_0441_GARRISON_SECTION_ID = "4558031" -MATH_0430_PAN_SECTION_ID = "4631097" - - -class TextbookTest(unittest.TestCase): - def __init__(self, *args, **kwargs): - unittest.TestCase.__init__(self, *args, **kwargs) - with (SAMPLE_PATH / "textbook_base_page.html").open() as f: - self.html_text = f.read() - with (SAMPLE_PATH / "textbook_subjects.json").open() as f: - self.subjects_data = json.load(f) - with (SAMPLE_PATH / "textbook_courses_CS.json").open() as f: - self.cs_data = json.load(f) - with (SAMPLE_PATH / "textbook_courses_MATH.json").open() as f: - self.math_data = json.load(f) - with (SAMPLE_PATH / "textbook_textbooks_CS_0441_garrison.json").open() as f: - self.cs_0441_textbook_data: list[dict[str, Any]] = json.load(f) - with (SAMPLE_PATH / "textbook_textbooks_MATH_0430_pan.json").open() as f: - self.math_0430_textbook_data: list[dict[str, Any]] = json.load(f) - - def setUp(self): - textbook.request_headers = None - textbook.subject_map = None - responses.start() - - def tearDown(self): - responses.stop() - responses.reset() - - def mock_base_site_success(self): - responses.add(responses.GET, "https://pitt.verbacompare.com/", body=self.html_text) - - def mock_base_site_failure(self): - responses.add(responses.GET, "https://pitt.verbacompare.com/", status=400) - - def mock_subject_map_success(self): - responses.add( - responses.GET, - f"https://pitt.verbacompare.com/compare/departments/?term={textbook.CURRENT_TERM_ID}", - json=self.subjects_data, - ) - - def mock_subject_map_failure(self): - responses.add( - responses.GET, f"https://pitt.verbacompare.com/compare/departments/?term={textbook.CURRENT_TERM_ID}", status=400 - ) - - def mock_cs_courses_success(self): - responses.add( - responses.GET, - f"https://pitt.verbacompare.com/compare/courses/?id={CS_SUBJECT_ID}&term_id={textbook.CURRENT_TERM_ID}", - json=self.cs_data, - ) - - def mock_cs_courses_failure(self): - responses.add( - responses.GET, - f"https://pitt.verbacompare.com/compare/courses/?id={CS_SUBJECT_ID}&term_id={textbook.CURRENT_TERM_ID}", - status=400, - ) - - def mock_cs_0441_garrison_books_success(self): - responses.add( - responses.GET, - f"https://pitt.verbacompare.com/compare/books?id={CS_0441_GARRISON_SECTION_ID}", - json=self.cs_0441_textbook_data, - ) - - def mock_cs_0441_garrison_books_none(self): - responses.add(responses.GET, f"https://pitt.verbacompare.com/compare/books?id={CS_0441_GARRISON_SECTION_ID}", json=[]) - - def mock_math_courses_success(self): - responses.add( - responses.GET, - f"https://pitt.verbacompare.com/compare/courses/?id={MATH_SUBJECT_ID}&term_id={textbook.CURRENT_TERM_ID}", - json=self.math_data, - ) - - def mock_math_courses_failure(self): - responses.add( - responses.GET, - f"https://pitt.verbacompare.com/compare/courses/?id={MATH_SUBJECT_ID}&term_id={textbook.CURRENT_TERM_ID}", - status=400, - ) - - def mock_math_0430_pan_books_success(self): - responses.add( - responses.GET, - f"https://pitt.verbacompare.com/compare/books?id={MATH_0430_PAN_SECTION_ID}", - json=self.math_0430_textbook_data, - ) - - def test_course_info(self): - self.mock_base_site_success() - self.mock_subject_map_success() - subject, course_num, instructor, section_num = "CS", "0441", "GARRISON III", "1245" - - course = textbook.CourseInfo(subject, course_num, instructor, section_num) - - self.assertEqual(course.subject, subject) - self.assertEqual(course.course_num, course_num) - self.assertEqual(course.instructor, instructor) - self.assertEqual(course.section_num, section_num) - - def test_course_info_convert_input(self): - self.mock_base_site_success() - self.mock_subject_map_success() - subject, course_num, instructor, section_num = "cs", "441", "garrison iii", "1245" - - course = textbook.CourseInfo(subject, course_num, instructor, section_num) - - self.assertEqual(course.subject, "CS") - self.assertEqual(course.course_num, "0441") - self.assertEqual(course.instructor, "GARRISON III") - self.assertEqual(course.section_num, section_num) - - def test_course_info_missing_instructor_and_section_num(self): - self.mock_base_site_success() - self.mock_subject_map_success() - subject, course_num = "cs", "0441" - - course = textbook.CourseInfo(subject, course_num) - - self.assertEqual(course.subject, "CS") - self.assertEqual(course.course_num, "0441") - self.assertIsNone(course.instructor) - self.assertIsNone(course.section_num) - - def test_course_info_invalid_subject(self): - self.mock_base_site_success() - self.mock_subject_map_success() - subject, course_num, instructor, section_num = "fake_subject", "0441", "GARRISON III", "1245" - - self.assertRaises(LookupError, textbook.CourseInfo, subject, course_num, instructor, section_num) - - def test_course_info_invalid_course_num(self): - self.mock_base_site_success() - self.mock_subject_map_success() - subject, course_num, instructor, section_num = "cs", "abc", "GARRISON III", "1245" - - self.assertRaises(ValueError, textbook.CourseInfo, subject, course_num, instructor, section_num) - - course_num = "44111" - - self.assertRaises(ValueError, textbook.CourseInfo, subject, course_num, instructor, section_num) - - def test_course_info_invalid_section_num(self): - self.mock_base_site_success() - self.mock_subject_map_success() - subject, course_num, instructor, section_num = "cs", "0441", "GARRISON III", "12456" - - self.assertRaises(ValueError, textbook.CourseInfo, subject, course_num, instructor, section_num) - - @mark.filterwarnings("ignore:Attempt") - @responses.activate - def test_course_info_failing_header_requests(self): - self.mock_base_site_failure() - - self.assertRaises(ConnectionError, textbook.CourseInfo, "CS", "0441", instructor="GARRISON III") - - @responses.activate - def test_course_info_no_headers(self): - responses.add(responses.GET, "https://pitt.verbacompare.com/", body="") - - self.assertRaises(ConnectionError, textbook.CourseInfo, "CS", "0441", instructor="GARRISON III") - - @responses.activate - def test_update_headers_rejects_csrf_meta_without_content(self): - responses.add( - responses.GET, - "https://pitt.verbacompare.com/", - body="", - ) - - with self.assertRaises(ConnectionError): - textbook._update_headers() - - @mark.filterwarnings("ignore:Attempt") - @responses.activate - def test_course_info_failing_subject_map_requests(self): - self.mock_base_site_success() - self.mock_subject_map_failure() - - self.assertRaises(ConnectionError, textbook.CourseInfo, "CS", "0441", instructor="GARRISON III") - - @responses.activate - def test_update_subject_map_with_existing_headers(self): - textbook.request_headers = {"X-CSRF-Token": CSRF_TOKEN} - self.mock_subject_map_success() - - textbook._update_subject_map() - - self.assertEqual(textbook.subject_map["CS"], CS_SUBJECT_ID) - - def test_textbook_from_json(self): - self.assertEqual(len(self.cs_0441_textbook_data), 1) - - textbook_info = textbook.Textbook.from_json(self.cs_0441_textbook_data[0]) - - self.assertIsNotNone(textbook_info) - self.assertEqual(textbook_info.title, "Ia Canvas Content") - self.assertEqual(textbook_info.author, "Redshelf Ia") - self.assertIsNone(textbook_info.edition) - self.assertEqual(textbook_info.isbn, "BSZWEWZWMZYJ") - self.assertEqual(textbook_info.citation, "Ia Canvas Content by Redshelf Ia. (ISBN: BSZWEWZWMZYJ).") - - def test_textbook_from_json_all_empty(self): - emptied_data: dict[str, Any] = self.cs_0441_textbook_data[0].copy() - emptied_data.pop("title") - emptied_data.pop("author") - emptied_data.pop("edition") - emptied_data.pop("isbn") - emptied_data.pop("citation") - - textbook_info = textbook.Textbook.from_json(emptied_data) - - self.assertIsNone(textbook_info) - - @responses.activate - def test_get_textbooks_for_course_section_num(self): - self.mock_base_site_success() - self.mock_subject_map_success() - self.mock_cs_courses_success() - self.mock_cs_0441_garrison_books_success() - course = textbook.CourseInfo("CS", "0441", section_num="1245") - - textbooks = textbook.get_textbooks_for_course(course) - - self.assertEqual(textbook.request_headers, {"X-CSRF-Token": CSRF_TOKEN}) - self.assertEqual(len(textbook.subject_map), 168) - self.assertEqual(textbook.subject_map["CS"], CS_SUBJECT_ID) - self.assertEqual(len(textbooks), 1) - self.assertEqual(textbooks[0].title, "Ia Canvas Content") - self.assertEqual(textbooks[0].author, "Redshelf Ia") - self.assertIsNone(textbooks[0].edition) - self.assertEqual(textbooks[0].isbn, "BSZWEWZWMZYJ") - self.assertEqual(textbooks[0].citation, "Ia Canvas Content by Redshelf Ia. (ISBN: BSZWEWZWMZYJ).") - - @responses.activate - def test_get_textbooks_for_course_invalid_section_num(self): - self.mock_base_site_success() - self.mock_subject_map_success() - self.mock_cs_courses_success() - self.mock_cs_0441_garrison_books_success() - course = textbook.CourseInfo("CS", "0441", section_num="0000") - - self.assertRaises(LookupError, textbook.get_textbooks_for_course, course) - - @responses.activate - def test_get_textbooks_for_course_instructor(self): - self.mock_base_site_success() - self.mock_subject_map_success() - self.mock_cs_courses_success() - self.mock_cs_0441_garrison_books_success() - course = textbook.CourseInfo("CS", "0441", instructor="GARRISON III") - - textbooks = textbook.get_textbooks_for_course(course) - - self.assertEqual(textbook.request_headers, {"X-CSRF-Token": CSRF_TOKEN}) - self.assertEqual(len(textbook.subject_map), 168) - self.assertEqual(textbook.subject_map["CS"], CS_SUBJECT_ID) - self.assertEqual(len(textbooks), 1) - self.assertEqual(textbooks[0].title, "Ia Canvas Content") - self.assertEqual(textbooks[0].author, "Redshelf Ia") - self.assertIsNone(textbooks[0].edition) - self.assertEqual(textbooks[0].isbn, "BSZWEWZWMZYJ") - self.assertEqual(textbooks[0].citation, "Ia Canvas Content by Redshelf Ia. (ISBN: BSZWEWZWMZYJ).") - - @responses.activate - def test_get_textbooks_for_course_invalid_instructor(self): - self.mock_base_site_success() - self.mock_subject_map_success() - self.mock_cs_courses_success() - self.mock_cs_0441_garrison_books_success() - course = textbook.CourseInfo("CS", "0441", instructor="RAMIREZ") - - self.assertRaises(LookupError, textbook.get_textbooks_for_course, course) - - @responses.activate - def test_get_textbooks_for_course_invalid_course(self): - self.mock_base_site_success() - self.mock_subject_map_success() - self.mock_cs_courses_success() - course = textbook.CourseInfo("CS", "0000") - - self.assertRaises(LookupError, textbook.get_textbooks_for_course, course) - - @responses.activate - def test_get_textbooks_for_course_deduce_section(self): - self.mock_base_site_success() - self.mock_subject_map_success() - self.mock_math_courses_success() - self.mock_math_0430_pan_books_success() - course = textbook.CourseInfo("MATH", "0430") - - textbooks = textbook.get_textbooks_for_course(course) - - self.assertEqual(len(textbooks), 1) - self.assertEqual(textbooks[0].title, "First Course In Abstract Algebra") - self.assertEqual(textbooks[0].author, "Fraleigh") - self.assertEqual(textbooks[0].edition, "7") - self.assertEqual(textbooks[0].isbn, "9780201763904") - self.assertEqual( - textbooks[0].citation, - "\u003cem\u003eFirst Course In Abstract Algebra\u003c/em\u003e by Fraleigh. " - "Pearson Education, 7th Edition, 2002. (ISBN: 9780201763904).", - ) - - @responses.activate - def test_get_textbooks_for_course_not_enough_info(self): - self.mock_base_site_success() - self.mock_subject_map_success() - self.mock_cs_courses_success() - course = textbook.CourseInfo("CS", "0441") - - self.assertRaises(LookupError, textbook.get_textbooks_for_course, course) - - @mark.filterwarnings("ignore:Attempt") - @responses.activate - def test_get_textbooks_for_course_failing_courses_requests(self): - self.mock_base_site_success() - self.mock_subject_map_success() - self.mock_cs_courses_failure() - course = textbook.CourseInfo("CS", "0441", instructor="GARRISON III") - - self.assertRaises(ConnectionError, textbook.get_textbooks_for_course, course) - - @responses.activate - def test_get_textbooks_for_course_no_textbook(self): - self.mock_base_site_success() - self.mock_subject_map_success() - self.mock_cs_courses_success() - self.mock_cs_0441_garrison_books_none() - course = textbook.CourseInfo("CS", "0441", instructor="GARRISON III") - - textbooks = textbook.get_textbooks_for_course(course) - - self.assertEqual(len(textbooks), 0) - - @mark.filterwarnings("ignore:No textbook info found") - @responses.activate - def test_get_textbooks_for_course_textbook_no_info(self): - emptied_data: dict[str, Any] = self.cs_0441_textbook_data[0].copy() - emptied_data.pop("title") - emptied_data.pop("author") - emptied_data.pop("edition") - emptied_data.pop("isbn") - emptied_data.pop("citation") - self.mock_base_site_success() - self.mock_subject_map_success() - self.mock_cs_courses_success() - responses.add( - responses.GET, f"https://pitt.verbacompare.com/compare/books?id={CS_0441_GARRISON_SECTION_ID}", json=[emptied_data] - ) - course = textbook.CourseInfo("CS", "0441", instructor="GARRISON III") - - textbook_info = textbook.get_textbooks_for_course(course) - - self.assertEqual(len(textbook_info), 0) - - @responses.activate - def test_get_textbooks_for_courses(self): - self.mock_base_site_success() - self.mock_subject_map_success() - self.mock_cs_courses_success() - self.mock_math_courses_success() - self.mock_cs_0441_garrison_books_success() - self.mock_math_0430_pan_books_success() - courses = [ - textbook.CourseInfo("CS", "0441", instructor="GARRISON III"), - textbook.CourseInfo("MATH", "0430", instructor="PAN"), - ] - - textbooks = textbook.get_textbooks_for_courses(courses) - - self.assertEqual(len(textbooks), 2) - self.assertEqual(textbooks[0].title, "Ia Canvas Content") - self.assertEqual(textbooks[0].author, "Redshelf Ia") - self.assertIsNone(textbooks[0].edition) - self.assertEqual(textbooks[0].isbn, "BSZWEWZWMZYJ") - self.assertEqual(textbooks[0].citation, "Ia Canvas Content by Redshelf Ia. (ISBN: BSZWEWZWMZYJ).") - - self.assertEqual(textbooks[1].title, "First Course In Abstract Algebra") - self.assertEqual(textbooks[1].author, "Fraleigh") - self.assertEqual(textbooks[1].edition, "7") - self.assertEqual(textbooks[1].isbn, "9780201763904") - self.assertEqual( - textbooks[1].citation, - "\u003cem\u003eFirst Course In Abstract Algebra\u003c/em\u003e by Fraleigh. " - "Pearson Education, 7th Edition, 2002. (ISBN: 9780201763904).", - ) - - @responses.activate - def test_get_textbooks_for_ids_propagates_request_failure(self): - textbook.request_headers = {"X-CSRF-Token": CSRF_TOKEN} - responses.add( - responses.GET, - f"https://pitt.verbacompare.com/compare/books?id={CS_0441_GARRISON_SECTION_ID}", - status=503, - ) - - with self.assertRaises(HTTPError): - textbook._get_textbooks_for_ids([CS_0441_GARRISON_SECTION_ID]) - - @responses.activate - def test_get_textbooks_for_ids_initializes_headers(self): - self.mock_base_site_success() - self.mock_cs_0441_garrison_books_success() - - books = textbook._get_textbooks_for_ids([CS_0441_GARRISON_SECTION_ID]) - - self.assertEqual(len(books), 1) - self.assertEqual(textbook.request_headers, {"X-CSRF-Token": CSRF_TOKEN}) - - @responses.activate - def test_get_textbooks_for_course_initializes_headers(self): - self.mock_base_site_success() - self.mock_subject_map_success() - course_info = textbook.CourseInfo("CS", "0441", instructor="GARRISON III") - textbook.request_headers = None - self.mock_base_site_success() - self.mock_cs_courses_success() - self.mock_cs_0441_garrison_books_success() - - self.assertEqual(len(textbook.get_textbooks_for_course(course_info)), 1) - - @responses.activate - def test_get_textbooks_for_course_initializes_subject_map(self): - self.mock_base_site_success() - self.mock_subject_map_success() - course_info = textbook.CourseInfo("CS", "0441", instructor="GARRISON III") - textbook.subject_map = None - self.mock_subject_map_success() - self.mock_cs_courses_success() - self.mock_cs_0441_garrison_books_success() - - self.assertEqual(len(textbook.get_textbooks_for_course(course_info)), 1) - - @responses.activate - def test_get_textbooks_for_courses_initializes_global_state(self): - self.mock_base_site_success() - self.mock_subject_map_success() - course_info = textbook.CourseInfo("CS", "0441", instructor="GARRISON III") - textbook.request_headers = None - textbook.subject_map = None - self.mock_base_site_success() - self.mock_subject_map_success() - self.mock_cs_courses_success() - self.mock_cs_0441_garrison_books_success() - - self.assertEqual(len(textbook.get_textbooks_for_courses([course_info])), 1) - - @mark.filterwarnings("ignore:Attempt") - @responses.activate - def test_get_textbooks_for_courses_failing_courses_requests(self): - self.mock_base_site_success() - self.mock_subject_map_success() - self.mock_cs_courses_failure() - self.mock_math_courses_failure() - courses = [ - textbook.CourseInfo("CS", "0441", instructor="GARRISON III"), - textbook.CourseInfo("MATH", "0430", instructor="PAN"), - ] - - self.assertRaises(ConnectionError, textbook.get_textbooks_for_courses, courses) +from pittapi.textbook import ( + BASE_URL, + BOOKS_URL, + COURSES_URL, + MAX_REQUEST_ATTEMPTS, + SUBJECTS_URL, + CourseInfo, + Textbook, + TextbookClient, + TextbookTerm, + find_section, + find_section_id, + parse_textbook, +) + +SAMPLES = Path("tests/samples") +BASE_HTML = (SAMPLES / "textbook_base_page.html").read_text() +SUBJECTS = json.loads((SAMPLES / "textbook_subjects.json").read_text()) +CS_COURSES = json.loads((SAMPLES / "textbook_courses_CS.json").read_text()) +MATH_COURSES = json.loads((SAMPLES / "textbook_courses_MATH.json").read_text()) +CS_BOOKS = json.loads((SAMPLES / "textbook_textbooks_CS_0441_garrison.json").read_text()) +MATH_BOOKS = json.loads((SAMPLES / "textbook_textbooks_MATH_0430_pan.json").read_text()) +CS_ID = "22457" +MATH_ID = "22528" +CS_SECTION = "4558031" +MATH_SECTION = "4631097" +HEADERS = {"X-CSRF-Token": "token"} +TERM = TextbookTerm("78104", "Fall 24", True, True) + + +def subjects_url(term=TERM): + return SUBJECTS_URL.format(term_id=term.id) + + +def courses_url(subject_id, term=TERM): + return COURSES_URL.format(department_id=subject_id, term_id=term.id) + + +@pytest.mark.parametrize( + ("course", "expected"), + [ + (CourseInfo("cs", "441", "garrison iii", "1245"), CourseInfo("CS", "0441", "GARRISON III", "1245")), + (CourseInfo("MATH", "0430"), CourseInfo("MATH", "0430")), + ], +) +def test_course_info_normalization(course, expected): + assert course == expected + + +@pytest.mark.parametrize("number", ["abc", "12345", ""]) +def test_invalid_course_number(number): + with pytest.raises(ValueError, match="invalid course number"): + CourseInfo("CS", number) + + +@pytest.mark.parametrize("section", ["123", "abcd"]) +def test_invalid_section_number(section): + with pytest.raises(ValueError, match="invalid section number"): + CourseInfo("CS", "0441", section_num=section) + + +@responses.activate +def test_initialize_headers(): + responses.add(responses.GET, BASE_URL, body=BASE_HTML) + client = TextbookClient() + client.initialize_headers() + assert client.headers + assert client.terms == (TERM,) + + +@responses.activate +def test_get_terms_caches_discovery_and_selecting_a_new_term_resets_subjects(): + responses.add(responses.GET, BASE_URL, body=BASE_HTML) + client = TextbookClient() + assert client.get_terms() == (TERM,) + assert client.get_terms() == (TERM,) + assert len(responses.calls) == 1 + + client.subject_ids = {"CS": CS_ID} + client.select_term(TERM) + assert client.subject_ids is None + client.subject_ids = {"CS": CS_ID} + client.select_term(TERM) + assert client.subject_ids == {"CS": CS_ID} + + +@responses.activate +def test_get_terms_allows_empty_discovery(): + html = BASE_HTML.replace( + '[{"id":"78104","name":"Fall 24","inquiry":true,"ordering":true}]', + "[]", + ) + responses.add(responses.GET, BASE_URL, body=html) + assert TextbookClient().get_terms() == () + + +@pytest.mark.parametrize( + "html", + [ + "", + "", + ], +) +@responses.activate +def test_initialize_headers_requires_token(html): + responses.add(responses.GET, BASE_URL, body=html) + with pytest.raises(requests.ConnectionError, match="credentials"): + TextbookClient().initialize_headers() + + +@pytest.mark.parametrize( + "html", + [ + "", + "", + (""), + ], +) +@responses.activate +def test_initialize_headers_requires_valid_terms(html): + responses.add(responses.GET, BASE_URL, body=html) + with pytest.raises(ValueError, match="term"): + TextbookClient().initialize_headers() + + +@responses.activate +def test_initialize_headers_retries_then_fails(): + responses.add(responses.GET, BASE_URL, status=400) + with pytest.raises(requests.ConnectionError, match=str(MAX_REQUEST_ATTEMPTS)): + TextbookClient().initialize_headers() + assert len(responses.calls) == MAX_REQUEST_ATTEMPTS + + +@responses.activate +def test_initialize_subjects_with_existing_and_missing_headers(): + responses.add(responses.GET, subjects_url(), json=SUBJECTS) + client = TextbookClient(TERM) + client.headers = HEADERS + client.initialize_subjects() + assert client.subject_ids["CS"] == CS_ID + + responses.add(responses.GET, BASE_URL, body=BASE_HTML) + responses.add(responses.GET, subjects_url(), json=SUBJECTS) + fresh_client = TextbookClient(TERM) + fresh_client.initialize_subjects() + assert fresh_client.headers + + +@responses.activate +def test_initialize_subjects_rejects_malformed_data(): + responses.add(responses.GET, subjects_url(), json=[{}]) + client = TextbookClient(TERM) + client.headers = HEADERS + with pytest.raises(ValueError, match="subject response"): + client.initialize_subjects() + + +@responses.activate +def test_initialize_subjects_refreshes_and_eventually_fails(): + responses.add(responses.GET, subjects_url(), status=400) + responses.add(responses.GET, BASE_URL, body=BASE_HTML) + client = TextbookClient(TERM) + client.headers = HEADERS + with pytest.raises(requests.ConnectionError, match="retrieve subjects"): + client.initialize_subjects() + + +@responses.activate +def test_get_courses_success_and_validation(): + responses.add(responses.GET, courses_url(CS_ID), json=CS_COURSES) + client = TextbookClient(TERM) + client.headers = HEADERS + client.subject_ids = {"CS": CS_ID} + assert client.get_courses("CS") == CS_COURSES + + client.subject_ids = None + with pytest.raises(LookupError, match="invalid textbook subject"): + client.get_courses("CS") + + +@responses.activate +def test_get_courses_initializes_headers_and_rejects_nonlist(): + responses.add(responses.GET, BASE_URL, body=BASE_HTML) + responses.add(responses.GET, courses_url(CS_ID), json={}) + client = TextbookClient(TERM) + client.subject_ids = {"CS": CS_ID} + with pytest.raises(ValueError, match="must contain a list"): + client.get_courses("CS") + + +@responses.activate +def test_get_courses_refreshes_and_fails(): + responses.add(responses.GET, courses_url(CS_ID), status=400) + responses.add(responses.GET, BASE_URL, body=BASE_HTML) + client = TextbookClient(TERM) + client.headers = HEADERS + client.subject_ids = {"CS": CS_ID} + with pytest.raises(requests.ConnectionError, match="retrieve CS courses"): + client.get_courses("CS") + + +@responses.activate +def test_get_textbooks_for_one_course(): + responses.add(responses.GET, BASE_URL, body=BASE_HTML) + responses.add(responses.GET, subjects_url(), json=SUBJECTS) + responses.add(responses.GET, courses_url(CS_ID), json=CS_COURSES) + responses.add(responses.GET, BOOKS_URL.format(section_id=CS_SECTION), json=CS_BOOKS) + + books = TextbookClient(TERM).get_textbooks_for_course(CourseInfo("CS", "0441", instructor="GARRISON III")) + + assert len(books) == 1 + assert books[0].title == "Ia Canvas Content" + + +@responses.activate +def test_get_textbooks_for_multiple_courses_and_cache_subject(): + client = TextbookClient(TERM) + client.headers = HEADERS + client.subject_ids = {"CS": CS_ID, "MATH": MATH_ID} + responses.add(responses.GET, courses_url(CS_ID), json=CS_COURSES) + responses.add(responses.GET, courses_url(MATH_ID), json=MATH_COURSES) + responses.add(responses.GET, BOOKS_URL.format(section_id=CS_SECTION), json=CS_BOOKS) + responses.add(responses.GET, BOOKS_URL.format(section_id=CS_SECTION), json=CS_BOOKS) + responses.add(responses.GET, BOOKS_URL.format(section_id=MATH_SECTION), json=MATH_BOOKS) + courses = [ + CourseInfo("CS", "0441", instructor="GARRISON III"), + CourseInfo("CS", "0441", section_num="1245"), + CourseInfo("MATH", "0430", instructor="PAN"), + ] + + books = client.get_textbooks_for_courses(courses) + + assert len(books) == 3 + assert sum(call.request.url == courses_url(CS_ID) for call in responses.calls) == 1 + + +def test_get_textbooks_rejects_unknown_subject(): + client = TextbookClient(TERM) + client.subject_ids = {} + with pytest.raises(LookupError, match="invalid textbook subject"): + client.get_textbooks_for_courses([CourseInfo("FAKE", "0001")]) + + +@responses.activate +def test_section_textbooks_initializes_headers_and_filters_empty_records(): + responses.add(responses.GET, BASE_URL, body=BASE_HTML) + responses.add(responses.GET, BOOKS_URL.format(section_id=CS_SECTION), json=[{}, *CS_BOOKS]) + books = TextbookClient(TERM).get_textbooks_for_section(CS_SECTION) + assert len(books) == 1 + + +@responses.activate +def test_section_textbooks_requires_list(): + responses.add(responses.GET, BOOKS_URL.format(section_id=CS_SECTION), json={}) + client = TextbookClient(TERM) + client.headers = HEADERS + with pytest.raises(ValueError, match="must contain a list"): + client.get_textbooks_for_section(CS_SECTION) + + +def test_textbook_requests_require_selected_term(): + client = TextbookClient() + course = CourseInfo("CS", "0441") + with pytest.raises(ValueError, match="select a textbook term"): + client.initialize_subjects() + with pytest.raises(ValueError, match="select a textbook term"): + client.get_courses("CS") + with pytest.raises(ValueError, match="select a textbook term"): + client.get_textbooks_for_course(course) + with pytest.raises(ValueError, match="select a textbook term"): + client.get_textbooks_for_section("1") + + +def test_find_section_variants(): + sections = [ + {"name": "0001", "instructor": "SAME", "id": "1"}, + {"name": "0002", "instructor": "SAME", "id": "2"}, + ] + assert find_section(sections, None, "0002") == "2" + assert find_section(sections, "SAME", None) == "1" + assert find_section(sections, None, None) == "1" + assert find_section([sections[0]], None, None) == "1" + + with pytest.raises(LookupError, match="section not found"): + find_section(sections, None, "9999") + with pytest.raises(LookupError, match="instructor not found"): + find_section(sections, "OTHER", None) + + ambiguous = [sections[0], sections[1] | {"instructor": "OTHER"}] + with pytest.raises(LookupError, match="provide an instructor"): + find_section(ambiguous, None, None) + + +def test_find_section_id_and_invalid_course(): + course = CourseInfo("CS", "0441", section_num="1245") + assert find_section_id(CS_COURSES, course) == CS_SECTION + with pytest.raises(LookupError, match="invalid textbook course"): + find_section_id(CS_COURSES, CourseInfo("CS", "0001")) + + +def test_parse_textbook(): + assert isinstance(parse_textbook(CS_BOOKS[0]), Textbook) + assert parse_textbook({}) is None