fix linting errors in utils module

gso/settings.py

@@ -8,6 +8,7 @@ import ipaddress
 import json
 import logging
 import os
+from pathlib import Path
 
 from pydantic import BaseSettings, NonNegativeInt
 
@@ -43,10 +44,14 @@ class InfoBloxParams(BaseSettings):
 
 
 class V4Netmask(NonNegativeInt):
+    """A valid netmask for an IPv4 network or address."""
+
     le = 32
 
 
 class V6Netmask(NonNegativeInt):
+    """A valid netmask for an IPv6 network or address."""
+
     le = 128
 
 
@@ -118,8 +123,8 @@ class OSSParams(BaseSettings):
 
 
 def load_oss_params() -> OSSParams:
-    """Look for OSS_PARAMS_FILENAME in the environment and load the parameters from that file."""
-    with open(os.environ["OSS_PARAMS_FILENAME"], encoding="utf-8") as file:
+    """Look for ``OSS_PARAMS_FILENAME`` in the environment and load the parameters from that file."""
+    with Path(os.environ["OSS_PARAMS_FILENAME"]).open(encoding="utf-8") as file:
         return OSSParams(**json.loads(file.read()))
 
 

gso/utils/__init__.py

+"""Utility module that contains helper methods, exceptions, etc."""

gso/utils/device_info.py

+"""Utility module that defines facts about different tiers of sites. Used by Netbox when creating a new device."""
+
 from pydantic import BaseModel
 
 
 class ModuleInfo(BaseModel):
+    """A collection of facts that define the tier of a site."""
+
     device_type: str
     module_bays_slots: list[int]
     module_type: str
@@ -10,7 +14,10 @@ class ModuleInfo(BaseModel):
 
 
 class TierInfo:
+    """Information for different tiers of sites."""
+
     def __init__(self) -> None:
+        """Initialise the different tiers of sites that exist."""
         self.Tier1 = ModuleInfo(
             device_type="7750 SR-7s",
             module_bays_slots=[1, 2],
@@ -27,6 +34,7 @@ class TierInfo:
         )
 
     def get_module_by_name(self, name: str) -> ModuleInfo:
+        """Retrieve a module by name."""
         return getattr(self, name)
 
 

gso/utils/exceptions.py

+"""Custom exceptions for :term:`GSO`."""
+
+
 class NotFoundError(Exception):
     """Exception raised for not found search."""
 

gso/utils/helpers.py

+"""Helper methods that are used across :term:`GSO`."""
+
 import ipaddress
 import re
 from ipaddress import IPv4Address
@@ -19,17 +21,21 @@ from gso.services.subscriptions import get_active_subscriptions_by_field_and_val
 
 
 class LAGMember(BaseModel):
+    """A :term:`LAG` member interface that consists of a name and description."""
+
     #  TODO: validate interface name
     interface_name: str
     interface_description: str
 
     def __hash__(self) -> int:
+        """Calculate the hash based on the interface name and description, so that uniqueness can be determined."""
         #  TODO: check if this is still needed
         return hash((self.interface_name, self.interface_description))
 
 
 @step("[COMMIT] Set ISIS metric to 90.000")
 def set_isis_to_90000(subscription: Iptrunk, process_id: UUIDstr, callback_route: str, tt_number: str) -> State:
+    """Workflow step for setting the :term:`ISIS` metric to 90k as an arbitrarily high value to drain a link."""
     old_isis_metric = subscription.iptrunk.iptrunk_isis_metric
     subscription.iptrunk.iptrunk_isis_metric = 90000
     provisioning_proxy.provision_ip_trunk(
@@ -175,20 +181,22 @@ def validate_ipv4_or_ipv6(value: str) -> str:
     """Validate that a value is a valid IPv4 or IPv6 address."""
     try:
         ipaddress.ip_address(value)
-        return value
     except ValueError as e:
         msg = "Enter a valid IPv4 or IPv6 address."
         raise ValueError(msg) from e
+    else:
+        return value
 
 
 def validate_country_code(country_code: str) -> str:
     """Validate that a country code is valid."""
     try:
         pycountry.countries.lookup(country_code)
-        return country_code
     except LookupError as e:
         msg = "Invalid or non-existent country code, it must be in ISO 3166-1 alpha-2 format."
         raise ValueError(msg) from e
+    else:
+        return country_code
 
 
 def validate_site_name(site_name: str) -> str:
@@ -198,8 +206,9 @@ def validate_site_name(site_name: str) -> str:
     """
     pattern = re.compile(r"^[A-Z]{3}[0-9]?$")
     if not pattern.match(site_name):
-        msg = "Enter a valid site name. It must consist of three uppercase letters (A-Z) followed by an optional single digit (0-9)."
-        raise ValueError(
-            msg,
+        msg = (
+            "Enter a valid site name. It must consist of three uppercase letters (A-Z) followed by an optional single "
+            "digit (0-9)."
         )
+        raise ValueError(msg)
     return site_name

gso/worker.py

+"""Module that sets up :term:`GSO` as a Celery worker. This will allow for the scheduling of regular task workflows."""
+
 from celery import Celery
 
 from gso import init_worker_app
@@ -5,7 +7,10 @@ from gso.settings import load_oss_params
 
 
 class OrchestratorCelery(Celery):
-    def on_init(self) -> None:
+    """A :term:`GSO` instance that functions as a Celery worker."""
+
+    def on_init(self) -> None:  # noqa: PLR6301
+        """Initialise a new Celery worker."""
         init_worker_app()
 
 

gso/workflows/router/create_router.py

@@ -107,7 +107,7 @@ def initialize_subscription(
 
 
 @step("Allocate loopback interfaces in IPAM")
-def ipam_allocate_loopback(subscription: RouterProvisioning, is_ias_connected: bool) -> State:
+def ipam_allocate_loopback(subscription: RouterProvisioning, is_ias_connected: bool) -> State:  # noqa: FBT001
     fqdn = subscription.router.router_fqdn
     loopback_v4, loopback_v6 = infoblox.allocate_host(f"lo0.{fqdn}", "LO", [fqdn], str(subscription.subscription_id))
 

test/schemas/test_types.py

@@ -24,9 +24,8 @@ def test_latitude(input_value, is_valid):
     if is_valid:
         assert LatitudeCoordinate.validate(input_value) == input_value
     else:
-        with pytest.raises(ValueError) as excinfo:
+        with pytest.raises(ValueError, match="Invalid latitude coordinate"):
             LatitudeCoordinate.validate(input_value)
-        assert "Invalid latitude coordinate" in str(excinfo.value)
 
 
 @pytest.mark.parametrize(