diff --git a/gso/settings.py b/gso/settings.py
index 8ccffc31e74656260538766f4e5c955c6700c16b..78a27d756b03de700f1f7afcbcca6fe681a960f9 100644
--- a/gso/settings.py
+++ b/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()))
diff --git a/gso/utils/__init__.py b/gso/utils/__init__.py
index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..62340c1d5a8267a8c0a65703365077ed3a17b192 100644
--- a/gso/utils/__init__.py
+++ b/gso/utils/__init__.py
@@ -0,0 +1 @@
+"""Utility module that contains helper methods, exceptions, etc."""
diff --git a/gso/utils/device_info.py b/gso/utils/device_info.py
index 5a139889229efd80918e45079c805f18461e9dbb..669fb55aa06c16a46dc83768062353747a88f342 100644
--- a/gso/utils/device_info.py
+++ b/gso/utils/device_info.py
@@ -1,7 +1,11 @@
+"""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)
diff --git a/gso/utils/exceptions.py b/gso/utils/exceptions.py
index 541eb0c6d0f9f129df47fae4757f961ad8200498..66ce6d9e7f70679a7f813f5f7f24d8ec0c53fa31 100644
--- a/gso/utils/exceptions.py
+++ b/gso/utils/exceptions.py
@@ -1,3 +1,6 @@
+"""Custom exceptions for :term:`GSO`."""
+
+
class NotFoundError(Exception):
"""Exception raised for not found search."""
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index 8bbcca4ff01c7ad228dea0c55cb8fabb006e65e8..c2396a16833550f4ef49ce544f00152c95afd672 100644
--- a/gso/utils/helpers.py
+++ b/gso/utils/helpers.py
@@ -1,3 +1,5 @@
+"""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
diff --git a/gso/worker.py b/gso/worker.py
index 2376ac1019ce703aee840e85c1f1c6db6716446d..b2abfe6f5a52192454d3d691ba1715df313fc6ac 100644
--- a/gso/worker.py
+++ b/gso/worker.py
@@ -1,3 +1,5 @@
+"""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()
diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index 17341c994b81b6522ebb9bb7dd7629e29575c539..c0ee85a5dbca530002eba57cf5f23ec776ae8f6c 100644
--- a/gso/workflows/router/create_router.py
+++ b/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))
diff --git a/test/schemas/test_types.py b/test/schemas/test_types.py
index 5265d08097fc2ef50b3a929a6408f3a450de868a..ea0f4688ed856c14f5d6660662a3b910d54f05ba 100644
--- a/test/schemas/test_types.py
+++ b/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(