Add new exceptions to APICEP adapter

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 "",

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,19 +107,76 @@ 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):
     requests_mock.get("https://ws.apicep.com/cep/37503130.json", status_code=404)
 
     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)