fix some linting errors from ruff

gso/main.py

-"""
-The main module, from where GSO is run.
-"""
+"""The main module, from where GSO is run."""
 from orchestrator import OrchestratorCore
 from orchestrator.cli.main import app as core_cli
 from orchestrator.settings import AppSettings

gso/products/__init__.py

-"""
-Module that updates the domain model of GSO. Should contain all types of
-subscriptions.
-"""
+"""Module that updates the domain model of GSO. Should contain all types of subscriptions."""
 from orchestrator.domain import SUBSCRIPTION_MODEL_REGISTRY
 
 from gso.products.product_types.device import Device

gso/services/_ipam.py

@@ -109,13 +109,12 @@ def _assert_host_in_service(
 
 
 def _find_networks(network_container=None, network=None, ip_version=4):
-    """
-    If network_container is not None, find all networks within the specified
+    """If network_container is not None, find all networks within the specified
     container.
     Otherwise, if network is not None, find the specified network.
     Otherwise find all networks.
     A list of all found networks is returned (an HTTP 200 code
-    may be returned with an empty list.)
+    may be returned with an empty list.).
     """
     assert ip_version in [4, 6]
     oss = settings.load_oss_params()
@@ -196,9 +195,7 @@ def _allocate_network(
 
 
 def allocate_service_ipv4_network(service_type="", comment="", extattrs={}) -> V4ServiceNetwork:
-    """
-    Allocate IPv4 network within the container of the specified service type.
-    """
+    """Allocate IPv4 network within the container of the specified service type."""
     oss = settings.load_oss_params()
     assert oss.IPAM
     ipam_params = oss.IPAM
@@ -207,9 +204,7 @@ def allocate_service_ipv4_network(service_type="", comment="", extattrs={}) -> V
 
 
 def allocate_service_ipv6_network(service_type="", comment="", extattrs={}) -> V6ServiceNetwork:
-    """
-    Allocate IPv6 network within the container of the specified service type.
-    """
+    """Allocate IPv6 network within the container of the specified service type."""
     oss = settings.load_oss_params()
     assert oss.IPAM
     ipam_params = oss.IPAM
@@ -218,8 +213,7 @@ def allocate_service_ipv6_network(service_type="", comment="", extattrs={}) -> V
 
 
 def _find_next_available_ip(infoblox_params, network_ref=""):
-    """
-    Find the next available IP address from a network given its ref.
+    """Find the next available IP address from a network given its ref.
     Returns "NETWORK_FULL" if there's no space in the network.
     Otherwise returns the next available IP address in the network.
     """
@@ -242,8 +236,7 @@ def _find_next_available_ip(infoblox_params, network_ref=""):
 def _allocate_host(
     hostname="", addrs=None, networks=None, cname_aliases=[], dns_view="default", extattrs={}
 ) -> Union[HostAddresses, str]:
-    """
-    If networks is not None, allocate host in those networks.
+    """If networks is not None, allocate host in those networks.
     Otherwise if addrs is not None, allocate host with those addresses.
     hostname parameter must be full name including domain name.
     Return "IPV4_NETWORK_FULL" or "IPV6_NETWORK_FULL"
@@ -334,8 +327,7 @@ def allocate_service_host(
     cname_aliases=None,
     extattrs={},
 ) -> HostAddresses:
-    """
-    Allocate host record with both IPv4 and IPv6 address, and respective DNS
+    """Allocate host record with both IPv4 and IPv6 address, and respective DNS
     A and AAAA records.
     - If service_networks is provided, and it's in a valid container,
     that one is used.
@@ -465,9 +457,7 @@ def allocate_service_host(
 
 
 def delete_service_network(ipnetwork=None, service_type="") -> Union[V4ServiceNetwork, V6ServiceNetwork]:
-    """
-    Delete IPv4 or IPv6 network by CIDR.
-    """
+    """Delete IPv4 or IPv6 network by CIDR."""
     oss = settings.load_oss_params()
     assert oss.IPAM
     ipam_params = oss.IPAM
@@ -526,8 +516,7 @@ def delete_service_network(ipnetwork=None, service_type="") -> Union[V4ServiceNe
 def delete_service_host(
     hostname="", host_addresses: HostAddresses = None, cname_aliases=[], service_type=""
 ) -> Union[V4HostAddress, V6HostAddress]:
-    """
-    Delete host record and associated CNAME records.
+    """Delete host record and associated CNAME records.
     All arguments passed to this function must match together a host record in
     IPAM, and all CNAME records associated to it must also be passed exactly.
     """

gso/services/provisioning_proxy.py

-"""
-The Provisioning Proxy service, which interacts with LSO running externally.
+"""The Provisioning Proxy service, which interacts with LSO running externally.
 LSO is responsible for executing Ansible playbooks, that deploy subscriptions.
 """
 import json
@@ -25,8 +24,7 @@ logger = logging.getLogger(__name__)
 
 
 class CUDOperation(strEnum):
-    """
-    Enum for different C(R)UD operations that the provisioning proxy supports.
+    """Enum for different C(R)UD operations that the provisioning proxy supports.
     Read is not applicable, hence these become CUD and not CRUD operations.
     """
 
@@ -39,8 +37,7 @@ class CUDOperation(strEnum):
 
 
 def _send_request(endpoint: str, parameters: dict, process_id: UUIDstr, operation: CUDOperation):
-    """
-    Internal function for sending a request to LSO. The callback address is
+    """Internal function for sending a request to LSO. The callback address is
     derived using the process ID provided.
 
     :param str endpoint: The LSO-specific endpoint to call, depending on the
@@ -78,8 +75,7 @@ def _send_request(endpoint: str, parameters: dict, process_id: UUIDstr, operatio
 
 
 def provision_device(subscription: DeviceProvisioning, process_id: UUIDstr, dry_run: bool = True):
-    """
-    Function that provisions a new device using LSO.
+    """Function that provisions a new device using LSO.
 
     :param :class:`DeviceProvisioning` subscription: The subscription object
         that is to be provisioned.
@@ -95,8 +91,7 @@ def provision_device(subscription: DeviceProvisioning, process_id: UUIDstr, dry_
 def provision_ip_trunk(
     subscription: IptrunkProvisioning, process_id: UUIDstr, config_object: str, dry_run: bool = True
 ):
-    """
-    Function that provisions an IP trunk service using LSO.
+    """Function that provisions an IP trunk service using LSO.
 
     :param :class:`IptrunkProvisioning` subscription: The subscription object
         that is to be provisioned.
@@ -141,8 +136,7 @@ def provision_ip_trunk(
 
 
 def deprovision_ip_trunk(subscription: Iptrunk, process_id: UUIDstr, dry_run: bool = True):
-    """
-    Function that provisions an IP trunk service using LSO.
+    """Function that provisions an IP trunk service using LSO.
 
     :param :class:`IptrunkProvisioning` subscription: The subscription object
         that is to be provisioned.

gso/services/resource_manager.py

@@ -4,12 +4,16 @@ from gso import settings
 
 
 def import_new_router(router_name, oss_params=settings.OSSParams):
-    r = requests.post(f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/initialize-router/{router_name}")
+    r = requests.post(
+        f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/initialize-router/{router_name}", timeout=10000
+    )
     r.raise_for_status()
 
 
 def next_lag(router_name, oss_params=settings.OSSParams):
-    r = requests.post(f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/next-lag/{router_name}")
+    r = requests.post(
+        f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/next-lag/{router_name}", timeout=10000
+    )
     r.raise_for_status()
     response = r.json()
     return response["name"]
@@ -18,7 +22,8 @@ def next_lag(router_name, oss_params=settings.OSSParams):
 def next_physical(router_name, lag_name, oss_params=settings.OSSParams):
     # TODO: speed needed (if first interface)
     r = requests.post(
-        f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/next-physical/{router_name}/{lag_name}"
+        f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/next-physical/{router_name}/{lag_name}",
+        timeout=10000,
     )
     r.raise_for_status()
     response = r.json()

gso/settings.py

-"""
-GSO settings, ensuring that the required parameters are set correctly.
-"""
+"""GSO settings, ensuring that the required parameters are set correctly."""
 import ipaddress
 import json
 import os
@@ -10,9 +8,7 @@ from pydantic import BaseSettings
 
 
 class GeneralParams(BaseSettings):
-    """
-    General parameters for a GSO configuration file.
-    """
+    """General parameters for a GSO configuration file."""
 
     #: The hostname that GSO is publicly served at, used for building the
     #: callback URL that the provisioning proxy uses.
@@ -20,9 +16,7 @@ class GeneralParams(BaseSettings):
 
 
 class InfoBloxParams(BaseSettings):
-    """
-    Parameters related to InfoBlox.
-    """
+    """Parameters related to InfoBlox."""
 
     scheme: str
     wapi_version: str
@@ -32,9 +26,7 @@ class InfoBloxParams(BaseSettings):
 
 
 class V4NetworkParams(BaseSettings):
-    """
-    A set of parameters that describe an IPv4 network in InfoBlox.
-    """
+    """A set of parameters that describe an IPv4 network in InfoBlox."""
 
     containers: list[ipaddress.IPv4Network]
     networks: list[ipaddress.IPv4Network]
@@ -42,9 +34,7 @@ class V4NetworkParams(BaseSettings):
 
 
 class V6NetworkParams(BaseSettings):
-    """
-    A set of parameters that describe an IPv6 network in InfoBlox.
-    """
+    """A set of parameters that describe an IPv6 network in InfoBlox."""
 
     containers: list[ipaddress.IPv6Network]
     networks: list[ipaddress.IPv6Network]
@@ -52,8 +42,7 @@ class V6NetworkParams(BaseSettings):
 
 
 class ServiceNetworkParams(BaseSettings):
-    """
-    Parameters for InfoBlox that describe IPv4 and v6 networks, and the
+    """Parameters for InfoBlox that describe IPv4 and v6 networks, and the
     corresponding domain name that should be used as a suffix.
     """
 
@@ -64,9 +53,7 @@ class ServiceNetworkParams(BaseSettings):
 
 
 class IPAMParams(BaseSettings):
-    """
-    A set of parameters related to IPAM.
-    """
+    """A set of parameters related to IPAM."""
 
     INFOBLOX: InfoBloxParams
     LO: ServiceNetworkParams
@@ -77,9 +64,7 @@ class IPAMParams(BaseSettings):
 
 
 class ProvisioningProxyParams(BaseSettings):
-    """
-    Parameters for the provisioning proxy.
-    """
+    """Parameters for the provisioning proxy."""
 
     scheme: str
     api_base: str
@@ -88,9 +73,7 @@ class ProvisioningProxyParams(BaseSettings):
 
 
 class OSSParams(BaseSettings):
-    """
-    The set of parameters required for running GSO.
-    """
+    """The set of parameters required for running GSO."""
 
     GENERAL: GeneralParams
     IPAM: IPAMParams
@@ -99,8 +82,7 @@ class OSSParams(BaseSettings):
 
 
 def load_oss_params() -> OSSParams:
-    """
-    look for OSS_PARAMS_FILENAME in the environment and load the parameters
+    """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:

gso/workflows/__init__.py

-"""
-init class that imports all workflows into GSO.
-"""
+"""init class that imports all workflows into GSO."""
 from orchestrator.workflows import LazyWorkflowInstance
 
 LazyWorkflowInstance("gso.workflows.device.create_device", "create_device")

gso/workflows/device/create_device.py

@@ -72,8 +72,7 @@ def iso_from_ipv4(ipv4_address):
     padded_octets = [f"{x:>03}" for x in ipv4_address.split(".")]
     joined_octets = "".join(padded_octets)
     re_split = ".".join(re.findall("....", joined_octets))
-    result = ".".join(["49.51e5.0001", re_split, "00"])
-    return result
+    return ".".join(["49.51e5.0001", re_split, "00"])
 
 
 @step("Get information from IPAM")

gso/workflows/iptrunk/modify_generic.py

 import ipaddress
 
 from orchestrator.forms import FormPage, ReadOnlyField
-from orchestrator.forms.validators import Choice, UniqueConstrainedList
+from orchestrator.forms.validators import UniqueConstrainedList
 from orchestrator.targets import Target
 from orchestrator.types import FormGenerator, State, UUIDstr
 from orchestrator.workflow import done, init, step, workflow
@@ -108,7 +108,7 @@ def provision_ip_trunk_iface_dry(subscription: Iptrunk, process_id: UUIDstr) ->
 
     return {
         "subscription": subscription,
-        "label_text": ("Provision of the Trunk interface [DRY] " "Please refresh to get the results of the playbook"),
+        "label_text": "Provision of the Trunk interface [DRY]" "Please refresh to get the results of the playbook",
     }
 
 
@@ -118,7 +118,7 @@ def provision_ip_trunk_iface_real(subscription: Iptrunk, process_id: UUIDstr) ->
 
     return {
         "subscription": subscription,
-        "label_text": ("Provision of the Trunk interface [REAL] " "Please refresh to get the results of the playbook"),
+        "label_text": "Provision of the Trunk interface [REAL]" "Please refresh to get the results of the playbook",
     }
 
 
@@ -128,7 +128,7 @@ def provision_ip_trunk_lldp_iface_dry(subscription: Iptrunk, process_id: UUIDstr
 
     return {
         "subscription": subscription,
-        "label_text": ("Provision of the LLDP interface [DRY]" "Please refresh to get the results of the playbook"),
+        "label_text": "Provision of the LLDP interface [DRY]" "Please refresh to get the results of the playbook",
     }
 
 
@@ -138,7 +138,7 @@ def provision_ip_trunk_lldp_iface_real(subscription: Iptrunk, process_id: UUIDst
 
     return {
         "subscription": subscription,
-        "label_text": ("Provision of the LLDP interface [REAL]" "Please refresh to get the results of the playbook"),
+        "label_text": "Provision of the LLDP interface [REAL]" "Please refresh to get the results of the playbook",
     }
 
 

pyproject.toml

@@ -51,6 +51,7 @@ exclude = [
     "__pycache__",
     "htmlcov",
     "venv",
+    "gso/migrations"
 ]
 ignore = [
     "C417",