Migrate Python linting to Ruff

* Change exceptions Pending and Revoked to PendingError and RevokedError (linting)
* Refactor hard-coded numbers (linting)

.github/workflows/checks.yaml

@@ -11,7 +11,10 @@ name: 'checks'
 
 jobs:
   linting:
-    uses: broadinstitute/shared-workflows/.github/workflows/[email protected]
+    uses: broadinstitute/shared-workflows/.github/workflows/poetry-linting.yml@tex_ruff
+    with:
+      use_pylama: false
+      use_ruff: true
   unit-tests:
     uses: broadinstitute/shared-workflows/.github/workflows/[email protected]
     with:

.gitignore

@@ -17,3 +17,4 @@ __pycache__/
 
 # VSCode settings
 .vscode
+.ruff_cache

.pre-commit-config.yaml

@@ -1,7 +1,7 @@
 ---
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.4.0
+    rev: v4.5.0
     hooks:
       - id: check-added-large-files
       - id: check-ast
@@ -18,7 +18,6 @@ repos:
       - id: detect-private-key
       - id: end-of-file-fixer
         exclude: '.bumpversion.cfg'
-      - id: fix-encoding-pragma
       - id: mixed-line-ending
       - id: name-tests-test
         args:
@@ -27,10 +26,12 @@ repos:
         args:
           - -b main
       - id: trailing-whitespace
-  - repo: https://github.com/broadinstitute/mirrors-pylama.git
-    rev: v1.0.2
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    # Ruff version.
+    rev: v0.0.292
     hooks:
-      - id: pylama
+      - id: ruff
+        args: [--fix, --exit-non-zero-on-fix]
   - repo: https://github.com/adrienverge/yamllint.git
     rev: v1.32.0
     hooks:

README.md

@@ -94,7 +94,7 @@ while(True):
         cert_pem = ssl.collect(cert_id=result["sslId"], cert_format="x509CO")
         print(cert_pem)
         break
-    except Pending:
+    except PendingError:
         print("Certificate is still pending...sleeping for 60s")
         sleep(60)
         continue

cert_manager/__init__.py

@@ -1,15 +1,16 @@
-# -*- coding: utf-8 -*-
 """Initialize the cert_manager module."""
 
+from ._helpers import PendingError
 from .acme import ACMEAccount
 from .admin import Admin
 from .client import Client
 from .domain import Domain
-from .report import Report
-from ._helpers import Pending
 from .organization import Organization
 from .person import Person
+from .report import Report
 from .smime import SMIME
 from .ssl import SSL
 
-__all__ = ["ACMEAccount", "Admin", "Client", "Domain", "Organization", "Pending", "Person", "Report", "SMIME", "SSL"]
+__all__ = [
+    "ACMEAccount", "Admin", "Client", "Domain", "Organization", "PendingError", "Person", "Report", "SMIME", "SSL"
+]

cert_manager/__version__.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 """Keep track of the version for the cert_manager module."""
 
 __title__ = "cert_manager"

cert_manager/_certificates.py

@@ -1,11 +1,11 @@
-# -*- coding: utf-8 -*-
 """Define the cert_manager._certificate.Certificates base class."""
 
 import logging
+
 from requests.exceptions import HTTPError
 
-from ._helpers import CustomFieldsError, Pending
 from ._endpoint import Endpoint
+from ._helpers import CustomFieldsError, PendingError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -41,6 +41,7 @@ def __init__(self, client, endpoint, api_version="v1"):
         # Set to None initially.  Will be filled in by methods later.
         self.__cert_types = None
         self.__custom_fields = None
+        self.__reason_maxlen = 512
 
     @property
     def types(self):
@@ -81,8 +82,7 @@ def custom_fields(self):
         return self.__custom_fields
 
     def _validate_custom_fields(self, custom_fields):
-        """Check the structure and contents of a list of dicts representing custom fields
-        Raise exceptions if validation fails
+        """Check the structure and contents of a list of custom fields dicts. Raise exceptions if validation fails.
 
         :raises Exception: if any of the validation steps fail
         """
@@ -112,7 +112,7 @@ def _validate_custom_fields(self, custom_fields):
     def collect(self, cert_id, cert_format):
         """Retrieve an existing certificate from the API.
 
-        This method will raise a Pending exception if the certificate is still in a pending state.
+        This method will raise a PendingError exception if the certificate is still in a pending state.
 
         :param int cert_id: The certificate ID
         :param str cert_format: The format in which to retreive the certificate. Allowed values: *self.valid_formats*
@@ -126,7 +126,7 @@ def collect(self, cert_id, cert_format):
         try:
             result = self._client.get(url)
         except HTTPError as exc:
-            raise Pending(f"certificate {cert_id} still in 'pending' state") from exc
+            raise PendingError(f"certificate {cert_id} still in 'pending' state") from exc
 
         # The certificate is ready for collection
         return result.content.decode(result.encoding)
@@ -229,8 +229,8 @@ def revoke(self, cert_id, reason=""):
         url = self._url(f"/revoke/{cert_id}")
 
         # Sectigo has a 512 character limit on the "reason" message, so catch that here.
-        if (not reason) or (len(reason) > 511):
-            raise ValueError("Sectigo limit: reason must be > 0 character and < 512 characters")
+        if not reason or len(reason) >= self.__reason_maxlen:
+            raise ValueError(f"Sectigo limit: reason must be > 0 character and < {self.__reason_maxlen} characters")
 
         data = {"reason": reason}
 

cert_manager/_endpoint.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 """Define the cert_manager._endpoint.Endpoint base class."""
 
 import logging
@@ -19,6 +18,8 @@ def __init__(self, client, endpoint, api_version="v1"):
         self._client = client
         self._api_version = api_version
         self._api_url = self.create_api_url(client.base_url, endpoint, self._api_version)
+        self._expected_code = 201
+        self._capture_err_code = 400
 
     @property
     def api_version(self):

cert_manager/_helpers.py

@@ -1,9 +1,8 @@
-# -*- coding: utf-8 -*-
 """Define helper functions used by classes in this module."""
 
-from functools import wraps
 import logging
 import re
+from functools import wraps
 
 from requests.exceptions import HTTPError
 
@@ -36,23 +35,18 @@ def log_traffic(*args, **kwargs):
 
             # Check if the URL or headers exist in the parameters
             # Note: *self* will be the first argument, so actual arguments start after that.
-            url = kwargs.get("url", "")
-            if not url:
-                if len(args) > 1:
-                    url = args[1]
-            headers = kwargs.get("headers", "")
-            if not headers:
-                if len(args) > 2:
-                    headers = args[2]
-            data = kwargs.get("data", "")
-            if not data:
-                if len(args) > 3:
-                    data = args[3]
+            params = ["url", "headers", "data"]
+            values = {}
+            for index, param in enumerate(params, start=1):
+                values[param] = kwargs.get(param, "")
+                if not values[param]:
+                    if len(args) > index:
+                        values[param] = args[index]
 
             # Print out before messages with URL and header data
-            traffic_logger.debug(f"Performing a {func_name} on url: {url}")
-            traffic_logger.debug(f"Extra request headers: {headers}")
-            traffic_logger.debug(f"Data: {data}")
+            traffic_logger.debug(f"Performing a {func_name} on url: {values['url']}")
+            traffic_logger.debug(f"Extra request headers: {values['headers']}")
+            traffic_logger.debug(f"Data: {values['data']}")
 
             # Run the wrapped function
             try:
@@ -149,13 +143,13 @@ def decorator(*args, **kwargs):
     return decorator
 
 
-class Pending(Exception):
+class PendingError(Exception):
     """Serve as a generic Exception indicating a certificate is in a pending state."""
     CODE = -183
 
 
-class Revoked(Exception):
-    """Serve as a generic Exception indicating a certificate has been revoked"""
+class RevokedError(Exception):
+    """Serve as a generic Exception indicating a certificate has been revoked."""
     CODE = -192
 
 

cert_manager/acme.py

@@ -1,8 +1,7 @@
-# -*- coding: utf-8 -*-
 """Define the cert_manager.acme.ACMEAccount class."""
 
-import re
 import logging
+import re
 
 from ._endpoint import Endpoint
 from ._helpers import paginate
@@ -11,7 +10,7 @@
 
 
 class ACMEAccountCreationResponseError(Exception):
-    """An error (other than HTTPError) occurred while processing ACME Account Creation API response"""
+    """An error (other than HTTPError) occurred while processing ACME Account Creation API response."""
 
 
 class ACMEAccount(Endpoint):
@@ -36,7 +35,6 @@ def __init__(self, client, api_version="v1"):
         """
         super().__init__(client=client, endpoint="/acme", api_version=api_version)
         self._api_url = self._url("/account")
-
         self.__acme_accounts = None
 
     def all(self, org_id, force=False):
@@ -111,7 +109,7 @@ def create(self, name, acme_server, org_id, ev_details=None):
         result = self._client.post(self._api_url, data=data)
 
         # for status >= 400, HTTPError is raised
-        if result.status_code != 201:
+        if result.status_code != self._expected_code:
             raise ACMEAccountCreationResponseError(f"Unexpected HTTP status {result.status_code}")
         try:
             loc = result.headers["Location"]

cert_manager/admin.py

@@ -1,8 +1,8 @@
-# -*- coding: utf-8 -*-
 """Define the cert_manager.admin.Admin class."""
 
-import re
 import logging
+import re
+
 from requests.exceptions import HTTPError
 
 from ._endpoint import Endpoint
@@ -11,7 +11,7 @@
 
 
 class AdminCreationResponseError(Exception):
-    """An error (other than HTTPError) occurred while processing Admin Creation API response"""
+    """An error (other than HTTPError) occurred while processing Admin Creation API response."""
 
 
 class Admin(Endpoint):
@@ -44,9 +44,8 @@ def all(self, force=False):
 
         return self.__admins
 
-    def create(self, login, email, forename, surname,  # pylint: disable=too-many-arguments
-               password, credentials, **kwargs):
-        """Create a new administrator
+    def create(self, login, email, forename, surname, password, credentials, **kwargs):  # noqa: PLR0913
+        """Create a new administrator.
 
         :param str login: Login name of admin to create
         :param str email: Email of admin to create
@@ -71,20 +70,21 @@ def create(self, login, email, forename, surname,  # pylint: disable=too-many-ar
             "password": password,
             "credentials": credentials,
         }
+
         for key, value in kwargs.items():
             data[key] = value
 
         try:
             result = self._client.post(self._api_url, data=data)
         except HTTPError as exc:
             status_code = exc.response.status_code
-            if status_code == 400:
+            if status_code == self._capture_err_code:
                 err_response = exc.response.json()
                 raise ValueError(err_response["description"]) from exc
             raise exc
 
         # for status >= 400, HTTPError is raised
-        if result.status_code != 201:
+        if result.status_code != self._expected_code:
             raise AdminCreationResponseError(f"Unexpected HTTP status {result.status_code}")
         try:
             loc = result.headers["Location"]
@@ -115,7 +115,7 @@ def get(self, admin_id):
         return result.json()
 
     def get_idps(self):
-        """Return a list of IDPs
+        """Return a list of IDPs.
 
         :return list: A list of dictionaries representing the IDPs
         """
@@ -153,7 +153,7 @@ def update(self, admin_id, **kwargs):
             result = self._client.put(url, data=data)
         except HTTPError as exc:
             status_code = exc.response.status_code
-            if status_code == 400:
+            if status_code == self._capture_err_code:
                 err_response = exc.response.json()
                 raise ValueError(err_response["description"]) from exc
             raise exc

cert_manager/client.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 """Define the cert_manager.client.Client class."""
 
 import logging