From 9c52d0f183e1bb76a241edf1e0da43b8c195a362 Mon Sep 17 00:00:00 2001 From: Charlotte Kostelic Date: Tue, 27 Feb 2024 12:13:32 -0500 Subject: [PATCH 1/3] added new api endpoints --- bookops_worldcat/metadata_api.py | 1085 +++++++++++++++++++++++++++++- tests/conftest.py | 22 + tests/test_metadata_api.py | 670 ++++++++++++++++++ 3 files changed, 1773 insertions(+), 4 deletions(-) diff --git a/bookops_worldcat/metadata_api.py b/bookops_worldcat/metadata_api.py index 031d7a4..8f83278 100644 --- a/bookops_worldcat/metadata_api.py +++ b/bookops_worldcat/metadata_api.py @@ -27,7 +27,7 @@ def __init__( ) -> None: """ Args: - authorization: WorlcatAccessToken object + authorization: WorldcatAccessToken object agent: "User-agent" parameter to be passed in the request header; usage strongly encouraged timeout: how long to wait for server to send data before @@ -82,6 +82,60 @@ def _url_manage_ih_unset(self, oclcNumber: str) -> str: def _url_manage_ih_current(self) -> str: return f"{self.BASE_URL}/manage/institution/holdings/current" + def _url_manage_bibs_validate(self, validationLevel: str) -> str: + return f"{self.BASE_URL}/manage/bibs/validate/{validationLevel}" + + def _url_manage_bibs_create(self) -> str: + return f"{self.BASE_URL}/manage/bibs" + + def _url_manage_bibs_match(self) -> str: + return f"{self.BASE_URL}/manage/bibs/match" + + def _url_manage_ih_set_on_record(self) -> str: + return f"{self.BASE_URL}/manage/institution/holdings/set" + + def _url_manage_ih_unset_on_record(self) -> str: + return f"{self.BASE_URL}/manage/institution/holdings/unset" + + def _url_manage_ih_codes(self) -> str: + return f"{self.BASE_URL}/manage/institution/holding-codes" + + def _url_manage_lbd_create(self) -> str: + return f"{self.BASE_URL}/manage/lbds" + + def _url_manage_lbd(self, controlNumber: Union[str, int]) -> str: + return f"{self.BASE_URL}/manage/lbds/{controlNumber}" + + def _url_manage_lhr_create(self) -> str: + return f"{self.BASE_URL}/manage/lhrs" + + def _url_manage_lhr(self, controlNumber: Union[str, int]) -> str: + return f"{self.BASE_URL}/manage/lhrs/{controlNumber}" + + def _url_search_general_holdings_summary(self) -> str: + return f"{self.BASE_URL}/search/summary-holdings" + + def _url_search_classification_bibs(self, oclcNumber: str) -> str: + return f"{self.BASE_URL}/search/classification-bibs/{oclcNumber}" + + def _url_search_lhr_shared_print(self) -> str: + return f"{self.BASE_URL}/search/retained-holdings" + + def _url_search_lhr_control_number(self, controlNumber: Union[str, int]) -> str: + return f"{self.BASE_URL}/search/my-holdings/{controlNumber}" + + def _url_search_lhr(self) -> str: + return f"{self.BASE_URL}/search/my-holdings" + + def _url_browse_lhr(self) -> str: + return f"{self.BASE_URL}/browse/my-holdings" + + def _url_search_lbd_control_number(self, controlNumber: Union[str, int]) -> str: + return f"{self.BASE_URL}/search/my-local-bib-data/{controlNumber}" + + def _url_search_lbd(self) -> str: + return f"{self.BASE_URL}/search/my-local-bib-data" + def get_brief_bib( self, oclcNumber: Union[int, str], hooks: Optional[Dict[str, Callable]] = None ) -> Optional[Response]: @@ -311,7 +365,7 @@ def search_brief_bibs_other_editions( heldByGroup: restricts to holdings held by group symbol heldBySymbol: restricts to holdings with specified intitution symbol - heldByInstitutionID: restrict to specified institution regisgtryId + heldByInstitutionID: restrict to specified institution registryId inLanguage: restrics the response to the single specified language, example: 'fre' inCataloglanguage: restrics the response to specified @@ -351,7 +405,7 @@ def search_brief_bibs_other_editions( preferredLanguage: language of metadata description, offset: start position of bibliographic records to return; default 1 - limit: maximum nuber of records to return; + limit: maximum number of records to return; maximum 50, default 10 orderBy: sort of restuls; available values: @@ -526,7 +580,7 @@ def search_brief_bibs( 'title' offset: start position of bibliographic records to return; default 1 - limit: maximum nuber of records to return; + limit: maximum number of records to return; maximum 50, default 10 hooks: Requests library hook system that can be used for signal event handling, see more at: @@ -787,3 +841,1026 @@ def search_shared_print_holdings( query = Query(self, prepared_request, timeout=self.timeout) return query.response + + def browse_my_holdings( + self, + callNumber: Optional[str] = None, + oclcNumber: Optional[Union[int, str]] = None, + holdingLocation: str = "", + shelvingLocation: str = "", + browsePosition: Optional[str] = None, + limit: Optional[int] = None, + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Browse local holdings + Uses /browse/my-holdings endpoint. + + Args: + callNumber: call number for item + oclcNumber: OCLC bibliographic record number; can be + an integer or string with or without OCLC # + prefix + holdingLocation: holding location for item + shelvingLocation: shelving location for item + browsePosition: position within browse list where the matching + record should be, default is 10 + limit: maximum number of records to return; + maximum 50, default 10 + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` instance + """ + if not holdingLocation: + raise TypeError("Argument 'holdingLocation' is missing.") + if not shelvingLocation: + raise TypeError("Argument 'shelvingLocation' is missing.") + + if oclcNumber is not None: + oclcNumber = verify_oclc_number(oclcNumber) + + url = self._url_browse_lhr() + header = {"Accept": "application/json"} + payload = { + "callNumber": callNumber, + "oclcNumber": oclcNumber, + "holdingLocation": holdingLocation, + "shelvingLocation": shelvingLocation, + "browsePosition": browsePosition, + "limit": limit, + } + + # prep request + req = Request("GET", url, params=payload, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def create_bib( + self, + record: Optional[str] = None, + recordFormat: Optional[str] = None, + responseFormat: Optional[str] = "application/marcxml+xml", + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Create a bib record in OCLC if it does not already exist + Uses /manage/bibs endpoint. + + Args: + record: MARC record to be created + recordFormat: format of MARC record, options: + 'application/marcxml+xml', 'application/marc' + responseFormat: format of returned record; options: + 'application/marcxml+xml', 'application/marc' + default is 'application/marcxml+xml' + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` instance + """ + if not record: + raise TypeError("Argument 'record' is missing.") + if not recordFormat: + raise TypeError("Argument 'recordFormat' is missing.") + + url = self._url_manage_bibs_create() + header = { + "Accept": responseFormat, + "content-type": recordFormat, + } + + # prep request + req = Request("POST", url, data=record, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def create_lbd( + self, + record: Optional[str] = None, + recordFormat: Optional[str] = None, + responseFormat: Optional[str] = "application/marcxml+xml", + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Given a local bibliographic data record, create it in WorldCat + Uses /manage/lbds endpoint. + + Args: + record: MARC record to be created + recordFormat: format of MARC record, options: + 'application/marcxml+xml', 'application/marc' + responseFormat: format of returned record; options: + 'application/marcxml+xml', 'application/marc' + default is 'application/marcxml+xml' + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` instance + """ + if not record: + raise TypeError("Argument 'record' is missing.") + if not recordFormat: + raise TypeError("Argument 'recordFormat' is missing.") + + url = self._url_manage_lbd_create() + header = { + "Accept": responseFormat, + "content-type": recordFormat, + } + + # prep request + req = Request("POST", url, data=record, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def create_lhr( + self, + record: Optional[str] = None, + recordFormat: Optional[str] = None, + responseFormat: Optional[str] = "application/marcxml+xml", + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Given a local holdings record, create it in WorldCat + Uses /manage/lhrs endpoint. + + Args: + record: Holdings record to be created + recordFormat: format of MARC record, options: + 'application/marcxml+xml', 'application/marc' + responseFormat: format of returned record; options: + 'application/marcxml+xml', 'application/marc' + default is 'application/marcxml+xml' + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` instance + """ + if not record: + raise TypeError("Argument 'record' is missing.") + if not recordFormat: + raise TypeError("Argument 'recordFormat' is missing.") + + url = self._url_manage_lhr_create() + header = { + "Accept": responseFormat, + "content-type": recordFormat, + } + + # prep request + req = Request("POST", url, data=record, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def delete_lbd( + self, + controlNumber: Union[int, str], + responseFormat: Optional[str] = "application/marcxml+xml", + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Given a control number, delete the associated Local Bibliographic Data record + Uses /manage/lbds/{controlNumber} endpoint. + + Args: + controlNumber: control number associated with Local Bibliographic + Data record; can be an integer, or string + response_format: format of returned record, options: + 'application/marcxml+xml', 'application/marc', + default is 'application/marcxml+xml' + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + url = self._url_manage_lbd(controlNumber) + header = {"Accept": responseFormat} + + # prep request + req = Request("DELETE", url, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def delete_lhr( + self, + controlNumber: Union[int, str], + responseFormat: Optional[str] = "application/marcxml+xml", + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Given a control number, delete a Local Holdings record. + Uses /manage/lhrs/{controlNumber} endpoint. + + Args: + controlNumber: control number associated with Local Holdings + record; can be an integer, or string + response_format: format of returned record, options: + 'application/marcxml+xml', 'application/marc', + default is 'application/marcxml+xml' + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + url = self._url_manage_lhr(controlNumber) + header = {"Accept": responseFormat} + + # prep request + req = Request("DELETE", url, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def get_bib_classification( + self, + oclcNumber: Union[int, str], + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Given an OCLC number, retrieve classification recommendations for the bib record + Uses /search/classification-bibs/{oclcNumber} endpoint. + + Args: + oclcNumber: OCLC bibliographic record number; can be + an integer or string with or without OCLC # prefix + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + oclcNumber = verify_oclc_number(oclcNumber) + + url = self._url_search_classification_bibs(oclcNumber) + header = {"Accept": "application/json"} + + # prep request + req = Request("GET", url, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def get_bib_holdings( + self, + oclcNumber: Union[int, str], + holdingsAllEditions: Optional[bool] = None, + holdingsAllVariantRecords: Optional[bool] = None, + holdingsFilterFormat: Optional[List[str]] = None, + heldInCountry: Optional[str] = None, + heldInState: Optional[str] = None, + heldByGroup: Optional[str] = None, + heldBySymbol: Optional[List[str]] = None, + heldByInstitutionID: Optional[List[int]] = None, + heldByLibraryType: Optional[List[str]] = None, + lat: Optional[float] = None, + lon: Optional[float] = None, + distance: Optional[int] = None, + unit: Optional[str] = None, + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Given an OCLC number get summary of holdings + Uses /search/summary-holdings endpoint. + + Args: + oclcNumber: OCLC bibliographic record number; can be + an integer or string with or without OCLC # + prefix + holdingsAllEditions: get holdings for all editions; + options: True, False, default is False + holdingsAllVariantRecords: get holdings for specific edition across + all variant records; options: True, False, + default is False + holdingsFilterFormat: get holdings for specific itemSubType, + example: book-digital + heldInCountry: limits to holdings held by institutions + in requested country + heldInState: limits to holdings held by institutions + in requested state, example: 'US-NY' + heldByGroup: limits to holdings held by institutions + indicated by group symbol + heldBySymbol: limits to holdings held by institutions + indicated by institution symbol + heldByInstitutionID: limits to holdings held by institutions + indicated by institution registryID + heldByLibraryType: limits to holdings held by library type, + options: 'PUBLIC', 'ALL' + lat: limit to latitude, example: 37.502508 + lon: limit to longitute, example: -122.22702 + distance: distance from latitude and longitude + unit: unit of distance param; options: + 'M' (miles) or 'K' (kilometers) + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + oclcNumber = verify_oclc_number(oclcNumber) + + url = self._url_search_general_holdings_summary() + header = {"Accept": "application/json"} + payload = { + "oclcNumber": oclcNumber, + "holdingsAllEditions": holdingsAllEditions, + "holdingsAllVariantRecords": holdingsAllVariantRecords, + "holdingsFilterFormat": holdingsFilterFormat, + "heldInCountry": heldInCountry, + "heldInState": heldInState, + "heldByGroup": heldByGroup, + "heldBySymbol": heldBySymbol, + "heldByInstitutionID": heldByInstitutionID, + "heldByLibraryType": heldByLibraryType, + "lat": lat, + "lon": lon, + "distance": distance, + "unit": unit, + } + + # prep request + req = Request("GET", url, params=payload, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def get_institution_holding_codes( + self, + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Retrieve the all holding codes for the authenticated institution. + Uses /manage/institution/holding-codes endpoint. + + Args: + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + url = self._url_manage_ih_codes() + header = {"Accept": "application/json"} + + # prep request + req = Request("GET", url, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def get_lbd_data( + self, + controlNumber: Union[int, str], + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Retrieve LBD Resource + Uses /search/my-local-bib-data/{controlNumber} endpoint. + + Args: + controlNumber: control number associated with Local Bibliographic + Data record; can be an integer, or string + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + url = self._url_search_lbd_control_number(controlNumber) + header = {"Accept": "application/json"} + + # prep request + req = Request("GET", url, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def get_lhr_data( + self, + controlNumber: Union[int, str], + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Retrieve LHR Resource + Uses /search/my-holdings/{controlNumber} endpoint. + + Args: + controlNumber: control number associated with Local Holdings + record; can be an integer, or string + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + url = self._url_search_lhr_control_number(controlNumber) + header = {"Accept": "application/json"} + + # prep request + req = Request("GET", url, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def get_lbd_record( + self, + controlNumber: Union[int, str], + response_format: Optional[str] = "application/marcxml+xml", + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Given a Control Number, retrieve a Local Bibliographic Data record. + Uses /manage/lbds/{controlNumber} endpoint. + + Args: + controlNumber: control number associated with Local Bibliographic + Data record; can be an integer, or string + response_format: format of returned record, options: + 'application/marcxml+xml', 'application/marc', + default is 'application/marcxml+xml' + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + url = self._url_manage_lbd(controlNumber) + header = {"Accept": response_format} + + # prep request + req = Request("GET", url, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def get_lhr_record( + self, + controlNumber: Union[int, str], + response_format: Optional[str] = "application/marcxml+xml", + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Send a GET request for a local holdings record + Uses /manage/lhrs/{controlNumber} endpoint. + + Args: + controlNumber: control number associated with Local Holdings + record; can be an integer, or string + response_format: format of returned record, options: + 'application/marcxml+xml', 'application/marc', + default is 'application/marcxml+xml' + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + url = self._url_manage_lhr(controlNumber) + header = {"Accept": response_format} + + # prep request + req = Request("GET", url, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def holding_set_on_record( + self, + record: Optional[str] = None, + recordFormat: Optional[str] = None, + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Given a MARC record in MARC XML or MARC21, set a holding on the record. + MARC record must contain OCLC number in 001 or 035 subfield a. + Only one MARC record is allowed in the request body. + Uses /manage/institution/holdings/set endpoint. + + Args: + record: MARC record on which to set holdings + recordFormat: format of MARC record, options: + 'application/marcxml+xml', 'application/marc' + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + if not record: + raise TypeError("Argument 'record' is missing.") + if not recordFormat: + raise TypeError("Argument 'recordFormat' is missing.") + + url = self._url_manage_ih_set_on_record() + header = { + "Accept": "application/json", + "content-type": recordFormat, + } + + # prep request + req = Request("POST", url, data=record, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def holding_unset_on_record( + self, + record: Optional[str] = None, + recordFormat: Optional[str] = None, + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Given a MARC record in MARC XML or MARC21, unset a holding on the record. + MARC record must contain OCLC number in 001 or 035 subfield a. + Only one MARC record is allowed in the request body. + Uses /manage/institution/holdings/unset endpoint. + + Args: + record: MARC record on which to unset holdings + recordFormat: format of MARC record, options: + 'application/marcxml+xml', 'application/marc' + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + if not record: + raise TypeError("Argument 'record' is missing.") + if not recordFormat: + raise TypeError("Argument 'recordFormat' is missing.") + + url = self._url_manage_ih_unset_on_record() + header = { + "Accept": "application/json", + "content-type": recordFormat, + } + + # prep request + req = Request("POST", url, data=record, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + return query.response + + def match_bib( + self, + record: Optional[str] = None, + recordFormat: Optional[str] = None, + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Given a bib record in MARC21 or MARCXML identify the best match in WorldCat. + Record must contain at minimum an 008 and 245. Response contains number of + potential matches in numberOfRecords and best match in briefRecords + Uses /manage/bibs/match endpoint. + + Args: + record: MARC record to be matched + recordFormat: format of MARC record, options: + 'application/marcxml+xml', 'application/marc' + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` instance + """ + if not record: + raise TypeError("Argument 'record' is missing.") + if not recordFormat: + raise TypeError("Argument 'recordFormat' is missing.") + + url = self._url_manage_bibs_match() + header = { + "Accept": "application/json", + "content-type": recordFormat, + } + + # prep request + req = Request("POST", url, data=record, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def replace_bib( + self, + oclcNumber: Union[int, str], + record: Optional[str] = None, + recordFormat: Optional[str] = None, + responseFormat: Optional[str] = "application/marcxml+xml", + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Given an OCLC number and MARC record, find record in WorldCat and replace it. + If the record does not exist in WorldCat, a new bib record will be created. + Uses /manage/bibs/{oclcNumber} endpoint. + + Args: + oclcNumber: OCLC bibliographic record number for record to be + replaced; can be an integer or string with or + without OCLC # prefix + record: MARC record to replace existing WorldCat record + recordFormat: format of MARC record, options: + 'application/marcxml+xml', 'application/marc' + responseFormat: format of returned record; options: + 'application/marcxml+xml', 'application/marc' + default is 'application/marcxml+xml' + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` instance + """ + oclcNumber = verify_oclc_number(oclcNumber) + + if not record: + raise TypeError("Argument 'record' is missing.") + if not recordFormat: + raise TypeError("Argument 'recordFormat' is missing.") + + url = self._url_manage_bibs(oclcNumber) + header = { + "Accept": responseFormat, + "content-type": recordFormat, + } + + # prep request + req = Request("PUT", url, data=record, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def replace_lbd( + self, + controlNumber: Union[int, str], + record: Optional[str] = None, + recordFormat: Optional[str] = None, + responseFormat: Optional[str] = "application/marcxml+xml", + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Given a Control Number, find the associated Local Bibliographic Data + Record and replace it. If the Control Number is not found in + WorldCat, then the provided Local Bibliographic Data Record will be created. + Uses /manage/lbds/{controlNumber} endpoint. + + Args: + controlNumber: control number associated with Local Bibliographic + Data record; can be an integer, or string + response_format: format of returned record, options: + 'application/marcxml+xml', 'application/marc', + default is 'application/marcxml+xml' + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + if not record: + raise TypeError("Argument 'record' is missing.") + if not recordFormat: + raise TypeError("Argument 'recordFormat' is missing.") + + url = self._url_manage_lbd(controlNumber) + header = { + "Accept": responseFormat, + "content-type": recordFormat, + } + + # prep request + req = Request("PUT", url, data=record, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def replace_lhr( + self, + controlNumber: Union[int, str], + record: Optional[str] = None, + recordFormat: Optional[str] = None, + responseFormat: Optional[str] = "application/marcxml+xml", + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Given a Control Number, find the associated Local Holdings + Record and replace it. If the Control Number is not found in + WorldCat, then the provided Local Holdings Record will be created. + Uses /manage/lhrs/{controlNumber} endpoint. + + Args: + controlNumber: control number associated with Local Holdings + record; can be an integer, or string + response_format: format of returned record, options: + 'application/marcxml+xml', 'application/marc', + default is 'application/marcxml+xml' + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + if not record: + raise TypeError("Argument 'record' is missing.") + if not recordFormat: + raise TypeError("Argument 'recordFormat' is missing.") + + url = self._url_manage_lhr(controlNumber) + header = { + "Accept": responseFormat, + "content-type": recordFormat, + } + + # prep request + req = Request("PUT", url, data=record, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def search_my_holdings( + self, + oclcNumber: Optional[Union[int, str]] = None, + barcode: Optional[str] = None, + orderBy: Optional[str] = None, + offset: Optional[int] = None, + limit: Optional[int] = None, + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Search LHR Resources + Uses /search/my-holdings endpoint. + + Args: + oclcNumber: OCLC bibliographic record number; can be + an integer, or string that can include + OCLC # prefix + barcode: barcode as a string, + orderBy: results sort key; + options: + 'commitmentExpirationDate' + 'location' + 'oclcSymbol' + default is 'oclcSymbol' + offset: start position of bibliographic records to + return; default is 1 + limit: maximum number of records to return; + maximum is 50, default is 10 + hooks: Requests library hook system that can be used for + signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + if oclcNumber is not None: + oclcNumber = verify_oclc_number(oclcNumber) + + url = self._url_search_lhr() + header = {"Accept": "application/json"} + payload = { + "oclcNumber": oclcNumber, + "barcode": barcode, + "orderBy": orderBy, + "offset": offset, + "limit": limit, + } + + # prep request + req = Request("GET", url, params=payload, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def search_my_local_bibs( + self, + q: str, + offset: Optional[int] = None, + limit: Optional[int] = None, + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Search LBD Resources + Uses /search/my-local-bib-data endpoint. + + Args: + q: query in the form of a keyword search or + fielded search; + examples: + ti:Zendegi + ti:"Czarne oceany" + bn:9781680502404 + kw:python databases + ti:Zendegi AND au:greg egan + (au:Okken OR au:Myers) AND su:python + offset: start position of bibliographic records to + return; default is 1 + limit: maximum number of records to return; + maximum is 50, default is 10 + hooks: Requests library hook system that can be used for + signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + if not q: + raise TypeError("Argument 'q' is requried to construct query.") + + url = self._url_search_lbd() + header = {"Accept": "application/json"} + payload = {"q": q, "offset": offset, "limit": limit} + + # prep request + req = Request("GET", url, params=payload, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def search_shared_print_lhr( + self, + oclcNumber: Optional[Union[int, str]] = None, + barcode: Optional[str] = None, + heldBySymbol: Optional[List[str]] = None, + heldByInstitutionID: Optional[List[int]] = None, + spProgram: Optional[List[str]] = None, + orderBy: Optional[str] = None, + offset: Optional[int] = None, + limit: Optional[int] = None, + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Search for shared print LHR Resources + Uses /search/retained-holdings endpoint. + + Args: + oclcNumber: OCLC bibliographic record number; can be + an integer, or string that can include + OCLC # prefix + barcode: barcode as a string, + heldBySymbol: restricts to holdings with specified intitution + symbol + heldByInstitutionID: restrict to specified institution registryId + spProgram: restricts responses to bibliographic records + associated with particular shared print + program + orderBy: results sort key; + options: + 'commitmentExpirationDate' + 'location' + 'oclcSymbol' + default is 'oclcSymbol' + offset: start position of bibliographic records to + return; default is 1 + limit: maximum number of records to return; + maximum is 50, default is 10 + hooks: Requests library hook system that can be used for + signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` object + """ + if oclcNumber is not None: + oclcNumber = verify_oclc_number(oclcNumber) + + url = self._url_search_lhr_shared_print() + header = {"Accept": "application/json"} + payload = { + "oclcNumber": oclcNumber, + "barcode": barcode, + "heldBySymbol": heldBySymbol, + "heldByInstitutionID": heldByInstitutionID, + "spProgram": spProgram, + "orderBy": orderBy, + "offset": offset, + "limit": limit, + } + + # prep request + req = Request("GET", url, params=payload, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response + + def validate_bib( + self, + record: Optional[str] = None, + recordFormat: Optional[str] = None, + validationLevel: str = "validateFull", + hooks: Optional[Dict[str, Callable]] = None, + ) -> Optional[Response]: + """ + Given a bib record, validate that record conforms to MARC standards + Uses /manage/bibs/validate/{validationLevel} endpoint. + + Args: + record: MARC record to be validated + recordFormat: format of MARC record, options: + 'application/marcxml+xml', 'application/marc' + validationLevel: Level at which to validate records + available values: 'validateFull', 'validateAdd', + 'validateReplace' + default is 'validateFull' + hooks: Requests library hook system that can be + used for signal event handling, see more at: + https://requests.readthedocs.io/en/master/user/advanced/#event-hooks + Returns: + `requests.Response` instance + """ + if not record: + raise TypeError("Argument 'record' is missing.") + + if not recordFormat: + raise TypeError("Argument 'recordFormat' is missing.") + + url = self._url_manage_bibs_validate(validationLevel) + header = { + "Accept": "application/json", + "content-type": recordFormat, + } + + # prep request + req = Request( + "POST", + url, + data=record, + headers=header, + hooks=hooks, + ) + prepared_request = self.prepare_request(req) + + # send request + query = Query(self, prepared_request, timeout=self.timeout) + + return query.response diff --git a/tests/conftest.py b/tests/conftest.py index 72522d3..9225ace 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -23,6 +23,28 @@ def live_keys(): os.environ["WCScopes"] = data["scopes"] +@pytest.fixture +def stub_marc_xml(): + stub_marc_xml = "00000nam a2200000 a 4500120827s2012 nyua 000 0 eng d 63011276 ocn850940548OCWMSengOCWMSOCLC Developer NetworkTest RecordFOR OCLC DEVELOPER NETWORK DOCUMENTATION" + return stub_marc_xml + + +@pytest.fixture +def stub_holding_xml(): + stub_holding_xml = "00000nx a2200000zi 4500312010zu1103280p 0 4001uueng0210908OCWMSEASTEAST-STACKS879456" + return stub_holding_xml + + +@pytest.fixture +def stub_marc21(): + fh = os.path.join( + os.environ["USERPROFILE"], "github/bookops-worldcat/temp/test.mrc" + ) + with open(fh, "rb") as stub: + stub_marc21 = stub.read() + return stub_marc21 + + class FakeUtcNow(datetime.datetime): @classmethod def now(cls, tzinfo=datetime.timezone.utc): diff --git a/tests/test_metadata_api.py b/tests/test_metadata_api.py index a3d2236..e615eba 100644 --- a/tests/test_metadata_api.py +++ b/tests/test_metadata_api.py @@ -149,6 +149,150 @@ def test_url_manage_ih_current(self, stub_session): == "https://metadata.api.oclc.org/worldcat/manage/institution/holdings/current" ) + @pytest.mark.parametrize( + "validationLevel", + ["vaidateFull", "validateAdd", "validateReplace"], + ) + def test_url_manage_bibs_validate(self, validationLevel, stub_session): + assert ( + stub_session._url_manage_bibs_validate(validationLevel) + == f"https://metadata.api.oclc.org/worldcat/manage/bibs/validate/{validationLevel}" + ) + + def test_url_manage_bibs_current_oclc_number(self, stub_session): + assert ( + stub_session._url_manage_bibs_current_oclc_number() + == "https://metadata.api.oclc.org/worldcat/manage/bibs/current" + ) + + def test_url_manage_bibs_create(self, stub_session): + assert ( + stub_session._url_manage_bibs_create() + == "https://metadata.api.oclc.org/worldcat/manage/bibs" + ) + + def test_url_manage_bibs(self, stub_session): + assert ( + stub_session._url_manage_bibs(oclcNumber="12345") + == "https://metadata.api.oclc.org/worldcat/manage/bibs/12345" + ) + + def test_url_manage_bibs_match(self, stub_session): + assert ( + stub_session._url_manage_bibs_match() + == "https://metadata.api.oclc.org/worldcat/manage/bibs/match" + ) + + def test_url_manage_ih_set_on_record(self, stub_session): + assert ( + stub_session._url_manage_ih_set_on_record() + == "https://metadata.api.oclc.org/worldcat/manage/institution/holdings/set" + ) + + def test_url_manage_ih_unset_on_record(self, stub_session): + assert ( + stub_session._url_manage_ih_unset_on_record() + == "https://metadata.api.oclc.org/worldcat/manage/institution/holdings/unset" + ) + + def test_url_manage_ih_codes(self, stub_session): + assert ( + stub_session._url_manage_ih_codes() + == "https://metadata.api.oclc.org/worldcat/manage/institution/holding-codes" + ) + + def test_url_manage_lbd_create(self, stub_session): + assert ( + stub_session._url_manage_lbd_create() + == "https://metadata.api.oclc.org/worldcat/manage/lbds" + ) + + @pytest.mark.parametrize( + "controlNumber", + ["12345", 12345], + ) + def test_url_manage_lbd(self, controlNumber, stub_session): + assert ( + stub_session._url_manage_lbd(controlNumber) + == f"https://metadata.api.oclc.org/worldcat/manage/lbds/{controlNumber}" + ) + + def test_url_manage_lhr_create(self, stub_session): + assert ( + stub_session._url_manage_lhr_create() + == "https://metadata.api.oclc.org/worldcat/manage/lhrs" + ) + + @pytest.mark.parametrize( + "controlNumber", + ["12345", 12345], + ) + def test_url_manage_lhr(self, controlNumber, stub_session): + assert ( + stub_session._url_manage_lhr(controlNumber) + == f"https://metadata.api.oclc.org/worldcat/manage/lhrs/{controlNumber}" + ) + + def test_url_search_general_holdings_summary(self, stub_session): + assert ( + stub_session._url_search_general_holdings_summary() + == "https://metadata.api.oclc.org/worldcat/search/summary-holdings" + ) + + @pytest.mark.parametrize( + "oclcNumber", + ["850940461", "850940463", 850940467], + ) + def test_url_search_classification_bibs(self, oclcNumber, stub_session): + assert ( + stub_session._url_search_classification_bibs(oclcNumber) + == f"https://metadata.api.oclc.org/worldcat/search/classification-bibs/{oclcNumber}" + ) + + def test_url_search_lhr_shared_print(self, stub_session): + assert ( + stub_session._url_search_lhr_shared_print() + == "https://metadata.api.oclc.org/worldcat/search/retained-holdings" + ) + + @pytest.mark.parametrize( + "controlNumber", + ["12345", 12345], + ) + def test_url_search_lhr_control_number(self, controlNumber, stub_session): + assert ( + stub_session._url_search_lhr_control_number(controlNumber) + == f"https://metadata.api.oclc.org/worldcat/search/my-holdings/{controlNumber}" + ) + + def test_url_search_lhr(self, stub_session): + assert ( + stub_session._url_search_lhr() + == "https://metadata.api.oclc.org/worldcat/search/my-holdings" + ) + + def test_url_browse_lhr(self, stub_session): + assert ( + stub_session._url_browse_lhr() + == "https://metadata.api.oclc.org/worldcat/browse/my-holdings" + ) + + @pytest.mark.parametrize( + "controlNumber", + ["12345", 12345], + ) + def test_url_search_lbd_control_number(self, controlNumber, stub_session): + assert ( + stub_session._url_search_lbd_control_number(controlNumber) + == f"https://metadata.api.oclc.org/worldcat/search/my-local-bib-data/{controlNumber}" + ) + + def test_url_search_lbd(self, stub_session): + assert ( + stub_session._url_search_lbd() + == "https://metadata.api.oclc.org/worldcat/search/my-local-bib-data" + ) + @pytest.mark.http_code(200) def test_get_brief_bib(self, stub_session, mock_session_response): assert stub_session.get_brief_bib(12345).status_code == 200 @@ -311,6 +455,357 @@ def test_search_shared_print_holdings_with_invalid_oclc_number_passsed( stub_session.search_shared_print_holdings(oclcNumber="odn12345") assert msg in str(exc.value) + @pytest.mark.http_code(200) + def test_browse_my_holdings(self, stub_session, mock_session_response): + assert ( + stub_session.browse_my_holdings( + oclcNumber="12345", holdingLocation="foo", shelvingLocation="bar" + ).status_code + == 200 + ) + + @pytest.mark.http_code(200) + def test_browse_my_holdings_no_oclc_number( + self, stub_session, mock_session_response + ): + assert ( + stub_session.browse_my_holdings( + holdingLocation="foo", shelvingLocation="bar" + ).status_code + == 200 + ) + + def test_browse_my_holdings_no_holdingLocation_passed(self, stub_session): + with pytest.raises(TypeError) as exc: + stub_session.browse_my_holdings(shelvingLocation="bar") + assert "Argument 'holdingLocation' is missing." in str(exc.value) + + def test_browse_my_holdings_no_shelvingLocation_passed(self, stub_session): + with pytest.raises(TypeError) as exc: + stub_session.browse_my_holdings(holdingLocation="foo") + assert "Argument 'shelvingLocation' is missing." in str(exc.value) + + @pytest.mark.http_code(200) + def test_create_bib(self, stub_session, mock_session_response, stub_marc_xml): + assert ( + stub_session.create_bib( + stub_marc_xml, recordFormat="application/marcxml+xml" + ).status_code + == 200 + ) + + def test_create_bib_no_record_passed(self, stub_session): + with pytest.raises(TypeError) as exc: + stub_session.create_bib(recordFormat="application/marcxml+xml") + assert "Argument 'record' is missing." in str(exc.value) + + def test_create_bib_no_recordFormat_passed( + self, stub_session, mock_session_response, stub_marc_xml + ): + with pytest.raises(TypeError) as exc: + stub_session.create_bib(stub_marc_xml) + assert "Argument 'recordFormat' is missing." in str(exc.value) + + @pytest.mark.http_code(200) + def test_create_lbd(self, stub_session, mock_session_response, stub_marc_xml): + assert ( + stub_session.create_lbd( + stub_marc_xml, recordFormat="application/marcxml+xml" + ).status_code + == 200 + ) + + def test_create_lbd_no_record_passed(self, stub_session): + with pytest.raises(TypeError) as exc: + stub_session.create_lbd(recordFormat="application/marcxml+xml") + assert "Argument 'record' is missing." in str(exc.value) + + def test_create_lbd_no_recordFormat_passed( + self, stub_session, mock_session_response, stub_marc_xml + ): + with pytest.raises(TypeError) as exc: + stub_session.create_lbd(stub_marc_xml) + assert "Argument 'recordFormat' is missing." in str(exc.value) + + @pytest.mark.http_code(200) + def test_create_lhr(self, stub_session, mock_session_response, stub_holding_xml): + assert ( + stub_session.create_lhr( + stub_holding_xml, recordFormat="application/marcxml+xml" + ).status_code + == 200 + ) + + def test_create_lhr_no_record_passed(self, stub_session): + with pytest.raises(TypeError) as exc: + stub_session.create_lhr(recordFormat="application/marcxml+xml") + assert "Argument 'record' is missing." in str(exc.value) + + def test_create_lhr_no_recordFormat_passed( + self, stub_session, mock_session_response, stub_holding_xml + ): + with pytest.raises(TypeError) as exc: + stub_session.create_lhr(stub_holding_xml) + assert "Argument 'recordFormat' is missing." in str(exc.value) + + @pytest.mark.http_code(200) + def test_delete_lbd(self, stub_session, mock_session_response): + assert stub_session.delete_lbd("12345").status_code == 200 + + @pytest.mark.http_code(200) + def test_delete_lhr(self, stub_session, mock_session_response): + assert stub_session.delete_lhr("12345").status_code == 200 + + @pytest.mark.http_code(200) + def test_get_bib_classification(self, stub_session, mock_session_response): + assert stub_session.get_bib_classification(12345).status_code == 200 + + def test_get_bib_classification_no_oclcNumber_passed(self, stub_session): + with pytest.raises(TypeError): + stub_session.get_bib_classification() + + @pytest.mark.http_code(200) + def test_get_bib_holdings(self, stub_session, mock_session_response): + assert stub_session.get_bib_holdings(oclcNumber=12345).status_code == 200 + + def test_get_bib_holdings_no_oclcNumber_passed(self, stub_session): + with pytest.raises(TypeError): + stub_session.get_bib_holdings(holdingsAllVariantRecords=True) + + @pytest.mark.http_code(200) + def test_get_lbd_data(self, stub_session, mock_session_response): + assert stub_session.get_lbd_data(12345).status_code == 200 + + def test_get_lbd_data_no_controlNumber_passed(self, stub_session): + with pytest.raises(TypeError): + stub_session.get_lbd_data() + + @pytest.mark.http_code(200) + def test_get_lhr_data(self, stub_session, mock_session_response): + assert stub_session.get_lhr_data(12345).status_code == 200 + + def test_get_lhr_data_no_controlNumber_passed(self, stub_session): + with pytest.raises(TypeError): + stub_session.get_lhr_data() + + @pytest.mark.http_code(200) + def test_get_lbd_record(self, stub_session, mock_session_response): + assert stub_session.get_lbd_record(12345).status_code == 200 + + def test_get_lbd_record_no_controlNumber_passed(self, stub_session): + with pytest.raises(TypeError): + stub_session.get_lbd_record() + + @pytest.mark.http_code(200) + def test_get_lhr_record(self, stub_session, mock_session_response): + assert stub_session.get_lhr_record(12345).status_code == 200 + + def test_get_lhr_record_no_controlNumber_passed(self, stub_session): + with pytest.raises(TypeError): + stub_session.get_lhr_record() + + @pytest.mark.http_code(200) + def test_holding_set_on_record( + self, stub_session, mock_session_response, stub_marc_xml + ): + assert ( + stub_session.holding_set_on_record( + stub_marc_xml, recordFormat="application/marcxml+xml" + ).status_code + == 200 + ) + + def test_holding_set_on_record_no_record_passed(self, stub_session): + with pytest.raises(TypeError) as exc: + stub_session.holding_set_on_record(recordFormat="application/marcxml+xml") + assert "Argument 'record' is missing." in str(exc.value) + + def test_holding_set_on_record_no_recordFormat_passed( + self, stub_session, mock_session_response, stub_marc_xml + ): + with pytest.raises(TypeError) as exc: + stub_session.holding_set_on_record(stub_marc_xml) + assert "Argument 'recordFormat' is missing." in str(exc.value) + + @pytest.mark.http_code(200) + def test_holding_unset_on_record( + self, stub_session, mock_session_response, stub_marc_xml + ): + assert ( + stub_session.holding_unset_on_record( + record=stub_marc_xml, recordFormat="application/marcxml+xml" + ).status_code + == 200 + ) + + def test_holding_unset_on_record_no_record_passed(self, stub_session): + with pytest.raises(TypeError) as exc: + stub_session.holding_unset_on_record(recordFormat="application/marcxml+xml") + assert "Argument 'record' is missing." in str(exc.value) + + def test_holding_unset_on_record_no_recordFormat_passed( + self, stub_session, mock_session_response, stub_marc_xml + ): + with pytest.raises(TypeError) as exc: + stub_session.holding_unset_on_record(stub_marc_xml) + assert "Argument 'recordFormat' is missing." in str(exc.value) + + @pytest.mark.http_code(200) + def test_match_bib(self, stub_session, mock_session_response, stub_marc_xml): + assert ( + stub_session.match_bib( + stub_marc_xml, recordFormat="application/marcxml+xml" + ).status_code + == 200 + ) + + def test_match_bib_no_record_passed(self, stub_session): + with pytest.raises(TypeError) as exc: + stub_session.match_bib(recordFormat="application/marcxml+xml") + assert "Argument 'record' is missing." in str(exc.value) + + def test_match_bib_no_recordFormat_passed( + self, stub_session, mock_session_response, stub_marc_xml + ): + with pytest.raises(TypeError) as exc: + stub_session.match_bib(stub_marc_xml) + assert "Argument 'recordFormat' is missing." in str(exc.value) + + @pytest.mark.http_code(200) + def test_replace_bib(self, stub_session, mock_session_response, stub_marc_xml): + assert ( + stub_session.replace_bib( + "12345", stub_marc_xml, recordFormat="application/marcxml+xml" + ).status_code + == 200 + ) + + @pytest.mark.http_code(200) + def test_replace_bib_no_record_passed(self, stub_session): + with pytest.raises(TypeError) as exc: + stub_session.replace_bib("12345", recordFormat="application/marcxml+xml") + assert "Argument 'record' is missing." in str(exc.value) + + @pytest.mark.http_code(200) + def test_replace_bib_no_recordFormat_passed( + self, stub_session, mock_session_response, stub_marc_xml + ): + with pytest.raises(TypeError) as exc: + stub_session.replace_bib("12345", stub_marc_xml) + assert "Argument 'recordFormat' is missing." in str(exc.value) + + @pytest.mark.http_code(200) + def test_replace_lbd(self, stub_session, mock_session_response, stub_marc_xml): + assert ( + stub_session.replace_lbd( + "12345", stub_marc_xml, recordFormat="application/marcxml+xml" + ).status_code + == 200 + ) + + @pytest.mark.http_code(200) + def test_replace_lbd_no_record_passed(self, stub_session): + with pytest.raises(TypeError) as exc: + stub_session.replace_lbd("12345", recordFormat="application/marcxml+xml") + assert "Argument 'record' is missing." in str(exc.value) + + @pytest.mark.http_code(200) + def test_replace_lbd_no_recordFormat_passed( + self, stub_session, mock_session_response, stub_marc_xml + ): + with pytest.raises(TypeError) as exc: + stub_session.replace_lbd("12345", stub_marc_xml) + assert "Argument 'recordFormat' is missing." in str(exc.value) + + @pytest.mark.http_code(200) + def test_replace_lhr(self, stub_session, mock_session_response, stub_holding_xml): + assert ( + stub_session.replace_lhr( + "12345", stub_holding_xml, recordFormat="application/marcxml+xml" + ).status_code + == 200 + ) + + @pytest.mark.http_code(200) + def test_replace_lhr_no_record_passed(self, stub_session): + with pytest.raises(TypeError) as exc: + stub_session.replace_lhr("12345", recordFormat="application/marcxml+xml") + assert "Argument 'record' is missing." in str(exc.value) + + @pytest.mark.http_code(200) + def test_replace_lhr_no_recordFormat_passed( + self, stub_session, mock_session_response, stub_holding_xml + ): + with pytest.raises(TypeError) as exc: + stub_session.replace_lhr("12345", stub_holding_xml) + assert "Argument 'recordFormat' is missing." in str(exc.value) + + @pytest.mark.http_code(200) + def test_search_my_holdings(self, stub_session, mock_session_response): + assert stub_session.search_my_holdings(oclcNumber=12345).status_code == 200 + + def test_search_my_holdings_invalid_oclc_number(self, stub_session): + msg = "Argument 'oclcNumber' does not look like real OCLC #." + with pytest.raises(InvalidOclcNumber) as exc: + stub_session.search_my_holdings(oclcNumber="odn12345") + assert msg in str(exc.value) + + @pytest.mark.http_code(200) + def test_search_my_local_bibs(self, stub_session, mock_session_response): + assert stub_session.search_my_local_bibs(q="ti:foo").status_code == 200 + + @pytest.mark.parametrize("argm", [(None), ("")]) + def test_search_my_local_bibs_missing_query(self, stub_session, argm): + with pytest.raises(TypeError) as exc: + stub_session.search_my_local_bibs(argm) + assert "Argument 'q' is requried to construct query." in str(exc.value) + + @pytest.mark.http_code(200) + def test_search_shared_print_lhr(self, stub_session, mock_session_response): + assert stub_session.search_shared_print_lhr(oclcNumber=12345).status_code == 200 + + def test_search_shared_print_lhr_with_invalid_oclc_number_passsed( + self, stub_session + ): + msg = "Argument 'oclcNumber' does not look like real OCLC #." + with pytest.raises(InvalidOclcNumber) as exc: + stub_session.search_shared_print_lhr(oclcNumber="odn12345") + assert msg in str(exc.value) + + @pytest.mark.http_code(200) + def test_validate_bib(self, stub_session, mock_session_response, stub_marc_xml): + assert ( + stub_session.validate_bib( + stub_marc_xml, + recordFormat="application/marcxml+xml", + validationLevel="validateFull", + ).status_code + == 200 + ) + + @pytest.mark.http_code(200) + def test_validate_bib_default( + self, stub_session, mock_session_response, stub_marc_xml + ): + assert ( + stub_session.validate_bib( + stub_marc_xml, recordFormat="application/marcxml+xml" + ).status_code + == 200 + ) + + def test_validate_bib_no_record_passed(self, stub_session): + with pytest.raises(TypeError) as exc: + stub_session.validate_bib(recordFormat="application/marcxml+xml") + assert "Argument 'record' is missing." in str(exc.value) + + def test_validate_bib_no_recordFormat_passed( + self, stub_session, mock_session_response, stub_marc_xml + ): + with pytest.raises(TypeError) as exc: + stub_session.validate_bib(stub_marc_xml) + assert "Argument 'recordFormat' is missing." in str(exc.value) + @pytest.mark.webtest class TestLiveMetadataSession: @@ -509,3 +1004,178 @@ def test_get_current_oclc_number(self, live_keys): jres = response.json() assert sorted(jres.keys()) == ["controlNumbers"] assert sorted(jres["controlNumbers"][0].keys()) == ["current", "requested"] + + def test_get_bib_classification(self, live_keys): + token = WorldcatAccessToken( + key=os.getenv("WCKey"), + secret=os.getenv("WCSecret"), + scopes=os.getenv("WCScopes"), + ) + + with MetadataSession(authorization=token) as session: + response = session.get_bib_classification(41266045) + + assert ( + response.url + == "https://metadata.api.oclc.org/worldcat/search/classification-bibs/41266045" + ) + assert response.status_code == 200 + assert sorted(response.json().keys()) == [ + "dewey", + "lc", + ] + + def test_get_current_oclc_number_str(self, live_keys): + token = WorldcatAccessToken( + key=os.getenv("WCKey"), + secret=os.getenv("WCSecret"), + scopes=os.getenv("WCScopes"), + ) + + with MetadataSession(authorization=token) as session: + response = session.get_current_oclc_number("41266045") + + assert response.status_code == 200 + assert ( + response.request.url + == "https://metadata.api.oclc.org/worldcat/manage/bibs/current?oclcNumbers=41266045" + ) + jres = response.json() + assert sorted(jres.keys()) == ["controlNumbers"] + assert sorted(jres["controlNumbers"][0].keys()) == ["current", "requested"] + + @pytest.mark.holdings + def test_get_institution_holding_codes(self, live_keys): + token = WorldcatAccessToken( + key=os.getenv("WCKey"), + secret=os.getenv("WCSecret"), + scopes=os.getenv("WCScopes"), + ) + + with MetadataSession(authorization=token) as session: + response = session.get_institution_holding_codes() + + assert ( + response.url + == "https://metadata.api.oclc.org/worldcat/manage/institution/holding-codes" + ) + assert response.status_code == 200 + assert sorted(response.json().keys()) == ["holdingLibraryCodes"] + assert {"code": "Print Collection", "name": "NYPC"} in response.json()[ + "holdingLibraryCodes" + ] + + @pytest.mark.holdings + def test_holding_set_unset_marcxml(self, live_keys, stub_marc_xml): + token = WorldcatAccessToken( + key=os.getenv("WCKey"), + secret=os.getenv("WCSecret"), + scopes=os.getenv("WCScopes"), + ) + + with MetadataSession(authorization=token) as session: + response = session.get_institution_holdings("850940548")[0] + holdings = response.json()["holdings"] + + # make sure no holdings are set initially + if len(holdings) > 0: + response = session.holding_unset_on_record( + stub_marc_xml, recordFormat="application/marcxml+xml" + ) + + response = session.holding_set_on_record( + stub_marc_xml, recordFormat="application/marcxml+xml" + ) + assert ( + response.url + == "https://metadata.api.oclc.org/worldcat/manage/institution/holdings/set" + ) + assert response.status_code == 200 + assert response.json()["action"] == "Set Holdings" + + response = session.holding_unset_on_record( + stub_marc_xml, recordFormat="application/marcxml+xml" + ) + assert response.status_code == 200 + assert ( + response.request.url + == "https://metadata.api.oclc.org/worldcat/manage/institution/holdings/unset" + ) + assert response.json()["action"] == "Unset Holdings" + + def test_match_bib_marcxml(self, live_keys, stub_marc_xml): + token = WorldcatAccessToken( + key=os.getenv("WCKey"), + secret=os.getenv("WCSecret"), + scopes=os.getenv("WCScopes"), + ) + + with MetadataSession(authorization=token) as session: + response = session.match_bib( + stub_marc_xml, recordFormat="application/marcxml+xml" + ) + assert response.status_code == 200 + assert sorted(response.json().keys()) == sorted( + ["numberOfRecords", "briefRecords"] + ) + + @pytest.mark.holdings + def test_search_bibs_holdings_oclc(self, live_keys): + fields = sorted(["briefRecords", "numberOfRecords"]) + token = WorldcatAccessToken( + key=os.getenv("WCKey"), + secret=os.getenv("WCSecret"), + scopes=os.getenv("WCScopes"), + ) + + with MetadataSession(authorization=token) as session: + response = session.search_bibs_holdings(oclcNumber="41266045") + + assert response.status_code == 200 + assert sorted(response.json().keys()) == fields + + def test_search_bibs_holdings_isbn(self, live_keys): + fields = sorted(["briefRecords", "numberOfRecords"]) + token = WorldcatAccessToken( + key=os.getenv("WCKey"), + secret=os.getenv("WCSecret"), + scopes=os.getenv("WCScopes"), + ) + + with MetadataSession(authorization=token) as session: + response = session.search_bibs_holdings(isbn="9781597801744") + + assert response.status_code == 200 + assert sorted(response.json().keys()) == fields + + def test_search_brief_bibs_other_editions(self, live_keys): + fields = sorted(["briefRecords", "numberOfRecords"]) + token = WorldcatAccessToken( + key=os.getenv("WCKey"), + secret=os.getenv("WCSecret"), + scopes=os.getenv("WCScopes"), + ) + + with MetadataSession(authorization=token) as session: + response = session.search_brief_bibs_other_editions(41266045) + + assert response.status_code == 200 + assert sorted(response.json().keys()) == fields + + def test_validate_bib(self, live_keys, stub_marc21): + token = WorldcatAccessToken( + key=os.getenv("WCKey"), + secret=os.getenv("WCSecret"), + scopes=os.getenv("WCScopes"), + ) + + with MetadataSession(authorization=token) as session: + response = session.validate_bib( + stub_marc21, recordFormat="application/marc" + ) + assert response.status_code == 200 + assert ( + response.url + == "https://metadata.api.oclc.org/worldcat/manage/bibs/validate/validateFull" + ) + assert sorted(response.json().keys()) == sorted(["httpStatus", "status"]) From 84a5a89724490523f70b760f5df6b3a295bfe008 Mon Sep 17 00:00:00 2001 From: Charlotte Kostelic Date: Tue, 27 Feb 2024 12:34:28 -0500 Subject: [PATCH 2/3] added tests --- tests/test_metadata_api.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/tests/test_metadata_api.py b/tests/test_metadata_api.py index e615eba..94afd95 100644 --- a/tests/test_metadata_api.py +++ b/tests/test_metadata_api.py @@ -744,6 +744,11 @@ def test_replace_lhr_no_recordFormat_passed( def test_search_my_holdings(self, stub_session, mock_session_response): assert stub_session.search_my_holdings(oclcNumber=12345).status_code == 200 + def test_search_my_holdings_no_oclcNumber_passed( + self, stub_session, mock_session_response + ): + assert stub_session.search_my_holdings(barcode=12345).status_code == 200 + def test_search_my_holdings_invalid_oclc_number(self, stub_session): msg = "Argument 'oclcNumber' does not look like real OCLC #." with pytest.raises(InvalidOclcNumber) as exc: @@ -764,6 +769,12 @@ def test_search_my_local_bibs_missing_query(self, stub_session, argm): def test_search_shared_print_lhr(self, stub_session, mock_session_response): assert stub_session.search_shared_print_lhr(oclcNumber=12345).status_code == 200 + @pytest.mark.http_code(200) + def test_search_shared_print_lhr_no_oclcNumber_passed( + self, stub_session, mock_session_response + ): + assert stub_session.search_shared_print_lhr(barcode=12345).status_code == 200 + def test_search_shared_print_lhr_with_invalid_oclc_number_passsed( self, stub_session ): From 8a91354d9f1f755f936b906d2919824e8ae8212d Mon Sep 17 00:00:00 2001 From: Charlotte Kostelic Date: Tue, 27 Feb 2024 12:57:56 -0500 Subject: [PATCH 3/3] added test --- tests/test_metadata_api.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/test_metadata_api.py b/tests/test_metadata_api.py index 94afd95..3dd52e4 100644 --- a/tests/test_metadata_api.py +++ b/tests/test_metadata_api.py @@ -335,6 +335,10 @@ def test_get_full_bib_None_oclcNumber_passed(self, stub_session): with pytest.raises(InvalidOclcNumber): stub_session.get_full_bib(oclcNumber=None) + @pytest.mark.http_code(200) + def test_get_institution_holding_codes(self, stub_session, mock_session_response): + assert stub_session.get_institution_holding_codes().status_code == 200 + @pytest.mark.http_code(200) def test_get_institution_holdings(self, stub_session, mock_session_response): assert stub_session.get_institution_holdings("12345")[0].status_code == 200