introduce the AuthenticationError Exception

lexicon/providers/aliyun.py

@@ -11,7 +11,7 @@
 import requests
 from six.moves import urllib
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -42,7 +42,7 @@ def _authenticate(self):
         response = self._request_aliyun("DescribeDomainInfo")
 
         if "DomainId" not in response:
-            raise ValueError(f"failed to fetch basic domain info for {self.domain}")
+            raise AuthenticationError(f"failed to fetch basic domain info for {self.domain}")
 
         self.domain_id = response["DomainId"]
 

lexicon/providers/aurora.py

@@ -8,7 +8,7 @@
 
 import requests
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -32,17 +32,14 @@ def __init__(self, config):
         self.api_endpoint = "https://api.auroradns.eu"
 
     def _authenticate(self):
-        zone = None
         payload = self._get("/zones")
 
         for item in payload:
             if item["name"] == self.domain:
-                zone = item
-
-        if not zone:
-            raise Exception("No domain found")
-
-        self.domain_id = zone["id"]
+                self.domain_id = item["id"]
+                break
+        else:
+            raise AuthenticationError("No domain found")
 
     # Create record. If record already exists with the same content, do nothing'
     def _create_record(self, rtype, name, content):

lexicon/providers/azure.py

@@ -16,7 +16,7 @@
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -279,7 +279,7 @@ def _authenticate(self):
         ]
 
         if not our_data:
-            raise Exception(
+            raise AuthenticationError(
                 "Resource group `{0}` in subscription `{1}` "
                 "does not contain the DNS zone `{2}`".format(
                     resource_group, subscription_id, self.domain

lexicon/providers/base.py

@@ -5,6 +5,10 @@
 from lexicon.config import ConfigResolver, legacy_config_resolver
 
 
+class AuthenticationError(Exception):
+    pass
+
+
 class Provider(ABC):
     """
     This is the base class for all lexicon Providers.
@@ -67,7 +71,7 @@ def authenticate(self) -> None:
         Authenticate against provider,
         Make any requests required to get the domain's id for this provider,
         so it can be used in subsequent calls.
-        Should throw an error if authentication fails for any reason,
+        Should throw AuthenticationError or requests.HTTPError if authentication fails for any reason,
         of if the domain does not exist.
         """
         self._authenticate()

lexicon/providers/cloudflare.py

@@ -4,7 +4,7 @@
 
 import requests
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -50,16 +50,16 @@ def _authenticate(self):
             payload = self._get("/zones", {"name": self.domain, "status": "active"})
 
             if not payload["result"]:
-                raise Exception("No domain found")
+                raise AuthenticationError("No domain found")
             if len(payload["result"]) > 1:
-                raise Exception("Too many domains found. This should not happen")
+                raise AuthenticationError("Too many domains found. This should not happen")
 
             self.domain_id = payload["result"][0]["id"]
         else:
             payload = self._get(f"/zones/{zone_id}")
 
             if not payload["result"]:
-                raise Exception(f"No domain found for Zone ID {zone_id}")
+                raise AuthenticationError(f"No domain found for Zone ID {zone_id}")
 
             self.domain_id = zone_id
 

lexicon/providers/cloudxns.py

@@ -8,7 +8,7 @@
 
 import requests
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -36,9 +36,8 @@ def _authenticate(self):
             if record["domain"] == self.domain + ".":
                 self.domain_id = record["id"]
                 break
-
-        if self.domain_id is None:
-            raise Exception("No domain found")
+        else:
+            raise AuthenticationError("No domain found")
 
     # Create record. If record already exists with the same content, do nothing'
     def _create_record(self, rtype, name, content):

lexicon/providers/conoha.py

@@ -4,7 +4,7 @@
 
 import requests
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -85,9 +85,9 @@ def _authenticate(self):
         payload = self._get("/domains", {"name": self._fqdn_name(self.domain)})
 
         if not payload["domains"]:
-            raise Exception("No domain found")
+            raise AuthenticationError("No domain found")
         if len(payload["domains"]) > 1:
-            raise Exception("Too many domains found. This should not happen")
+            raise AuthenticationError("Too many domains found. This should not happen")
 
         self.domain_id = payload["domains"][0]["id"]
 

lexicon/providers/constellix.py

@@ -24,7 +24,7 @@
 
 import requests
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -61,10 +61,9 @@ def _authenticate(self):
             if domain["name"] == self.domain:
                 self.domain_id = domain["id"]
                 self.domain_details = domain
-                continue
-
-        if not self.domain_id:
-            raise Exception("No domain found")
+                break
+        else:
+            raise AuthenticationError("No domain found")
 
     # Create record. If record already exists with the same content, do nothing'
 

lexicon/providers/dinahosting.py

@@ -5,7 +5,7 @@
 
 import requests
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -61,14 +61,13 @@ def _authenticate(self):
         payload = self._post("", data)
 
         if not payload["data"]:
-            raise Exception("No domain found")
-        domain_found = False
+            raise AuthenticationError("No domain found")
+
         for data in payload["data"]:
             if data["domain"] == self.domain:
-                domain_found = True
                 break
-        if not domain_found:
-            raise Exception("Requested domain is not among the owned domains")
+        else:
+            raise AuthenticationError("Requested domain is not among the owned domains")
 
         self.domain_id = self.domain
 

lexicon/providers/directadmin.py

@@ -6,7 +6,7 @@
 import requests
 from requests.auth import HTTPBasicAuth
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -46,16 +46,14 @@ def _authenticate(self):
         except requests.exceptions.HTTPError as err:
             # A 401 error will be returned in case of incorrect or missing
             # credentials
-            cause = err.response.json()["error"]
-            raise Exception(cause)
+            raise err
 
         # The response is a URL encoded array containing all available domains
         domains = [domain.split("=").pop() for domain in response.split("&")]
-
         try:
             self.domain_id = domains.index(self.domain)
-        except BaseException:
-            raise Exception(f"Domain {self.domain} not found")
+        except ValueError:
+            raise AuthenticationError(f"Domain {self.domain} not found")
 
     def _create_record(self, rtype, name, content):
         # Refuse to create duplicate records

lexicon/providers/dnsimple.py

@@ -4,7 +4,7 @@
 
 import requests
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -42,7 +42,7 @@ def _authenticate(self):
         payload = self._get("/accounts")
 
         if not payload[0]["id"]:
-            raise Exception("No account id found")
+            raise AuthenticationError("No account id found")
 
         for account in payload:
             if account["plan_identifier"] is None:
@@ -60,9 +60,9 @@ def _authenticate(self):
             if dompayload and dompayload[0]["id"]:
                 self.account_id = account["id"]
                 self.domain_id = dompayload[0]["id"]
-
-        if not self.account_id:
-            raise Exception(f"No domain found like {self.domain}")
+                break
+        else:
+            raise AuthenticationError(f"No domain found like {self.domain}")
 
     # Create record. If record already exists with the same content, do nothing
 

lexicon/providers/dnsmadeeasy.py

@@ -10,7 +10,7 @@
 from requests.adapters import HTTPAdapter
 from urllib3.util.retry import Retry  # type: ignore
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -70,7 +70,7 @@ def _authenticate(self):
                 raise
 
         if not payload or not payload["id"]:
-            raise Exception("No domain found")
+            raise AuthenticationError("No domain found")
 
         self.domain_id = payload["id"]
 

lexicon/providers/dnspark.py

@@ -4,7 +4,7 @@
 
 import requests
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -30,7 +30,7 @@ def _authenticate(self):
         payload = self._get(f"/dns/{self.domain}")
 
         if not payload["additional"]["domain_id"]:
-            raise Exception("No domain found")
+            raise AuthenticationError("No domain found")
 
         self.domain_id = payload["additional"]["domain_id"]
 

lexicon/providers/dnspod.py

@@ -3,7 +3,7 @@
 
 import requests
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -29,7 +29,7 @@ def _authenticate(self):
         payload = self._post("/Domain.Info", {"domain": self.domain})
 
         if payload["status"]["code"] != "1":
-            raise Exception(payload["status"]["message"])
+            raise AuthenticationError(payload["status"]["message"])
 
         self.domain_id = payload["domain"]["id"]
 

lexicon/providers/dreamhost.py

@@ -6,7 +6,7 @@
 
 import requests
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -102,13 +102,13 @@ def _authenticate(self):
         payload = self._get("domain-list_domains")
         data = payload.get("data", None)
         if data is None:
-            raise Exception("Domain not found")
+            raise AuthenticationError("Domain not found")
 
         for domain in data:
             if domain.get("domain", "") == self.domain:
                 self.domain_id = self.domain
         if self.domain_id is None:
-            raise Exception("Domain not found")
+            raise AuthenticationError("Domain not found")
 
     def _create_record(self, rtype, name, content):
         name = self._full_name(name)

lexicon/providers/dynu.py

@@ -4,7 +4,7 @@
 
 import requests
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -33,8 +33,8 @@ def _authenticate(self):
             if domain["name"].lower() == self.domain.lower():
                 self.domain_id = domain["id"]
                 break
-        if not self.domain_id:
-            raise Exception("No matching domain found")
+        else:
+            raise AuthenticationError("No matching domain found")
 
     # Create record. If record already exists with the same content, do nothing.
     def _create_record(self, rtype, name, content):

lexicon/providers/easydns.py

@@ -4,7 +4,7 @@
 
 import requests
 
-from lexicon.providers.base import Provider as BaseProvider
+from lexicon.providers.base import Provider as BaseProvider, AuthenticationError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -34,7 +34,7 @@ def _authenticate(self):
         payload = self._get(f"/domain/{self.domain}")
 
         if payload["data"]["exists"] == "N":
-            raise Exception("No domain found")
+            raise AuthenticationError("No domain found")
 
         self.domain_id = payload["data"]["id"]