From 982473698baef64e55b9c0fb8a33f1877bc9eaa8 Mon Sep 17 00:00:00 2001 From: Michell Stuttgart Date: Sun, 24 Nov 2024 22:15:44 -0300 Subject: [PATCH] Add new exceptions to APICEP adapter --- brazilcep/apicep.py | 40 +++++++++++++++++++++--- tests/test_apicep.py | 72 ++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 104 insertions(+), 8 deletions(-) diff --git a/brazilcep/apicep.py b/brazilcep/apicep.py index eddf345..997b3c2 100644 --- a/brazilcep/apicep.py +++ b/brazilcep/apicep.py @@ -23,15 +23,42 @@ def fetch_address(cep: str, timeout: Union[None, int], proxies: Union[None, dict a REST API to query CEP requests. Args: - cep: CEP to be searched. - timeout: How many seconds to wait for the server to return data before giving up. - proxies: Dictionary mapping protocol to the URL of the proxy. + cep: CEP to be searched + timeout: How many seconds to wait for the server to return data before giving up + proxies: Dictionary mapping protocol to the URL of the proxy + + Raises: + exceptions.ConnectionError: raised by a connection error + exceptions.HTTPError: raised by HTTP error + exceptions.URLRequired: raised by using a invalid URL to make a request + exceptions.TooManyRedirects: raised by too many redirects + exceptions.Timeout: raised by request timed out + exceptions.InvalidCEP: raised to invalid CEP requests + exceptions.BlockedByFlood: raised by flood of requests + exceptions.CEPNotFound: raised to CEP not founded requests + exceptions.BrazilCEPException: Base class for exception Returns: - Respective address data from CEP. + Address data from CEP """ - response = requests.get(URL.format(cep), timeout=timeout, proxies=proxies) + try: + response = requests.get(URL.format(cep), timeout=timeout, proxies=proxies) + + except requests.exceptions.ConnectionError as exc: + raise exceptions.ConnectionError(exc) + + except requests.exceptions.HTTPError as exc: + raise exceptions.HTTPError(exc) + + except requests.exceptions.URLRequired as exc: + raise exceptions.URLRequired(exc) + + except requests.exceptions.TooManyRedirects as exc: + raise exceptions.TooManyRedirects(exc) + + except requests.exceptions.Timeout as exc: + raise exceptions.Timeout(exc) if response.status_code == 200: address = json.loads(response.text) @@ -45,6 +72,9 @@ def fetch_address(cep: str, timeout: Union[None, int], proxies: Union[None, dict if address["status"] == 404: raise exceptions.CEPNotFound() + if address["status"] == 429 and address["message"] == "Blocked by flood cdn": + raise exceptions.BlockedByFlood() + return { "district": address.get("district") or "", "cep": address.get("code") or "", diff --git a/tests/test_apicep.py b/tests/test_apicep.py index fd0a37c..915bad8 100644 --- a/tests/test_apicep.py +++ b/tests/test_apicep.py @@ -1,6 +1,7 @@ import os import pytest +import requests from dotenv import load_dotenv from brazilcep import WebService, exceptions, get_address_from_cep @@ -24,6 +25,14 @@ def test_fetch_address_success_real(): assert address["uf"] == "MG" +@pytest.mark.skipif(SKIP_REAL_TEST, reason="Skip real teste API.") +@pytest.mark.skipif(IN_GITHUB_ACTIONS, reason="Test doesn't work in Github Actions.") +def test_fetch_address_cep_not_found_real(): + + with pytest.raises(exceptions.InvalidCEP): + get_address_from_cep("37.503-13", webservice=WebService.APICEP) + + def test_fetch_address_success(requests_mock): req_mock_text = """{ "status":200, @@ -98,15 +107,26 @@ def test_fetch_address_invalid_cep(requests_mock): def test_fetch_address_blocked_by_flood(requests_mock): - req_mock_text = """{ + + req_mock_text_400 = """{ "status":400, "message": "Blocked by flood" }""" - requests_mock.get("https://ws.apicep.com/cep/3750313.json", text=req_mock_text) + requests_mock.get("https://ws.apicep.com/cep/37503130.json", text=req_mock_text_400) with pytest.raises(exceptions.BlockedByFlood): - get_address_from_cep("37503-13", webservice=WebService.APICEP) + get_address_from_cep("37503-130", webservice=WebService.APICEP) + + req_mock_text_429 = """{ + "status":429, + "message": "Blocked by flood cdn" + }""" + + requests_mock.get("https://ws.apicep.com/cep/37503130.json", text=req_mock_text_429) + + with pytest.raises(exceptions.BlockedByFlood): + get_address_from_cep("37503-130", webservice=WebService.APICEP) def test_fetch_address_404(requests_mock): @@ -114,3 +134,49 @@ def test_fetch_address_404(requests_mock): with pytest.raises(exceptions.BrazilCEPException): get_address_from_cep("37503-130", webservice=WebService.APICEP) + + +def test_connection_error(requests_mock): + + requests_mock.get( + "https://ws.apicep.com/cep/37503130.json", exc=requests.exceptions.ConnectionError + ) + + with pytest.raises(exceptions.ConnectionError): + get_address_from_cep("37503-130", webservice=WebService.APICEP) + + +def test_http_error(requests_mock): + + requests_mock.get("https://ws.apicep.com/cep/37503130.json", exc=requests.exceptions.HTTPError) + + with pytest.raises(exceptions.HTTPError): + get_address_from_cep("37503-130", webservice=WebService.APICEP) + + +def test_url_required_error(requests_mock): + + requests_mock.get( + "https://ws.apicep.com/cep/37503130.json", exc=requests.exceptions.URLRequired + ) + + with pytest.raises(exceptions.URLRequired): + get_address_from_cep("37503-130", webservice=WebService.APICEP) + + +def test_too_many_redirects_error(requests_mock): + + requests_mock.get( + "https://ws.apicep.com/cep/37503130.json", exc=requests.exceptions.TooManyRedirects + ) + + with pytest.raises(exceptions.TooManyRedirects): + get_address_from_cep("37503-130", webservice=WebService.APICEP) + + +def test_timeout_error(requests_mock): + + requests_mock.get("https://ws.apicep.com/cep/37503130.json", exc=requests.exceptions.Timeout) + + with pytest.raises(exceptions.Timeout): + get_address_from_cep("37503-130", webservice=WebService.APICEP)