diff --git a/bookops_worldcat/metadata_api.py b/bookops_worldcat/metadata_api.py index 30b32f0..031d7a4 100644 --- a/bookops_worldcat/metadata_api.py +++ b/bookops_worldcat/metadata_api.py @@ -17,6 +17,8 @@ class MetadataSession(WorldcatSession): """OCLC Metadata API wrapper session. Inherits `requests.Session` methods""" + BASE_URL = "https://metadata.api.oclc.org/worldcat" + def __init__( self, authorization: WorldcatAccessToken, @@ -50,78 +52,35 @@ def _split_into_legal_volume( for i in range(0, len(oclc_numbers), n): yield ",".join(oclc_numbers[i : i + n]) # noqa: E203 - def URL_BASE(self) -> str: - return "https://metadata.api.oclc.org/worldcat" - - def _url_base(self) -> str: - return "https://worldcat.org" - - def _url_search_base(self) -> str: - return "https://americas.metadata.api.oclc.org/worldcat/search/v1" - def _url_search_shared_print_holdings(self) -> str: - base_url = self.URL_BASE() - return f"{base_url}/search/bibs-retained-holdings" + return f"{self.BASE_URL}/search/bibs-retained-holdings" def _url_search_general_holdings(self) -> str: - base_url = self.URL_BASE() - return f"{base_url}/search/bibs-summary-holdings" + return f"{self.BASE_URL}/search/bibs-summary-holdings" def _url_search_brief_bibs(self) -> str: - base_url = self.URL_BASE() - return f"{base_url}/search/brief-bibs" + return f"{self.BASE_URL}/search/brief-bibs" def _url_search_brief_bibs_oclc_number(self, oclcNumber: str) -> str: - base_url = self.URL_BASE() - return f"{base_url}/search/brief-bibs/{oclcNumber}" + return f"{self.BASE_URL}/search/brief-bibs/{oclcNumber}" def _url_search_brief_bibs_other_editions(self, oclcNumber: str) -> str: - base_url = self.URL_BASE() - return f"{base_url}/search/brief-bibs/{oclcNumber}/other-editions" - - def _url_lhr_control_number(self, controlNumber: str) -> str: - base_url = self._url_search_base() - return f"{base_url}/my-holdings/{controlNumber}" + return f"{self.BASE_URL}/search/brief-bibs/{oclcNumber}/other-editions" - def _url_lhr_search(self) -> str: - base_url = self._url_search_base() - return f"{base_url}/my-holdings" + def _url_manage_bibs(self, oclcNumber: str) -> str: + return f"{self.BASE_URL}/manage/bibs/{oclcNumber}" - def _url_lhr_shared_print(self) -> str: - base_url = self._url_search_base() - return f"{base_url}/retained-holdings" - - def _url_manage_bib(self, oclcNumber: str) -> str: - base_url = self.URL_BASE() - return f"{base_url}/manage/bibs/{oclcNumber}" - - def _url_manage_bib_current_oclc_number(self) -> str: - base_url = self.URL_BASE() - return f"{base_url}/manage/bibs/current" - - def _url_bib_holding_libraries(self) -> str: - base_url = self._url_base() - return f"{base_url}/bib/holdinglibraries" + def _url_manage_bibs_current_oclc_number(self) -> str: + return f"{self.BASE_URL}/manage/bibs/current" def _url_manage_ih_set(self, oclcNumber: str) -> str: - base_url = self.URL_BASE() - return f"{base_url}/manage/institution/holdings/{oclcNumber}/set" + return f"{self.BASE_URL}/manage/institution/holdings/{oclcNumber}/set" def _url_manage_ih_unset(self, oclcNumber: str) -> str: - base_url = self.URL_BASE() - return f"{base_url}/manage/institution/holdings/{oclcNumber}/unset" + return f"{self.BASE_URL}/manage/institution/holdings/{oclcNumber}/unset" def _url_manage_ih_current(self) -> str: - base_url = self.URL_BASE() - return f"{base_url}/manage/institution/holdings/current" - - def _url_bib_holdings_batch_action(self) -> str: - base_url = self._url_base() - return f"{base_url}/ih/datalist" - - def _url_bib_holdings_multi_institution_batch_action(self) -> str: - base_url = self._url_base() - return f"{base_url}/ih/institutionlist" + return f"{self.BASE_URL}/manage/institution/holdings/current" def get_brief_bib( self, oclcNumber: Union[int, str], hooks: Optional[Dict[str, Callable]] = None @@ -178,7 +137,7 @@ def get_full_bib( """ oclcNumber = verify_oclc_number(oclcNumber) - url = self._url_manage_bib(oclcNumber) + url = self._url_manage_bibs(oclcNumber) header = {"Accept": response_format} # prep request @@ -194,7 +153,7 @@ def get_institution_holdings( self, oclcNumbers: Union[str, List[Union[str, int]]], hooks: Optional[Dict[str, Callable]] = None, - ) -> Optional[Response]: + ) -> List[Optional[Response]]: """ Retrieves Worlcat holdings status of a record with provided OCLC number. The service automatically recognizes institution based on the issued access @@ -213,20 +172,24 @@ def get_institution_holdings( Returns: `requests.Response` object """ + responses = [] vetted_numbers = verify_oclc_numbers(oclcNumbers) url = self._url_manage_ih_current() header = {"Accept": "application/json"} - payload = {"oclcNumbers": ",".join(vetted_numbers)} - # prep request - req = Request("GET", url, params=payload, headers=header, hooks=hooks) - prepared_request = self.prepare_request(req) + for batch in self._split_into_legal_volume(oclc_numbers=vetted_numbers, n=10): + payload = {"oclcNumbers": batch} - # send request - query = Query(self, prepared_request, timeout=self.timeout) + # prep request + req = Request("GET", url, params=payload, headers=header, hooks=hooks) + prepared_request = self.prepare_request(req) - return query.response + # send request + query = Query(self, prepared_request, timeout=self.timeout) + responses.append(query.response) + + return responses def holding_set( self, @@ -295,216 +258,6 @@ def holding_unset( return query.response - def holdings_set( - self, - oclcNumbers: Union[str, List], - inst: Optional[str] = None, - instSymbol: Optional[str] = None, - response_format: str = "application/atom+json", - hooks: Optional[Dict[str, Callable]] = None, - ) -> List[Optional[Response]]: - """ - Set institution holdings for multiple OCLC numbers - Uses /ih/datalist endpoint. - - Args: - oclcNumbers: list of OCLC control numbers for which holdings - should be set; - they can be integers or strings with or - without OCLC # prefix; - if str the numbers must be separated by comma - inst: registry ID of the institution whose holdings - are being checked - instSymbol: optional; OCLC symbol of the institution whose - holdings are being checked - response_format: 'application/atom+json' (default) or - 'application/atom+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: - list of `requests.Response` objects - """ - responses = [] - vetted_numbers = verify_oclc_numbers(oclcNumbers) - - url = self._url_bib_holdings_batch_action() - header = {"Accept": response_format} - - # split into batches of 50 and issue request for each batch - for batch in self._split_into_legal_volume(vetted_numbers): - payload = { - "oclcNumbers": batch, - "inst": inst, - "instSymbol": instSymbol, - } - - # prep request - req = Request("POST", url, params=payload, headers=header, hooks=hooks) - prepared_request = self.prepare_request(req) - - # send request - query = Query(self, prepared_request, timeout=self.timeout) - - responses.append(query.response) - - return responses - - def holdings_unset( - self, - oclcNumbers: Union[str, List], - cascade: str = "0", - inst: Optional[str] = None, - instSymbol: Optional[str] = None, - response_format: str = "application/atom+json", - hooks: Optional[Dict[str, Callable]] = None, - ) -> List[Optional[Response]]: - """ - Set institution holdings for multiple OCLC numbers - Uses /ih/datalist endpoint. - - Args: - oclcNumbers: list of OCLC control numbers for which holdings - should be set; - they can be integers or strings with or - without OCLC # prefix; - if str the numbers must be separated by comma - cascade: 0 or 1, default 0; - 0 - don't remove holdings if local holding - record or local bibliographic records exists; - 1 - remove holding and delete local holdings - record and local bibliographic record - inst: registry ID of the institution whose holdings - are being checked - instSymbol: optional; OCLC symbol of the institution whose - holdings are being checked - response_format: 'application/atom+json' (default) or - 'application/atom+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: - list of `requests.Response` objects - """ - responses = [] - vetted_numbers = verify_oclc_numbers(oclcNumbers) - - url = self._url_bib_holdings_batch_action() - header = {"Accept": response_format} - - # split into batches of 50 and issue request for each batch - for batch in self._split_into_legal_volume(vetted_numbers): - payload = { - "oclcNumbers": batch, - "cascade": cascade, - "inst": inst, - "instSymbol": instSymbol, - } - - # prep request - req = Request("DELETE", url, params=payload, headers=header, hooks=hooks) - prepared_request = self.prepare_request(req) - - # send request - query = Query(self, prepared_request, timeout=self.timeout) - - responses.append(query.response) - - return responses - - def holdings_set_multi_institutions( - self, - oclcNumber: Union[int, str], - instSymbols: str, - response_format: str = "application/atom+json", - hooks: Optional[Dict[str, Callable]] = None, - ) -> Optional[Response]: - """ - Batch sets intitution holdings for multiple intitutions - - Uses /ih/institutionlist endpoint - - Args: - oclcNumber: OCLC bibliographic record number; can be an - integer, or string with or without OCLC # prefix - instSymbols: a comma-separated list of OCLC symbols of the - institution whose holdings are being set - response_format: 'application/atom+json' (default) or - 'application/atom+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 - """ - oclcNumber = verify_oclc_number(oclcNumber) - - url = self._url_bib_holdings_multi_institution_batch_action() - header = {"Accept": response_format} - payload = { - "oclcNumber": oclcNumber, - "instSymbols": instSymbols, - } - - # prep request - req = Request("POST", 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 holdings_unset_multi_institutions( - self, - oclcNumber: Union[int, str], - instSymbols: str, - cascade: str = "0", - response_format: str = "application/atom+json", - hooks: Optional[Dict[str, Callable]] = None, - ) -> Optional[Response]: - """ - Batch unsets intitution holdings for multiple intitutions - - Uses /ih/institutionlist endpoint - - Args: - oclcNumber: OCLC bibliographic record number; can be an - integer, or string with or without OCLC # prefix - instSymbols: a comma-separated list of OCLC symbols of the - institution whose holdings are being set - cascade: 0 or 1, default 0; - 0 - don't remove holdings if local holding - record or local bibliographic records exists; - 1 - remove holding and delete local holdings - record and local bibliographic record - response_format: 'application/atom+json' (default) or - 'application/atom+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 - """ - oclcNumber = verify_oclc_number(oclcNumber) - - url = self._url_bib_holdings_multi_institution_batch_action() - header = {"Accept": response_format} - payload = { - "oclcNumber": oclcNumber, - "instSymbols": instSymbols, - "cascade": cascade, - } - - # prep request - req = Request("DELETE", 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_brief_bibs_other_editions( self, oclcNumber: Union[int, str], @@ -856,7 +609,7 @@ def get_current_oclc_number( vetted_numbers = verify_oclc_numbers(oclcNumbers) header = {"Accept": "application/json"} - url = self._url_manage_bib_current_oclc_number() + url = self._url_manage_bibs_current_oclc_number() payload = {"oclcNumbers": ",".join(vetted_numbers)} # prep request diff --git a/bookops_worldcat/query.py b/bookops_worldcat/query.py index 1a99722..7dcc387 100644 --- a/bookops_worldcat/query.py +++ b/bookops_worldcat/query.py @@ -57,18 +57,7 @@ def __init__( try: self.response = session.send(prepared_request, timeout=timeout) - - if "/ih/data" in prepared_request.url: # type: ignore - if self.response.status_code == 409: - # HTTP 409 code returns when trying to set/unset - # holdings on already set/unset record - # It is reasonable not to raise any exceptions - # in this case - pass # pragma: no cover - else: - self.response.raise_for_status() - else: - self.response.raise_for_status() + self.response.raise_for_status() except HTTPError as exc: raise WorldcatRequestError( diff --git a/tests/test_metadata_api.py b/tests/test_metadata_api.py index 8626b4c..a3d2236 100644 --- a/tests/test_metadata_api.py +++ b/tests/test_metadata_api.py @@ -82,17 +82,8 @@ def test_split_into_legal_volume(self, stub_session, oclcNumbers, expectation): all_batches = [b for b in batches] assert all_batches == expectation - def test_new_url_base(self, stub_session): - assert stub_session.URL_BASE() == "https://metadata.api.oclc.org/worldcat" - def test_url_base(self, stub_session): - assert stub_session._url_base() == "https://worldcat.org" - - def test_url_search_base(self, stub_session): - assert ( - stub_session._url_search_base() - == "https://americas.metadata.api.oclc.org/worldcat/search/v1" - ) + assert stub_session.BASE_URL == "https://metadata.api.oclc.org/worldcat" def test_url_search_shared_print_holdings(self, stub_session): assert ( @@ -128,42 +119,18 @@ def test_url_search_brief_bibs_other_editions(self, stub_session): == "https://metadata.api.oclc.org/worldcat/search/brief-bibs/12345/other-editions" ) - def test_url_lhr_control_number(self, stub_session): - assert ( - stub_session._url_lhr_control_number(controlNumber="12345") - == "https://americas.metadata.api.oclc.org/worldcat/search/v1/my-holdings/12345" - ) - - def test_url_lhr_search(self, stub_session): - assert ( - stub_session._url_lhr_search() - == "https://americas.metadata.api.oclc.org/worldcat/search/v1/my-holdings" - ) - - def test_url_lhr_shared_print(self, stub_session): - assert ( - stub_session._url_lhr_shared_print() - == "https://americas.metadata.api.oclc.org/worldcat/search/v1/retained-holdings" - ) - def test_url_manage_bib(self, stub_session): assert ( - stub_session._url_manage_bib(oclcNumber="12345") + stub_session._url_manage_bibs(oclcNumber="12345") == "https://metadata.api.oclc.org/worldcat/manage/bibs/12345" ) def test_url_manage_bib_current_oclc_number(self, stub_session): assert ( - stub_session._url_manage_bib_current_oclc_number() + stub_session._url_manage_bibs_current_oclc_number() == "https://metadata.api.oclc.org/worldcat/manage/bibs/current" ) - def test_url_bib_holding_libraries(self, stub_session): - assert ( - stub_session._url_bib_holding_libraries() - == "https://worldcat.org/bib/holdinglibraries" - ) - def test_url_manage_ih_set(self, stub_session): assert ( stub_session._url_manage_ih_set(oclcNumber="12345") @@ -182,18 +149,6 @@ def test_url_manage_ih_current(self, stub_session): == "https://metadata.api.oclc.org/worldcat/manage/institution/holdings/current" ) - def test_url_bib_holdings_batch_action(self, stub_session): - assert ( - stub_session._url_bib_holdings_batch_action() - == "https://worldcat.org/ih/datalist" - ) - - def test_url_bib_holdings_multi_institution_batch_action(self, stub_session): - assert ( - stub_session._url_bib_holdings_multi_institution_batch_action() - == "https://worldcat.org/ih/institutionlist" - ) - @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 @@ -238,7 +193,7 @@ def test_get_full_bib_None_oclcNumber_passed(self, stub_session): @pytest.mark.http_code(200) def test_get_institution_holdings(self, stub_session, mock_session_response): - assert stub_session.get_institution_holdings("12345").status_code == 200 + assert stub_session.get_institution_holdings("12345")[0].status_code == 200 def test_get_institution_holdings_no_oclcNumber_passed(self, stub_session): with pytest.raises(TypeError): @@ -272,159 +227,6 @@ def test_holding_unset_None_oclcNumber_passed(self, stub_session): with pytest.raises(InvalidOclcNumber): stub_session.holding_unset(oclcNumber=None) - @pytest.mark.parametrize( - "argm,expectation", - [ - (None, pytest.raises(InvalidOclcNumber)), - ([], pytest.raises(InvalidOclcNumber)), - (["bt2111111111"], pytest.raises(InvalidOclcNumber)), - (["850940548"], does_not_raise()), - (["ocn850940548"], does_not_raise()), - ("850940548,850940552, 850940554", does_not_raise()), - (["850940548", "850940552", "850940554"], does_not_raise()), - ([850940548, 850940552, 850940554], does_not_raise()), - ], - ) - @pytest.mark.http_code(207) - def test_holdings_set(self, argm, expectation, stub_session, mock_session_response): - with expectation: - stub_session.holdings_set(argm) - - def test_holdings_set_no_oclcNumber_passed(self, stub_session): - with pytest.raises(TypeError): - stub_session.holdings_set() - - @pytest.mark.parametrize( - "argm,expectation", - [ - (None, pytest.raises(InvalidOclcNumber)), - ([], pytest.raises(InvalidOclcNumber)), - (["bt2111111111"], pytest.raises(InvalidOclcNumber)), - (["850940548"], does_not_raise()), - (["ocn850940548"], does_not_raise()), - ("850940548,850940552, 850940554", does_not_raise()), - (["850940548", "850940552", "850940554"], does_not_raise()), - ([850940548, 850940552, 850940554], does_not_raise()), - ], - ) - @pytest.mark.http_code(207) - def test_holdings_unset( - self, argm, expectation, stub_session, mock_session_response - ): - with expectation: - stub_session.holdings_unset(argm) - - def test_holdings_unset_no_oclcNumber_passed(self, stub_session): - with pytest.raises(TypeError): - stub_session.holdings_unset() - - @pytest.mark.http_code(207) - def test_holdings_unset_stale_token( - self, mock_now, stub_session, mock_session_response - ): - stub_session.authorization.token_expires_at = datetime.datetime.now( - datetime.timezone.utc - ) - datetime.timedelta(0, 1) - with does_not_raise(): - assert stub_session.authorization.is_expired() is True - stub_session.holdings_unset([850940548, 850940552, 850940554]) - assert stub_session.authorization.token_expires_at == datetime.datetime( - 2020, 1, 1, 17, 19, 58, tzinfo=datetime.timezone.utc - ) - assert stub_session.authorization.is_expired() is False - - @pytest.mark.http_code(200) - def test_holdings_set_multi_institutions(self, stub_session, mock_session_response): - results = stub_session.holdings_set_multi_institutions( - oclcNumber=850940548, instSymbols="BKL,NYP" - ) - assert results.status_code == 200 - - def test_holdings_set_multi_institutions_missing_oclc_number(self, stub_session): - with pytest.raises(TypeError): - stub_session.holdings_set_multi_institutions(instSymbols="NYP,BKL") - - def test_holdings_set_multi_institutions_missing_inst_symbols(self, stub_session): - with pytest.raises(TypeError): - stub_session.holdings_set_multi_institutions(oclcNumber=123) - - def test_holdings_set_multi_institutions_invalid_oclc_number(self, stub_session): - with pytest.raises(InvalidOclcNumber): - stub_session.holdings_set_multi_institutions( - oclcNumber="odn1234", instSymbols="NYP,BKL" - ) - - @pytest.mark.http_code(200) - def test_holdings_set_multi_institutions_stale_token( - self, mock_now, stub_session, mock_session_response - ): - stub_session.authorization.token_expires_at = datetime.datetime.now( - datetime.timezone.utc - ) - datetime.timedelta(0, 1) - with does_not_raise(): - assert stub_session.authorization.is_expired() is True - stub_session.holdings_set_multi_institutions( - oclcNumber=850940548, instSymbols="NYP,BKL" - ) - assert stub_session.authorization.token_expires_at == datetime.datetime( - 2020, 1, 1, 17, 19, 58, tzinfo=datetime.timezone.utc - ) - assert stub_session.authorization.is_expired() is False - - @pytest.mark.http_code(403) - def test_holdings_set_multi_institutions_permission_error( - self, stub_session, mock_session_response - ): - with pytest.raises(WorldcatRequestError) as exc: - stub_session.holdings_set_multi_institutions( - oclcNumber=850940548, instSymbols="NYP,BKL" - ) - - assert ( - "403 Client Error: 'foo' for url: https://foo.bar?query. Server response: spam" - in str(exc.value) - ) - - @pytest.mark.http_code(200) - def test_holdings_unset_multi_institutions( - self, stub_session, mock_session_response - ): - results = stub_session.holdings_unset_multi_institutions( - 850940548, "BKL,NYP", cascade="1" - ) - assert results.status_code == 200 - - def test_holdings_unset_multi_institutions_missing_oclc_number(self, stub_session): - with pytest.raises(TypeError): - stub_session.holdings_unset_multi_institutions(instSymbols="NYP,BKL") - - def test_holdings_unset_multi_institutions_missing_inst_symbols(self, stub_session): - with pytest.raises(TypeError): - stub_session.holdings_unset_multi_institutions(oclcNumber=123) - - def test_holdings_unset_multi_institutions_invalid_oclc_number(self, stub_session): - with pytest.raises(InvalidOclcNumber): - stub_session.holdings_unset_multi_institutions( - oclcNumber="odn1234", instSymbols="NYP,BKL" - ) - - @pytest.mark.http_code(200) - def test_holdings_unset_multi_institutions_stale_token( - self, mock_now, stub_session, mock_session_response - ): - stub_session.authorization.token_expires_at = datetime.datetime.now( - datetime.timezone.utc - ) - datetime.timedelta(0, 1) - with does_not_raise(): - assert stub_session.authorization.is_expired() is True - stub_session.holdings_unset_multi_institutions( - oclcNumber=850940548, instSymbols="NYP,BKL" - ) - assert stub_session.authorization.token_expires_at == datetime.datetime( - 2020, 1, 1, 17, 19, 58, tzinfo=datetime.timezone.utc - ) - assert stub_session.authorization.is_expired() is False - @pytest.mark.http_code(200) def test_search_brief_bibs_other_editions( self, stub_session, mock_session_response @@ -438,7 +240,7 @@ def test_search_brief_bibs_other_editions_invalid_oclc_number(self, stub_session assert msg in str(exc.value) @pytest.mark.http_code(200) - def test_seach_brief_bibs(self, stub_session, mock_session_response): + def test_search_brief_bibs(self, stub_session, mock_session_response): assert stub_session.search_brief_bibs(q="ti:Zendegi").status_code == 200 @pytest.mark.parametrize("argm", [(None), ("")]) @@ -585,7 +387,7 @@ def test_get_institution_holdings(self, live_keys): ) with MetadataSession(authorization=token) as session: - response = session.get_institution_holdings("982651100") + response = session.get_institution_holdings("982651100")[0] assert ( response.url @@ -611,7 +413,7 @@ def test_holding_set_unset(self, live_keys): ) with MetadataSession(authorization=token) as session: - response = session.get_institution_holdings("850940548") + response = session.get_institution_holdings("850940548")[0] holdings = response.json()["holdings"] # make sure no holdings are set initially @@ -635,70 +437,6 @@ def test_holding_set_unset(self, live_keys): ) assert response.json()["action"] == "Unset Holdings" - @pytest.mark.holdings - def test_holdings_set(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.holdings_set([850940548, 850940552, 850940554]) - assert type(response) is list - assert response[0].status_code == 207 - assert ( - response[0].url - == "https://worldcat.org/ih/datalist?oclcNumbers=850940548%2C850940552%2C850940554" - ) - assert sorted(response[0].json().keys()) == sorted( - ["entries", "extensions"] - ) - assert sorted(response[0].json()["entries"][0]) == sorted( - ["title", "content", "updated"] - ) - assert sorted(response[0].json()["entries"][0]["content"]) == sorted( - [ - "requestedOclcNumber", - "currentOclcNumber", - "institution", - "status", - "detail", - ] - ) - - @pytest.mark.holdings - def test_holdings_unset(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.holdings_unset([850940548, 850940552, 850940554]) - assert type(response) is list - assert response[0].status_code == 207 - assert ( - response[0].url - == "https://worldcat.org/ih/datalist?oclcNumbers=850940548%2C850940552%2C850940554&cascade=0" - ) - assert sorted(response[0].json().keys()) == sorted( - ["entries", "extensions"] - ) - assert sorted(response[0].json()["entries"][0]) == sorted( - ["title", "content", "updated"] - ) - assert sorted(response[0].json()["entries"][0]["content"]) == sorted( - [ - "requestedOclcNumber", - "currentOclcNumber", - "institution", - "status", - "detail", - ] - ) - def test_brief_bibs_other_editions(self, live_keys): fields = sorted(["briefRecords", "numberOfRecords"]) token = WorldcatAccessToken( diff --git a/tests/test_query.py b/tests/test_query.py index b4f626b..78cc8d5 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -150,13 +150,3 @@ def test_query_unexpected_exception(stub_session, mock_unexpected_error): Query(stub_session, prepped) assert "Unexpected request error: " in str(exc.value) - - -@pytest.mark.http_code(409) -def test_query_holding_endpoint_409_http_code(stub_session, mock_session_response): - req = Request("POST", "https://worldcat.org/ih/data", params={"foo": "bar"}) - prepped = stub_session.prepare_request(req) - with does_not_raise(): - query = Query(stub_session, prepped) - - assert query.response.status_code == 409