From 40bd8735f50f1f932f8b987892dc1fb004eea74e Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Fri, 29 Nov 2024 16:32:27 +0100
Subject: [PATCH] Reformat all docstrings from Sphinx to Google style
---
gso/auth/oidc.py | 22 +--
gso/cli/imports.py | 16 +--
gso/migrations/env.py | 2 -
gso/products/product_blocks/bgp_session.py | 16 +--
gso/products/product_blocks/edge_port.py | 10 +-
gso/products/product_blocks/iptrunk.py | 4 +-
.../product_blocks/l3_core_service.py | 6 +-
.../product_blocks/lan_switch_interconnect.py | 34 ++---
.../product_blocks/layer_2_circuit.py | 6 +-
gso/products/product_blocks/office_router.py | 8 +-
gso/products/product_blocks/opengear.py | 12 +-
gso/products/product_blocks/pop_vlan.py | 26 ++--
gso/products/product_blocks/router.py | 10 +-
.../product_blocks/service_binding_port.py | 32 ++---
gso/products/product_blocks/site.py | 6 +-
.../product_blocks/super_pop_switch.py | 8 +-
gso/products/product_blocks/switch.py | 10 +-
.../product_types/lan_switch_interconnect.py | 4 +-
gso/products/product_types/pop_vlan.py | 6 +-
gso/schedules/scheduling.py | 14 +-
gso/services/infoblox.py | 131 +++++++++---------
gso/services/kentik_client.py | 88 ++++++------
gso/services/librenms_client.py | 47 +++++--
gso/services/lso_client.py | 70 +++++-----
gso/services/mailer.py | 5 +-
gso/services/netbox_client.py | 7 +-
gso/services/partners.py | 9 +-
gso/services/processes.py | 4 +-
gso/services/sharepoint.py | 11 +-
gso/services/subscriptions.py | 126 +++++++++--------
gso/utils/helpers.py | 56 +++++---
gso/utils/shared_enums.py | 2 +-
gso/utils/types/interfaces.py | 24 ++--
gso/utils/types/netbox_router.py | 11 +-
gso/utils/types/tt_number.py | 6 +-
gso/utils/types/virtual_identifiers.py | 2 +-
gso/workflows/edge_port/create_edge_port.py | 8 +-
.../edge_port/create_imported_edge_port.py | 2 +-
gso/workflows/edge_port/modify_edge_port.py | 6 +-
.../edge_port/terminate_edge_port.py | 4 +-
.../iptrunk/create_imported_iptrunk.py | 4 +-
gso/workflows/iptrunk/create_iptrunk.py | 22 +--
gso/workflows/iptrunk/deploy_twamp.py | 4 +-
gso/workflows/iptrunk/migrate_iptrunk.py | 32 ++---
.../iptrunk/modify_trunk_interface.py | 10 +-
gso/workflows/iptrunk/validate_iptrunk.py | 11 +-
.../create_imported_layer_2_circuit.py | 4 +-
.../create_imported_l3_core_service.py | 2 +-
.../l3_core_service/create_l3_core_service.py | 12 +-
.../l3_core_service/modify_l3_core_service.py | 4 +-
...create_imported_lan_switch_interconnect.py | 4 +-
.../import_lan_switch_interconnect.py | 2 +-
.../terminate_lan_switch_interconnect.py | 2 +-
.../validate_lan_switch_interconnect.py | 4 +-
.../create_imported_office_router.py | 2 +-
.../opengear/create_imported_opengear.py | 2 +-
.../router/create_imported_router.py | 2 +-
gso/workflows/router/create_router.py | 9 +-
gso/workflows/router/import_router.py | 2 +-
gso/workflows/router/update_ibgp_mesh.py | 2 +-
gso/workflows/router/validate_router.py | 7 +-
gso/workflows/site/create_imported_site.py | 2 +-
.../create_imported_super_pop_switch.py | 2 +-
.../import_super_pop_switch.py | 2 +-
gso/workflows/switch/terminate_switch.py | 2 +-
pyproject.toml | 3 +
test/conftest.py | 13 +-
test/fixtures/common_fixtures.py | 3 +-
test/workflows/__init__.py | 4 +-
.../edge_port/test_create_edge_port.py | 2 +-
.../edge_port/test_validate_edge_port.py | 2 +-
utils/__init__.py | 2 +-
utils/netboxcli.py | 8 +-
73 files changed, 561 insertions(+), 498 deletions(-)
diff --git a/gso/auth/oidc.py b/gso/auth/oidc.py
index b65de542..1a10f6ca 100644
--- a/gso/auth/oidc.py
+++ b/gso/auth/oidc.py
@@ -43,10 +43,9 @@ def ensure_openid_config_loaded(func: Callable) -> Callable:
class OIDCAuthentication(OIDCAuth):
- """OIDCUser class extends the :term:`HTTPBearer` class to do extra verification.
+ """OIDCUser class extends the ``HTTPBearer`` class to do extra verification.
- The class will act as follows:
- 1. Validate the Credentials at :term: `AAI` proxy by calling the UserInfo endpoint
+ The class will act as follows: Validate the Credentials at the AAI proxy by calling the UserInfo endpoint.
"""
@staticmethod
@@ -58,10 +57,12 @@ class OIDCAuthentication(OIDCAuth):
async def userinfo(self, async_request: AsyncClient, token: str) -> OIDCUserModel:
"""Get the userinfo from the openid server.
- :param AsyncClient async_request: The async request
- :param str token: the access_token
- :return: OIDCUserModel: OIDC user model from openid server
+ Args:
+ async_request: The async request.
+ token: The access token.
+ Returns:
+ OIDCUserModel: OIDC user model from openid server.
"""
assert self.openid_config is not None, "OpenID config is not loaded" # noqa: S101
@@ -104,9 +105,12 @@ class OIDCAuthentication(OIDCAuth):
async def introspect_token(self, async_request: AsyncClient, token: str) -> dict:
"""Introspect the access token to see if it is a valid token.
- :param async_request: The async request
- :param token: the access_token
- :return: dict from openid server
+ Args:
+ async_request: The async request
+ token: the access_token
+
+ Returns:
+ dict from openid server
"""
assert self.openid_config is not None, "OpenID config is not loaded" # noqa: S101
diff --git a/gso/cli/imports.py b/gso/cli/imports.py
index c57d80bd..14a45ad0 100644
--- a/gso/cli/imports.py
+++ b/gso/cli/imports.py
@@ -64,11 +64,11 @@ class CreatePartner(BaseModel):
class SiteImportModel(BaseSiteValidatorModel):
- """The required input for importing an existing :class:`gso.products.product_types.site`."""
+ """The required input for importing an existing ``gso.products.product_types.site``."""
class RouterImportModel(BaseModel):
- """Required fields for importing an existing :class:`gso.product.product_types.router`."""
+ """Required fields for importing an existing ``gso.product.product_types.router``."""
partner: str
router_site: str
@@ -82,7 +82,7 @@ class RouterImportModel(BaseModel):
class SwitchImportModel(BaseModel):
- """Required fields for importing an existing :class:`gso.product.product_types.switch`."""
+ """Required fields for importing an existing ``gso.product.product_types.switch``."""
fqdn: str
ts_port: PortNumber
@@ -92,7 +92,7 @@ class SwitchImportModel(BaseModel):
class SuperPopSwitchImportModel(BaseModel):
- """Required fields for importing an existing :class:`gso.product.product_types.super_pop_switch`."""
+ """Required fields for importing an existing ``gso.product.product_types.super_pop_switch``."""
partner: str
super_pop_switch_site: str
@@ -102,7 +102,7 @@ class SuperPopSwitchImportModel(BaseModel):
class OfficeRouterImportModel(BaseModel):
- """Required fields for importing an existing :class:`gso.product.product_types.office_router`."""
+ """Required fields for importing an existing ``gso.product.product_types.office_router``."""
partner: str
office_router_site: str
@@ -113,7 +113,7 @@ class OfficeRouterImportModel(BaseModel):
class IptrunkImportModel(BaseModel):
- """Required fields for importing an existing :class:`gso.products.product_types.iptrunk`."""
+ """Required fields for importing an existing ``gso.products.product_types.iptrunk``."""
partner: str
geant_s_sid: str | None
@@ -178,7 +178,7 @@ class IptrunkImportModel(BaseModel):
class OpenGearImportModel(BaseModel):
- """Required fields for importing an existing :class:`gso.products.product_types.opengear`."""
+ """Required fields for importing an existing ``gso.products.product_types.opengear``."""
partner: str
opengear_site: str
@@ -189,7 +189,7 @@ class OpenGearImportModel(BaseModel):
class EdgePortImportModel(BaseModel):
- """Required fields for importing an existing :class:`gso.products.product_types.edge_port`."""
+ """Required fields for importing an existing ``gso.products.product_types.edge_port``."""
node: str
service_type: EdgePortType
diff --git a/gso/migrations/env.py b/gso/migrations/env.py
index 968abeb9..fb535946 100644
--- a/gso/migrations/env.py
+++ b/gso/migrations/env.py
@@ -30,7 +30,6 @@ def run_migrations_offline() -> None:
Calls to context.execute() here emit the given string to the
script output.
-
"""
url = config.get_main_option("sqlalchemy.url")
context.configure(
@@ -46,7 +45,6 @@ def run_migrations_online() -> None:
In this scenario we need to create an Engine
and associate a connection with the context.
-
"""
# this callback is used to prevent an auto-migration from being generated
diff --git a/gso/products/product_blocks/bgp_session.py b/gso/products/product_blocks/bgp_session.py
index 2a57eb69..8113654a 100644
--- a/gso/products/product_blocks/bgp_session.py
+++ b/gso/products/product_blocks/bgp_session.py
@@ -1,4 +1,4 @@
-""":term:`BGP` session product block."""
+"""BGP session product block."""
import strawberry
from orchestrator.domain.base import ProductBlockModel
@@ -11,7 +11,7 @@ from gso.utils.types.ip_address import IPAddress
@strawberry.enum
class IPFamily(strEnum):
- """Possible IP families of a :term:`BGP` peering."""
+ """Possible IP families of a BGP peering."""
V4UNICAST = "ipv4"
V6UNICAST = "ipv6"
@@ -21,14 +21,14 @@ class IPFamily(strEnum):
@strawberry.enum
class IPTypes(strEnum):
- """Possible IP types of a :term:`BGP` peering."""
+ """Possible IP types of a BGP peering."""
IPV4 = "ipv4"
IPV6 = "ipv6"
class BGPSessionInactive(ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="BGPSession"):
- """A :term:`BGP` session that is currently inactive. See :class:`BGPSession`."""
+ """A BGP session that is currently inactive. See ``BGPSession``."""
peer_address: IPAddress | None = None
families: list[IPFamily] = Field(default_factory=list)
@@ -45,7 +45,7 @@ class BGPSessionInactive(ProductBlockModel, lifecycle=[SubscriptionLifecycle.INI
class BGPSessionProvisioning(BGPSessionInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """A :term:`BGP` session that is currently being provisioned. See :class:`BGPSession`."""
+ """A BGP session that is currently being provisioned. See ``BGPSession``."""
peer_address: IPAddress
families: list[IPFamily]
@@ -62,7 +62,7 @@ class BGPSessionProvisioning(BGPSessionInactive, lifecycle=[SubscriptionLifecycl
class BGPSession(BGPSessionProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
- """A :term:`BGP` session that is currently deployed in the network."""
+ """A BGP session that is currently deployed in the network."""
#: The peering address of the session.
peer_address: IPAddress
@@ -70,7 +70,7 @@ class BGPSession(BGPSessionProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE
families: list[IPFamily]
#: Whether any custom policies exist for this session.
has_custom_policies: bool
- #: The authentication key of the :term:`BGP` session.
+ #: The authentication key of the BGP session.
authentication_key: str | None
#: Whether multi-path is enabled.
multipath_enabled: bool
@@ -82,7 +82,7 @@ class BGPSession(BGPSessionProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE
is_passive: bool
#: Whether Remote Triggered Blackhole is enabled
rtbh_enabled: bool
- #: Settings for :term:`BFD`.
+ #: Settings for BFD.
bfd_enabled: bool
#: The IP type of the session.
ip_type: IPTypes
diff --git a/gso/products/product_blocks/edge_port.py b/gso/products/product_blocks/edge_port.py
index 3ec09b2a..b3400e0f 100644
--- a/gso/products/product_blocks/edge_port.py
+++ b/gso/products/product_blocks/edge_port.py
@@ -60,7 +60,7 @@ class EdgePortAEMemberBlock(EdgePortAEMemberBlockProvisioning, lifecycle=[Subscr
class EdgePortBlockInactive(
ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="EdgePortBlock"
):
- """An edge port that's currently inactive. See :class:`EdgePortBlock`."""
+ """An edge port that's currently inactive. See ``EdgePortBlock``."""
node: RouterBlockInactive | None = None
edge_port_name: str | None = None
@@ -77,7 +77,7 @@ class EdgePortBlockInactive(
class EdgePortBlockProvisioning(EdgePortBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """An edge port that's being provisioned. See :class:`EdgePortBlock`."""
+ """An edge port that's being provisioned. See ``EdgePortBlock``."""
node: RouterBlockProvisioning
edge_port_name: str
@@ -98,11 +98,11 @@ class EdgePortBlock(EdgePortBlockProvisioning, lifecycle=[SubscriptionLifecycle.
#: The router that this edge port is connected to.
node: RouterBlock
- #: The name of the edge port, in our case, corresponds to the name of the :term:`LAG` interface.
+ #: The name of the edge port, in our case, corresponds to the name of the LAG interface.
edge_port_name: str
#: A description of the edge port.
edge_port_description: str | None = None
- #: Indicates whether :term:`LACP` is enabled for this edge port.
+ #: Indicates whether LACP is enabled for this edge port.
enable_lacp: bool
#: The type of encapsulation used on this edge port, by default DOT1Q.
encapsulation: EncapsulationType = EncapsulationType.DOT1Q
@@ -118,5 +118,5 @@ class EdgePortBlock(EdgePortBlockProvisioning, lifecycle=[SubscriptionLifecycle.
ignore_if_down: bool = False
#: The GEANT GA ID associated with this edge port, if any.
geant_ga_id: str | None = None
- #: A list of :term:`LAG` members associated with this edge port.
+ #: A list of LAG members associated with this edge port.
edge_port_ae_members: LAGMemberList[EdgePortAEMemberBlock] # type: ignore[assignment]
diff --git a/gso/products/product_blocks/iptrunk.py b/gso/products/product_blocks/iptrunk.py
index 7978edd6..e68911dd 100644
--- a/gso/products/product_blocks/iptrunk.py
+++ b/gso/products/product_blocks/iptrunk.py
@@ -94,7 +94,7 @@ class IptrunkBlockInactive(
lifecycle=[SubscriptionLifecycle.INITIAL],
product_block_name="IptrunkBlock",
):
- """A trunk that's currently inactive, see :class:`IptrunkBlock`."""
+ """A trunk that's currently inactive, see ``IptrunkBlock``."""
geant_s_sid: str | None = None
iptrunk_description: str | None = None
@@ -108,7 +108,7 @@ class IptrunkBlockInactive(
class IptrunkBlockProvisioning(IptrunkBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """A trunk that's currently being provisioned, see :class:`IptrunkBlock`."""
+ """A trunk that's currently being provisioned, see ``IptrunkBlock``."""
geant_s_sid: str | None = None
iptrunk_description: str | None = None
diff --git a/gso/products/product_blocks/l3_core_service.py b/gso/products/product_blocks/l3_core_service.py
index add0724b..129b9477 100644
--- a/gso/products/product_blocks/l3_core_service.py
+++ b/gso/products/product_blocks/l3_core_service.py
@@ -31,20 +31,20 @@ class AccessPort(AccessPortProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE
#: The type of Access Port
ap_type: APType
- #: The corresponding :term:`SBP` of this Access Port.
+ #: The corresponding SBP of this Access Port.
sbp: ServiceBindingPort
class L3CoreServiceBlockInactive(
ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="L3CoreServiceBlock"
):
- """An inactive L3 Core service subscription. See :class:`L3CoreServiceBlock`."""
+ """An inactive L3 Core service subscription. See ``L3CoreServiceBlock``."""
ap_list: list[AccessPortInactive] = Field(default_factory=list)
class L3CoreServiceBlockProvisioning(L3CoreServiceBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """A provisioning L3 Core Service subscription. See :class:`L3CoreServiceBlock`."""
+ """A provisioning L3 Core Service subscription. See ``L3CoreServiceBlock``."""
ap_list: list[AccessPortProvisioning] # type: ignore[assignment]
diff --git a/gso/products/product_blocks/lan_switch_interconnect.py b/gso/products/product_blocks/lan_switch_interconnect.py
index 945efc5a..a3128a72 100644
--- a/gso/products/product_blocks/lan_switch_interconnect.py
+++ b/gso/products/product_blocks/lan_switch_interconnect.py
@@ -14,7 +14,7 @@ class LanSwitchInterconnectInterfaceBlockInactive(
lifecycle=[SubscriptionLifecycle.INITIAL],
product_block_name="LanSwitchInterconnectInterfaceBlock",
):
- """An inactive :term:`LAN` Switch Interconnect interface."""
+ """An inactive LAN Switch Interconnect interface."""
interface_name: str | None = None
interface_description: str | None = None
@@ -23,7 +23,7 @@ class LanSwitchInterconnectInterfaceBlockInactive(
class LanSwitchInterconnectInterfaceBlockProvisioning(
LanSwitchInterconnectInterfaceBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
):
- """A :term:`LAN` Switch Interconnect interface that is being provisioned."""
+ """A LAN Switch Interconnect interface that is being provisioned."""
interface_name: str
interface_description: str
@@ -43,7 +43,7 @@ class LanSwitchInterconnectRouterSideBlockInactive(
lifecycle=[SubscriptionLifecycle.INITIAL],
product_block_name="LanSwitchInterconnectRouterSideBlock",
):
- """An inactive :term:`LAN` Switch Interconnect router side."""
+ """An inactive LAN Switch Interconnect router side."""
node: RouterBlockInactive
ae_iface: str | None = None
@@ -54,7 +54,7 @@ class LanSwitchInterconnectRouterSideBlockInactive(
class LanSwitchInterconnectRouterSideBlockProvisioning(
LanSwitchInterconnectRouterSideBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
):
- """A :term:`LAN` Switch Interconnect router side that is being provisioned."""
+ """A LAN Switch Interconnect router side that is being provisioned."""
node: RouterBlockProvisioning
ae_iface: str | None = None
@@ -65,7 +65,7 @@ class LanSwitchInterconnectRouterSideBlockProvisioning(
class LanSwitchInterconnectRouterSideBlock(
LanSwitchInterconnectRouterSideBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]
):
- """An active :term:`LAN` Switch Interconnect router side."""
+ """An active LAN Switch Interconnect router side."""
node: RouterBlock
ae_iface: str
@@ -78,7 +78,7 @@ class LanSwitchInterconnectSwitchSideBlockInactive(
lifecycle=[SubscriptionLifecycle.INITIAL],
product_block_name="LanSwitchInterconnectSwitchSideBlock",
):
- """An inactive :term:`LAN` Switch Interconnect switch side."""
+ """An inactive LAN Switch Interconnect switch side."""
switch: SwitchBlockInactive
ae_iface: str | None = None
@@ -89,7 +89,7 @@ class LanSwitchInterconnectSwitchSideBlockInactive(
class LanSwitchInterconnectSwitchSideBlockProvisioning(
LanSwitchInterconnectSwitchSideBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
):
- """A :term:`LAN` Switch Interconnect switch side that is being provisioned."""
+ """A LAN Switch Interconnect switch side that is being provisioned."""
switch: SwitchBlockProvisioning
ae_iface: str | None = None
@@ -100,7 +100,7 @@ class LanSwitchInterconnectSwitchSideBlockProvisioning(
class LanSwitchInterconnectSwitchSideBlock(
LanSwitchInterconnectSwitchSideBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]
):
- """An active :term:`LAN` Switch Interconnect switch side."""
+ """An active LAN Switch Interconnect switch side."""
switch: SwitchBlock
ae_iface: str
@@ -113,7 +113,7 @@ class LanSwitchInterconnectBlockInactive(
lifecycle=[SubscriptionLifecycle.INITIAL],
product_block_name="LanSwitchInterconnectBlock",
):
- """A :term:`LAN` Switch Interconnect that's currently inactive, see :class:`LanSwitchInterconnectBlock`."""
+ """A LAN Switch Interconnect that's currently inactive, see ``LanSwitchInterconnectBlock``."""
lan_switch_interconnect_description: str | None = None
lan_switch_interconnect_ip_network: IPv4NetworkType | None = None
@@ -126,7 +126,7 @@ class LanSwitchInterconnectBlockInactive(
class LanSwitchInterconnectBlockProvisioning(
LanSwitchInterconnectBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
):
- """A :term:`LAN` Switch Interconnect that's currently being provisioned, see :class:`LanSwitchInterconnectBlock`."""
+ """A LAN Switch Interconnect that's currently being provisioned, see ``LanSwitchInterconnectBlock``."""
lan_switch_interconnect_description: str | None = None
lan_switch_interconnect_ip_network: IPv4NetworkType | None
@@ -137,17 +137,17 @@ class LanSwitchInterconnectBlockProvisioning(
class LanSwitchInterconnectBlock(LanSwitchInterconnectBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
- """A :term:`LAN` Switch Interconnect that's currently deployed in the network."""
+ """A LAN Switch Interconnect that's currently deployed in the network."""
- #: A human-readable description of this :term:`LAN` Switch Interconnect.
+ #: A human-readable description of this LAN Switch Interconnect.
lan_switch_interconnect_description: str
- #: The :term:`IP` resources for this :term:`VLAN` Switch Interconnect.
+ #: The IP resources for this VLAN Switch Interconnect.
lan_switch_interconnect_ip_network: IPv4NetworkType | None
- #: The address space of the :term:`VLAN` Switch Interconnect. It can be private or public.
+ #: The address space of the VLAN Switch Interconnect. It can be private or public.
address_space: AddressSpace
- #: The minimum amount of links the :term:`LAN` Switch Interconnect should consist of.
+ #: The minimum amount of links the LAN Switch Interconnect should consist of.
minimum_links: int
- #: The router side of the :term:`LAN` Switch Interconnect.
+ #: The router side of the LAN Switch Interconnect.
router_side: LanSwitchInterconnectRouterSideBlock
- #: The switch side of the :term:`LAN` Switch Interconnect.
+ #: The switch side of the LAN Switch Interconnect.
switch_side: LanSwitchInterconnectSwitchSideBlock
diff --git a/gso/products/product_blocks/layer_2_circuit.py b/gso/products/product_blocks/layer_2_circuit.py
index 261caa48..40968dfe 100644
--- a/gso/products/product_blocks/layer_2_circuit.py
+++ b/gso/products/product_blocks/layer_2_circuit.py
@@ -69,7 +69,7 @@ class Layer2CircuitType(strEnum):
class Layer2CircuitBlockInactive(
ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="Layer2CircuitBlock"
):
- """An inactive Layer 2 Circuit, see :class:`Layer2CircuitBlock`."""
+ """An inactive Layer 2 Circuit, see ``Layer2CircuitBlock``."""
layer_2_circuit_sides: Layer2CircuitSides[Layer2CircuitSideBlockInactive]
virtual_circuit_id: VC_ID | None = None
@@ -82,7 +82,7 @@ class Layer2CircuitBlockInactive(
class Layer2CircuitBlockProvisioning(Layer2CircuitBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """A provisioning Layer 2 Circuit, see :class:`Layer2CircuitBlock`."""
+ """A provisioning Layer 2 Circuit, see ``Layer2CircuitBlock``."""
layer_2_circuit_sides: Layer2CircuitSides[Layer2CircuitSideBlockProvisioning]
virtual_circuit_id: VC_ID
@@ -103,7 +103,7 @@ class Layer2CircuitBlock(Layer2CircuitBlockProvisioning, lifecycle=[Subscription
virtual_circuit_id: VC_ID
#: The type of circuit, can be tagged or untagged.
layer_2_circuit_type: Layer2CircuitType
- #: If tagged, the lower and upper bounds will set the :term:`VLAN` range.
+ #: If tagged, the lower and upper bounds will set the VLAN range.
vlan_range_lower_bound: VLAN_ID | None
#: Lower and Upper bounds are including.
vlan_range_upper_bound: VLAN_ID | None
diff --git a/gso/products/product_blocks/office_router.py b/gso/products/product_blocks/office_router.py
index bff50dd3..0a16b4f4 100644
--- a/gso/products/product_blocks/office_router.py
+++ b/gso/products/product_blocks/office_router.py
@@ -1,4 +1,4 @@
-"""Product block for :class:`office router` products."""
+"""Product block for office router products."""
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle
@@ -17,7 +17,7 @@ class OfficeRouterBlockInactive(
lifecycle=[SubscriptionLifecycle.INITIAL],
product_block_name="OfficeRouterBlock",
):
- """An office router that's being currently inactive. See :class:`OfficeRouterBlock`."""
+ """An office router that's being currently inactive. See ``OfficeRouterBlock``."""
office_router_fqdn: str | None = None
office_router_ts_port: PortNumber | None = None
@@ -28,7 +28,7 @@ class OfficeRouterBlockInactive(
class OfficeRouterBlockProvisioning(OfficeRouterBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """An office router that's being provisioned. See :class:`RouterBlock`."""
+ """An office router that's being provisioned. See ``RouterBlock``."""
office_router_fqdn: str | None = None
office_router_ts_port: PortNumber | None = None
@@ -49,7 +49,7 @@ class OfficeRouterBlock(OfficeRouterBlockProvisioning, lifecycle=[SubscriptionLi
office_router_lo_ipv4_address: IPv4AddressType
#: The IPv6 loopback address of the office router.
office_router_lo_ipv6_address: IPv6AddressType
- #: The :class:`Site` that this office router resides in. Both physically and computationally.
+ #: The ``Site`` that this office router resides in. Both physically and computationally.
office_router_site: SiteBlock
#: The vendor of an office router. Defaults to Juniper.
vendor: Vendor = Vendor.JUNIPER
diff --git a/gso/products/product_blocks/opengear.py b/gso/products/product_blocks/opengear.py
index 14a4db84..8fd88b31 100644
--- a/gso/products/product_blocks/opengear.py
+++ b/gso/products/product_blocks/opengear.py
@@ -1,4 +1,4 @@
-"""Product block for :class:`Opengear` products."""
+"""Product block for ``Opengear`` products."""
import ipaddress
@@ -17,7 +17,7 @@ class OpengearBlockInactive(
lifecycle=[SubscriptionLifecycle.INITIAL],
product_block_name="OpengearBlock",
):
- """An Opengear that's being currently inactive. See :class:`OpengearBlock`."""
+ """An Opengear that's being currently inactive. See ``OpengearBlock``."""
opengear_hostname: str | None = None
opengear_site: SiteBlockInactive | None = None
@@ -27,7 +27,7 @@ class OpengearBlockInactive(
class OpengearBlockProvisioning(OpengearBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """An Opengear that's being provisioned. See :class:`OpengearBlock`."""
+ """An Opengear that's being provisioned. See ``OpengearBlock``."""
opengear_hostname: str
opengear_site: SiteBlockProvisioning
@@ -43,9 +43,9 @@ class OpengearBlock(OpengearBlockProvisioning, lifecycle=[SubscriptionLifecycle.
opengear_hostname: str
#: The site where the Opengear device is located.
opengear_site: SiteBlock
- #: The :term:`WAN` address of the Opengear device.
+ #: The WAN address of the Opengear device.
opengear_wan_address: ipaddress.IPv4Address
- #: The :term:`WAN` netmask of the Opengear device.
+ #: The WAN netmask of the Opengear device.
opengear_wan_netmask: ipaddress.IPv4Address
- #: The :term:`WAN` gateway of the Opengear device.
+ #: The WAN gateway of the Opengear device.
opengear_wan_gateway: ipaddress.IPv4Address
diff --git a/gso/products/product_blocks/pop_vlan.py b/gso/products/product_blocks/pop_vlan.py
index e90d6655..f4ee9c40 100644
--- a/gso/products/product_blocks/pop_vlan.py
+++ b/gso/products/product_blocks/pop_vlan.py
@@ -32,7 +32,7 @@ PortList = Annotated[list[T], AfterValidator(validate_unique_list), Doc("A list
class PopVlanPortBlockInactive(
ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="PopVlanPortBlock"
):
- """An inactive Pop VLAN port."""
+ """An inactive PoP VLAN port."""
port_name: str | None = None
port_description: str | None = None
@@ -40,7 +40,7 @@ class PopVlanPortBlockInactive(
class PopVlanPortBlockProvisioning(PopVlanPortBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """A Pop VLAN port that is being provisioned."""
+ """A PoP VLAN port that is being provisioned."""
port_name: str | None
port_description: str | None
@@ -48,7 +48,7 @@ class PopVlanPortBlockProvisioning(PopVlanPortBlockInactive, lifecycle=[Subscrip
class PopVlanPortBlock(PopVlanPortBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
- """An active Pop :term:`VLAN` port."""
+ """An active PoP VLAN port."""
port_name: str
port_description: str
@@ -60,7 +60,7 @@ class PopVlanBlockInactive(
lifecycle=[SubscriptionLifecycle.INITIAL],
product_block_name="PopVlanBlock",
):
- """A Pop :term:`VLAN` that's currently inactive, see :class:`PopVlanBlock`."""
+ """A PoP VLAN that's currently inactive, see ``PopVlanBlock``."""
vlan_id: int
pop_vlan_description: str | None = None
@@ -72,7 +72,7 @@ class PopVlanBlockInactive(
class PopVlanBlockProvisioning(PopVlanBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """A Pop :term:`VLAN` that's currently being provisioned, see :class:`PopVlanBlock`."""
+ """A Pop VLAN that's currently being provisioned, see ``PopVlanBlock``."""
vlan_id: int
pop_vlan_description: str | None
@@ -84,19 +84,19 @@ class PopVlanBlockProvisioning(PopVlanBlockInactive, lifecycle=[SubscriptionLife
class PopVlanBlock(PopVlanBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
- """A Pop :term:`VLAN` that's currently deployed in the network."""
+ """A Pop VLAN that's currently deployed in the network."""
- #: The :term:`VLAN` ID of the Pop :term:`VLAN`.
+ #: The VLAN ID of the Pop VLAN.
vlan_id: int
- #: The description of the Pop :term:`VLAN`.
+ #: The description of the Pop VLAN.
pop_vlan_description: str
- #: The :term:`LAN` Switch Interconnect that this Pop :term:`VLAN` is connected to.
+ #: The LAN Switch Interconnect that this Pop VLAN is connected to.
lan_switch_interconnect: LanSwitchInterconnectBlock
- #: The ports of the Pop :term:`VLAN`.
+ #: The ports of the Pop VLAN.
ports: PortList[PopVlanPortBlock] # type: ignore[assignment]
- #: The level of the layer preference for the Pop :term:`VLAN` (L2 or L3).
+ #: The level of the layer preference for the Pop VLAN (L2 or L3).
layer_preference: LayerPreference
- #: IPv4 network for the Pop :term:`VLAN` if layer preference is L3.
+ #: IPv4 network for the Pop VLAN if layer preference is L3.
ipv4_network: IPv4Network | None
- #: IPv6 network for the Pop :term:`VLAN` if layer preference is L3.
+ #: IPv6 network for the Pop VLAN if layer preference is L3.
ipv6_network: IPv6Network | None
diff --git a/gso/products/product_blocks/router.py b/gso/products/product_blocks/router.py
index 7f64c3f5..5ffc2ef9 100644
--- a/gso/products/product_blocks/router.py
+++ b/gso/products/product_blocks/router.py
@@ -1,4 +1,4 @@
-"""Product block for :class:`Router` products."""
+"""Product block for ``Router`` products."""
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle, strEnum
@@ -25,7 +25,7 @@ class RouterBlockInactive(
lifecycle=[SubscriptionLifecycle.INITIAL],
product_block_name="RouterBlock",
):
- """A router that's being currently inactive. See :class:`RouterBlock`."""
+ """A router that's being currently inactive. See ``RouterBlock``."""
router_fqdn: str | None = None
router_ts_port: PortNumber | None = None
@@ -39,7 +39,7 @@ class RouterBlockInactive(
class RouterBlockProvisioning(RouterBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """A router that's being provisioned. See :class:`RouterBlock`."""
+ """A router that's being provisioned. See ``RouterBlock``."""
router_fqdn: str
router_ts_port: PortNumber
@@ -67,9 +67,9 @@ class RouterBlock(RouterBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTI
router_lo_ipv6_address: IPv6AddressType
#: The ISO NET of the router, used for ISIS support.
router_lo_iso_address: str
- #: The role of the router, which can be any of the values defined in :class:`RouterRole`.
+ #: The role of the router, which can be any of the values defined in ``RouterRole``.
router_role: RouterRole
- #: The :class:`Site` that this router resides in. Both physically and computationally.
+ #: The ``Site`` that this router resides in. Both physically and computationally.
router_site: SiteBlock
#: The vendor of a router.
vendor: Vendor
diff --git a/gso/products/product_blocks/service_binding_port.py b/gso/products/product_blocks/service_binding_port.py
index 2c9cd2ce..6fd7467c 100644
--- a/gso/products/product_blocks/service_binding_port.py
+++ b/gso/products/product_blocks/service_binding_port.py
@@ -1,6 +1,6 @@
"""Service Binding Port.
-A service binding port is used to logically attach an edge port to a customer service using a :term:`VLAN`.
+A service binding port is used to logically attach an edge port to a customer service using a VLAN.
"""
from orchestrator.domain.base import ProductBlockModel
@@ -21,7 +21,7 @@ from gso.utils.types.virtual_identifiers import VLAN_ID
class BFDSettingsInactive(
ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="BFDSettings"
):
- """Settings for :term:`BFD`, see :class:`BFDSettings`."""
+ """Settings for BFD, see ``BFDSettings``."""
bfd_enabled: bool = False
bfd_interval_rx: int | None = None
@@ -30,7 +30,7 @@ class BFDSettingsInactive(
class BFDSettingsProvisioning(BFDSettingsInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """Settings for :term:`BFD`, see :class:`BFDSettings`."""
+ """Settings for BFD, see ``BFDSettings``."""
bfd_enabled: bool
bfd_interval_rx: int | None = None
@@ -39,22 +39,22 @@ class BFDSettingsProvisioning(BFDSettingsInactive, lifecycle=[SubscriptionLifecy
class BFDSettings(BFDSettingsProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
- """A set of settings for :term:`BFD`."""
+ """A set of settings for BFD."""
- #: Whether :term:`BFD` is enabled.
+ #: Whether BFD is enabled.
bfd_enabled: bool
- #: The :term:`BFD` interval RX, if enabled.
+ #: The BFD interval RX, if enabled.
bfd_interval_rx: int | None
- #: The :term:`BFD` interval TX, if enabled.
+ #: The BFD interval TX, if enabled.
bfd_interval_tx: int | None
- #: The :term:`BFD` multiplier, if enabled.
+ #: The BFD multiplier, if enabled.
bfd_multiplier: int | None
class ServiceBindingPortInactive(
ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="ServiceBindingPort"
):
- """A Service Binding Port that's currently inactive. See :class:`ServiceBindingPort`."""
+ """A Service Binding Port that's currently inactive. See ``ServiceBindingPort``."""
is_tagged: bool | None = None
vlan_id: VLAN_ID | None = None
@@ -72,7 +72,7 @@ class ServiceBindingPortInactive(
class ServiceBindingPortProvisioning(ServiceBindingPortInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """A Service Binding Port that's currently being provisioned. See :class:`ServiceBindingPort`."""
+ """A Service Binding Port that's currently being provisioned. See ``ServiceBindingPort``."""
is_tagged: bool
vlan_id: VLAN_ID | None = None
@@ -92,9 +92,9 @@ class ServiceBindingPortProvisioning(ServiceBindingPortInactive, lifecycle=[Subs
class ServiceBindingPort(ServiceBindingPortProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
"""A service binding port that is actively used in the network."""
- #: Whether this :term:`VLAN` is tagged or not.
+ #: Whether this VLAN is tagged or not.
is_tagged: bool
- #: The :term:`VLAN` ID.
+ #: The VLAN ID.
vlan_id: VLAN_ID | None = None
#: Is this service binding port layer 2 or 3?
sbp_type: SBPType
@@ -110,11 +110,11 @@ class ServiceBindingPort(ServiceBindingPortProvisioning, lifecycle=[Subscription
custom_firewall_filters: bool
#: The GÉANT service ID of this binding port.
geant_sid: str
- #: The :term:`BGP` sessions associated with this service binding port.
+ #: The BGP sessions associated with this service binding port.
bgp_session_list: list[BGPSession] # type: ignore[assignment]
- #: The Edge Port on which this :term:`SBP` resides.
+ #: The Edge Port on which this SBP resides.
edge_port: EdgePortBlock
- #: :term:`BFD` settings for IPv4
+ #: BFD settings for IPv4
v4_bfd_settings: BFDSettings
- #: :term:`BFD` settings for IPv6
+ #: BFD settings for IPv6
v6_bfd_settings: BFDSettings
diff --git a/gso/products/product_blocks/site.py b/gso/products/product_blocks/site.py
index e4d7231e..11c8e336 100644
--- a/gso/products/product_blocks/site.py
+++ b/gso/products/product_blocks/site.py
@@ -26,7 +26,7 @@ class SiteBlockInactive(
lifecycle=[SubscriptionLifecycle.INITIAL],
product_block_name="SiteBlock",
):
- """A site that's currently inactive, see :class:`SiteBlock`."""
+ """A site that's currently inactive, see ``SiteBlock``."""
site_name: SiteName | None = None
site_city: str | None = None
@@ -41,7 +41,7 @@ class SiteBlockInactive(
class SiteBlockProvisioning(SiteBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """A site that's currently being provisioned, see :class:`SiteBlock`."""
+ """A site that's currently being provisioned, see ``SiteBlock``."""
site_name: SiteName
site_city: str
@@ -76,7 +76,7 @@ class SiteBlock(SiteBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE])
site_internal_id: int
#: The BGP community ID of a site, used to advertise routes learned at this site.
site_bgp_community_id: int
- #: The tier of a site, as described in :class:`SiteTier`.
+ #: The tier of a site, as described in ``SiteTier``.
site_tier: SiteTier
#: The address of the terminal server that this router is connected to. The terminal server provides out of band
#: access. This is required in case a link goes down, or when a router is initially added to the network and it
diff --git a/gso/products/product_blocks/super_pop_switch.py b/gso/products/product_blocks/super_pop_switch.py
index 872f37b5..63f16192 100644
--- a/gso/products/product_blocks/super_pop_switch.py
+++ b/gso/products/product_blocks/super_pop_switch.py
@@ -1,4 +1,4 @@
-"""Product block for :class:`Super PoP Switch` products."""
+"""Product block for Super PoP Switch products."""
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle
@@ -17,7 +17,7 @@ class SuperPopSwitchBlockInactive(
lifecycle=[SubscriptionLifecycle.INITIAL],
product_block_name="SuperPopSwitchBlock",
):
- """A Super PoP switch that's being currently inactive. See :class:`SuperPopSwitchBlock`."""
+ """A Super PoP switch that's being currently inactive. See ``SuperPoPSwitchBlock``."""
super_pop_switch_fqdn: str | None = None
super_pop_switch_ts_port: PortNumber | None = None
@@ -27,7 +27,7 @@ class SuperPopSwitchBlockInactive(
class SuperPopSwitchBlockProvisioning(SuperPopSwitchBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """A Super PoP switch that's being provisioned. See :class:`SuperPopSwitchBlock`."""
+ """A Super PoP switch that's being provisioned. See ``SuperPoPSwitchBlock``."""
super_pop_switch_fqdn: str | None = None
super_pop_switch_ts_port: PortNumber | None = None
@@ -45,7 +45,7 @@ class SuperPopSwitchBlock(SuperPopSwitchBlockProvisioning, lifecycle=[Subscripti
super_pop_switch_ts_port: PortNumber
#: The IPv4 management address of the Super PoP switch.
super_pop_switch_mgmt_ipv4_address: IPv4AddressType
- #: The :class:`Site` that this Super PoP switch resides in. Both physically and computationally.
+ #: The ``Site`` that this Super PoP switch resides in. Both physically and computationally.
super_pop_switch_site: SiteBlock
#: The vendor of a Super PoP switch. Defaults to Juniper.
vendor: Vendor = Vendor.JUNIPER
diff --git a/gso/products/product_blocks/switch.py b/gso/products/product_blocks/switch.py
index 790807ad..f6bdb877 100644
--- a/gso/products/product_blocks/switch.py
+++ b/gso/products/product_blocks/switch.py
@@ -1,4 +1,4 @@
-"""Product block for :class:`Switch` products."""
+"""Product block for ``Switch`` products."""
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle
@@ -24,7 +24,7 @@ class SwitchBlockInactive(
lifecycle=[SubscriptionLifecycle.INITIAL],
product_block_name="SwitchBlock",
):
- """A switch that's being currently inactive. See :class:`SwitchBlock`."""
+ """A switch that's being currently inactive. See ``SwitchBlock``."""
fqdn: str | None = None
ts_port: PortNumber | None = None
@@ -34,7 +34,7 @@ class SwitchBlockInactive(
class SwitchBlockProvisioning(SwitchBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """A switch that's being provisioned. See :class:`SwitchBlock`."""
+ """A switch that's being provisioned. See ``SwitchBlock``."""
fqdn: str
ts_port: PortNumber
@@ -46,11 +46,11 @@ class SwitchBlockProvisioning(SwitchBlockInactive, lifecycle=[SubscriptionLifecy
class SwitchBlock(SwitchBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
"""A switch that's currently deployed in the network."""
- #: The :term:`FQDN` of the switch.
+ #: The FQDN of the switch.
fqdn: str
#: The port of the terminal server that this switch is connected to. Used to offer out of band access.
ts_port: PortNumber
- #: The :class:`Site` that this switch resides in. Both physically and computationally.
+ #: The ``Site`` that this switch resides in. Both physically and computationally.
site: SiteBlock
#: The vendor of the switch.
switch_vendor: Vendor
diff --git a/gso/products/product_types/lan_switch_interconnect.py b/gso/products/product_types/lan_switch_interconnect.py
index 0f83fe37..51fb620b 100644
--- a/gso/products/product_types/lan_switch_interconnect.py
+++ b/gso/products/product_types/lan_switch_interconnect.py
@@ -29,7 +29,7 @@ class LanSwitchInterconnect(LanSwitchInterconnectProvisioning, lifecycle=[Subscr
class ImportedLanSwitchInterconnectInactive(SubscriptionModel, is_base=True):
- """An imported, inactive :term:`LAN` Switch Interconnect."""
+ """An imported, inactive LAN Switch Interconnect."""
lan_switch_interconnect: LanSwitchInterconnectBlockInactive
@@ -37,6 +37,6 @@ class ImportedLanSwitchInterconnectInactive(SubscriptionModel, is_base=True):
class ImportedLanSwitchInterconnect(
ImportedLanSwitchInterconnectInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE]
):
- """An imported :term:`LAN` Switch Interconnect."""
+ """An imported LAN Switch Interconnect."""
lan_switch_interconnect: LanSwitchInterconnectBlock
diff --git a/gso/products/product_types/pop_vlan.py b/gso/products/product_types/pop_vlan.py
index 7567f4ea..ffb6e3c6 100644
--- a/gso/products/product_types/pop_vlan.py
+++ b/gso/products/product_types/pop_vlan.py
@@ -1,4 +1,4 @@
-"""The product type for Pop VLAN."""
+"""The product type for PoP VLAN."""
from orchestrator.domain.base import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle
@@ -7,13 +7,13 @@ from gso.products.product_blocks.pop_vlan import PopVlanBlock, PopVlanBlockInact
class PopVlanInactive(SubscriptionModel, is_base=True):
- """A Pop VLAN that is inactive."""
+ """A PoP VLAN that is inactive."""
pop_vlan: PopVlanBlockInactive
class PopVlanProvisioning(PopVlanInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
- """A Pop VLAN that is being provisioned."""
+ """A PoP VLAN that is being provisioned."""
pop_vlan: PopVlanBlockProvisioning
diff --git a/gso/schedules/scheduling.py b/gso/schedules/scheduling.py
index b688ecbe..508af69e 100644
--- a/gso/schedules/scheduling.py
+++ b/gso/schedules/scheduling.py
@@ -24,17 +24,15 @@ class CronScheduleConfig(BaseModel):
def scheduler(cron_scheduler_config: CronScheduleConfig) -> Callable[[Callable], Callable]:
"""Schedule a Celery task using crontab-like timing.
- Examples
- --------
- - ``minute='*/15'``: Run every 15 minutes.
- - ``hour='*/3'``: Run every 3 hours.
- - ``day_of_week='mon-fri'``: Run on weekdays only.
- - ``day_of_month='1-7,15-21'``: Run on the first and third weeks of the month.
- - ``month_of_year='*/3'``: Run on the first month of each quarter.
+ Examples:
+ - ``minute='*/15'``: Run every 15 minutes.
+ - ``hour='*/3'``: Run every 3 hours.
+ - ``day_of_week='mon-fri'``: Run on weekdays only.
+ - ``day_of_month='1-7,15-21'``: Run on the first and third weeks of the month.
+ - ``month_of_year='*/3'``: Run on the first month of each quarter.
All time units can be specified with lists of numbers or crontab pattern strings for advanced scheduling.
All specified time parts (minute, hour, day, etc.) must align for a task to run.
-
"""
def decorator(task_func: Callable) -> Callable:
diff --git a/gso/services/infoblox.py b/gso/services/infoblox.py
index 3139eab4..cad3fb43 100644
--- a/gso/services/infoblox.py
+++ b/gso/services/infoblox.py
@@ -23,8 +23,8 @@ class AllocationError(Exception):
def _setup_connection() -> tuple[connector.Connector, IPAMParams]:
"""Set up a new connection with an Infoblox instance.
- :return: A tuple that has an Infoblox ``Connector`` instance, and IPAM parameters.
- :rtype: tuple[:class:`Connector`, IPAMParams]
+ Returns:
+ A tuple that has an Infoblox ``Connector`` instance, and IPAM parameters.
"""
oss = load_oss_params().IPAM
options = {
@@ -37,7 +37,7 @@ def _setup_connection() -> tuple[connector.Connector, IPAMParams]:
return connector.Connector(options), oss
-def _allocate_network(
+def _allocate_network( # noqa: PLR0917
conn: connector.Connector,
dns_view: str,
network_view: str,
@@ -48,15 +48,15 @@ def _allocate_network(
"""Allocate a new network in Infoblox.
The function will go over all given containers, and try to allocate a network within the available IP space. If no
- space is available, this method raises an :class:`AllocationError`.
-
- :param :class:`infoblox_client.connector.Connector` conn: An active Infoblox connection.
- :param str dns_view: The Infoblox ``dns_view`` in which the network should be allocated.
- :param str network_view: The Infoblox ``network_view`` where the network should be allocated.
- :param int netmask: The netmask of the desired network. Can be up to 32 for v4 networks, and 128 for v6 networks.
- :param list [str] containers: A list of network containers in which the network should be allocated, given in
- CIDR notation.
- :param str comment: Optionally, a comment can be added to the network allocation.
+ space is available, this method raises an ``AllocationError``.
+
+ Args:
+ conn: An active Infoblox connection.
+ dns_view: The Infoblox ``dns_view`` in which the network should be allocated.
+ network_view: The Infoblox ``network_view`` where the network should be allocated.
+ netmask: The netmask of the desired network. Can be up to 32 for v4 networks, and 128 for v6 networks.
+ containers: A list of network containers in which the network should be allocated, given in CIDR notation.
+ comment: Optionally, a comment can be added to the network allocation.
"""
for container in [ipaddress.ip_network(con) for con in containers]:
for network in container.subnets(new_prefix=netmask):
@@ -76,14 +76,14 @@ def _allocate_network(
def hostname_available(hostname: str) -> bool:
"""Check whether a hostname is still available **in Infoblox**.
- Check whether Infoblox already has a :class:`infoblox_client.objects.HostRecord` that matches the given hostname.
+ Check whether Infoblox already has a ``infoblox_client.objects.HostRecord`` that matches the given hostname.
.. warning::
This method only checks within the Infoblox instance, and not the rest of the internet. The hostname could
therefore still be taken elsewhere.
- :param hostname: The hostname to be checked.
- :type hostname: str
+ Args:
+ hostname: The hostname to be checked.
"""
conn, _ = _setup_connection()
return objects.HostRecord.search(conn, name=hostname) is None
@@ -92,13 +92,12 @@ def hostname_available(hostname: str) -> bool:
def allocate_v4_network(service_type: str, comment: str | None = "") -> ipaddress.IPv4Network:
"""Allocate a new IPv4 network in Infoblox.
- Allocate an IPv4 network for a specific service type. The service type should be defined in the OSS
- parameters of :term:`GSO`, from which the containers and netmask will be used.
+ Allocate an IPv4 network for a specific service type. The service type should be defined in the OSS parameters of
+ GSO, from which the containers and netmask will be used.
- :param service_type: The service type for which the network is allocated.
- :type service_type: str
- :param comment: A comment to be added to the allocated network in Infoblox.
- :type comment: str, optional
+ Args:
+ service_type: The service type for which the network is allocated.
+ comment: A comment to be added to the allocated network in Infoblox.
"""
conn, oss = _setup_connection()
netmask = getattr(oss, service_type).V4.mask
@@ -112,13 +111,12 @@ def allocate_v4_network(service_type: str, comment: str | None = "") -> ipaddres
def allocate_v6_network(service_type: str, comment: str | None = "") -> ipaddress.IPv6Network:
"""Allocate a new IPv6 network in Infoblox.
- Allocate an IPv6 network for a specific service type. The service type should be defined in the :term:`OSS`
- parameters of :term:`GSO`, from which the containers and netmask will be used.
+ Allocate an IPv6 network for a specific service type. The service type should be defined in the OSS parameters of
+ GSO, from which the containers and netmask will be used.
- :param service_type: The service type for which the network is allocated.
- :type service_type: str
- :param comment: A comment to be added to the allocated network in Infoblox.
- :type comment: str, optional
+ Args:
+ service_type: The service type for which the network is allocated.
+ comment: A comment to be added to the allocated network in Infoblox.
"""
conn, oss = _setup_connection()
netmask = getattr(oss, service_type).V6.mask
@@ -132,10 +130,10 @@ def allocate_v6_network(service_type: str, comment: str | None = "") -> ipaddres
def find_network_by_cidr(
ip_network: ipaddress.IPv4Network | ipaddress.IPv6Network,
) -> objects.Network | None:
- """Find a network in Infoblox by its :term:`CIDR`.
+ """Find a network in Infoblox by its CIDR.
- :param ip_network: The :term:`CIDR` that is searched.
- :type ip_network: ipaddress.IPv4Network | ipaddress.IPv6Network
+ Args:
+ ip_network: The CIDR that is searched.
"""
conn, _ = _setup_connection()
return objects.Network.search(conn, cidr=str(ip_network))
@@ -144,11 +142,11 @@ def find_network_by_cidr(
def delete_network(ip_network: ipaddress.IPv4Network | ipaddress.IPv6Network) -> None:
"""Delete a network in Infoblox.
- Delete a network that is allocated in Infoblox, by passing the :term:`CIDR` to be deleted. The :term:`CIDR` must
- exactly match an existing entry in Infoblox, otherwise this method raises a :class:`DeletionError`
+ Delete a network that is allocated in Infoblox, by passing the CIDR to be deleted. The CIDR must exactly match an
+ existing entry in Infoblox.
- :param ip_network: The network that should get deleted.
- :type ip_network: ipaddress.IPv4Network | ipaddress.IPv6Network
+ Args:
+ ip_network: The network that should get deleted.
"""
network = find_network_by_cidr(ip_network)
if network:
@@ -165,18 +163,15 @@ def allocate_host(
Create a new host record in Infoblox, by providing a hostname, and the service type that is associated with this new
host. Most likely to be a loopback interface. If the hostname is not available in Infoblox (due to a potential
- collision) this method raises an :class:`AllocationError`.
-
- :param hostname: The :term:`FQDN` of the new host
- :type hostname: str
- :param service_type: The service type from which IP resources should be used.
- :type service_type: str
- :param cname_aliases: A list of any :term:`CNAME` aliases that should be associated with this host. Most often this
- will be a single loopback address.
- :type cname_aliases: list[str]
- :param comment: A comment that is added to the host record in Infoblox, should be the ``subscription_id`` of the new
- :class:`Router` subscription.
- :type comment: str
+ collision) this method raises an ``AllocationError``.
+
+ Args:
+ hostname: The FQDN of the new host
+ service_type: The service type from which IP resources should be used.
+ cname_aliases: A list of any CNAME aliases that should be associated with this host. Most often this will be a
+ single loopback address.
+ comment: A comment that is added to the host record in Infoblox, should be the ``subscription_id`` of the new
+ ``Router`` subscription.
"""
if not hostname_available(hostname):
msg = f"Cannot allocate new host, FQDN {hostname} already taken."
@@ -241,13 +236,13 @@ def create_host_by_ip(
) -> None:
"""Create a new host record with a given IPv4 and IPv6 address.
- :param str hostname: The :term:`FQDN` of the new host.
- :param IPv4Address ipv4_address: The IPv4 address of the new host.
- :param IPv6Address ipv6_address: The IPv6 address of the new host.
- :param str service_type: The relevant service type, used to deduce the correct ``dns_view`` and ``network_view`` in
- Infoblox.
- :param str comment: The comment stored in this Infoblox record, most likely the relevant ``subscription_id`` in
- :term:`GSO`.
+ Args:
+ hostname: The FQDN of the new host.
+ ipv4_address: The IPv4 address of the new host.
+ ipv6_address: The IPv6 address of the new host.
+ service_type: The relevant service type, used to deduce the correct ``dns_view`` and ``network_view`` in
+ Infoblox.
+ comment: The comment stored in this Infoblox record, most likely the relevant ``subscription_id`` in GSO.
"""
if not hostname_available(hostname):
msg = f"FQDN '{hostname}' is already in use, allocation aborted."
@@ -271,8 +266,8 @@ def create_host_by_ip(
def find_host_by_ip(ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> objects.HostRecord | None:
"""Find a host record in Infoblox by its associated IP address.
- :param ip_addr: The IP address of a host that is searched for.
- :type ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address
+ Args:
+ ip_addr: The IP address of a host that is searched for.
"""
conn, _ = _setup_connection()
if ip_addr.version == 4: # noqa: PLR2004, the 4 in IPv4 is well-known and not a "magic value."
@@ -289,10 +284,10 @@ def find_host_by_ip(ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> o
def find_host_by_fqdn(fqdn: str) -> objects.HostRecord:
- """Find a host record by its associated :term:`FQDN`.
+ """Find a host record by its associated FQDN.
- :param fqdn: The :term:`FQDN` of a host that is searched for.
- :type fqdn: str
+ Args:
+ fqdn: The FQDN of a host that is searched for.
"""
conn, _ = _setup_connection()
return objects.HostRecord.search(
@@ -303,10 +298,12 @@ def find_host_by_fqdn(fqdn: str) -> objects.HostRecord:
def find_v6_host_by_fqdn(fqdn: str) -> objects.HostRecordV6:
- """Find a host record by its associated :term:`FQDN`.
+ """Find a host record by its associated FQDN.
This specific method will return the IPv6 variant of a record, if it exists.
- :param str fqdn: The :term:`FQDN` of a host that is searched for.
+
+ Args:
+ fqdn: The FQDN of a host that is searched for.
"""
conn, _ = _setup_connection()
return objects.HostRecordV6.search(
@@ -317,11 +314,10 @@ def find_v6_host_by_fqdn(fqdn: str) -> objects.HostRecordV6:
def delete_host_by_ip(ip_addr: IPv4AddressType | ipaddress.IPv6Address) -> None:
"""Delete a host from Infoblox.
- Delete a host record in Infoblox, by providing the IP address that is associated with the record. Raises a
- :class:`DeletionError` if no record can be found in Infoblox.
+ Delete a host record in Infoblox, by providing the IP address that is associated with the record.
- :param ip_addr: The IP address of the host record that should get deleted.
- :type ip_addr: IPv4AddressType | ipaddress.IPv6Address
+ Args:
+ ip_addr: The IP address of the host record that should get deleted.
"""
host = find_host_by_ip(ip_addr)
if host:
@@ -334,11 +330,10 @@ def delete_host_by_ip(ip_addr: IPv4AddressType | ipaddress.IPv6Address) -> None:
def delete_host_by_fqdn(fqdn: str) -> None:
"""Delete a host from Infoblox.
- Delete a host record in Infoblox, by providing the :term:`FQDN` that is associated with the record. Raises a
- :class:`DeletionError` if no record can be found in Infoblox.
+ Delete a host record in Infoblox, by providing the FQDN that is associated with the record.
- :param fqdn: The :term:`FQDN` of the host record that should get deleted.
- :type fqdn: str
+ Args:
+ fqdn: The FQDN of the host record that should get deleted.
"""
host = find_host_by_fqdn(fqdn)
if host:
diff --git a/gso/services/kentik_client.py b/gso/services/kentik_client.py
index a958aca5..cb9c3093 100644
--- a/gso/services/kentik_client.py
+++ b/gso/services/kentik_client.py
@@ -61,7 +61,8 @@ class KentikClient:
def get_devices(self) -> list[dict[str, Any]]:
"""List all devices in Kentik.
- Returns a list of shape ``[{**device_1}, {**device_2}, ..., {**device_n}]}``.
+ Returns:
+ a list of shape ``[{**device_1}, {**device_2}, ..., {**device_n}]}``.
"""
return self._send_request("GET", "v5/devices")["devices"]
@@ -73,11 +74,12 @@ class KentikClient:
return device
def get_device_by_name(self, device_name: str) -> dict[str, Any]:
- """Fetch a device in Kentik by its :term:`FQDN`.
+ """Fetch a device in Kentik by its FQDN.
If the device is not found, returns an empty dict.
- :param str device_name: The :term:`FQDN` of the sought device.
+ Args:
+ device_name: The FQDN of the sought device.
"""
devices = self.get_devices()
for device in devices:
@@ -101,7 +103,8 @@ class KentikClient:
.. vale off
- :param str site_slug: The name of the site, should be a three-letter slug like COR or POZ.
+ Args:
+ site_slug: The name of the site, should be a three-letter slug like COR or POZ.
"""
sites = self.get_sites()
for site in sites:
@@ -113,39 +116,40 @@ class KentikClient:
def get_plans(self) -> list[dict[str, Any]]:
"""Get all Kentik plans available.
- Returns a list of ``plans`` that each have the following shape:
-
- .. vale off
- .. code-block:: json
-
- "plan": {
- "active": true,
- "bgp_enabled": true,
- "cdate" "1970-01-01T01:01:01.000Z",
- "company_id": 111111,
- "description": "A description of this plan",
- "deviceTypes": [
- {"device_type": "router"},
- {"device_type": "host-nprobe-dns-www"}
- ],
- "devices": [
- {
- "id": "111111",
- "device_name": "rt0.city.tld.internal",
- "device_type": "router"
- },
- ],
- "edate": "2999-01-01T09:09:09.000Z",
- "fast_retention": 10,
- "full_retention": 5,
- "id": 11111,
- "max_bigdata_fps": 100,
- "max_devices": 9001,
- "max_fps": 200,
- "name": "KENTIK-PLAN-01",
- "metadata": {},
- }
- .. vale on
+ Returns:
+ a list of ``plans`` that each have the following shape:
+
+ .. vale off
+ .. code-block:: json
+
+ "plan": {
+ "active": true,
+ "bgp_enabled": true,
+ "cdate" "1970-01-01T01:01:01.000Z",
+ "company_id": 111111,
+ "description": "A description of this plan",
+ "deviceTypes": [
+ {"device_type": "router"},
+ {"device_type": "host-nprobe-dns-www"}
+ ],
+ "devices": [
+ {
+ "id": "111111",
+ "device_name": "rt0.city.tld.internal",
+ "device_type": "router"
+ },
+ ],
+ "edate": "2999-01-01T09:09:09.000Z",
+ "fast_retention": 10,
+ "full_retention": 5,
+ "id": 11111,
+ "max_bigdata_fps": 100,
+ "max_devices": 9001,
+ "max_fps": 200,
+ "name": "KENTIK-PLAN-01",
+ "metadata": {},
+ }
+ .. vale on
"""
return self._send_request("GET", "v5/plans")["plans"]
@@ -158,7 +162,8 @@ class KentikClient:
If the plan is not found, returns an empty dict.
- :param str plan_name: The name of the plan.
+ Args:
+ plan_name: The name of the plan.
"""
plans = self.get_plans()
for plan in plans:
@@ -208,8 +213,9 @@ class KentikClient:
def remove_device(self, device_id: str, *, archive: bool) -> None:
"""Remove a device from Kentik.
- :param str device_id: The Kentik internal ID of the device that is to be removed.
- :param bool archive: Archive the device instead of completely deleting it.
+ Args:
+ device_id: The Kentik internal ID of the device that is to be removed.
+ archive: Archive the device instead of completely deleting it.
"""
if not archive:
self._send_delete(f"v5/device/{device_id}")
@@ -217,6 +223,6 @@ class KentikClient:
self._send_delete(f"v5/device/{device_id}")
def remove_device_by_fqdn(self, fqdn: str, *, archive: bool) -> None:
- """Remove a device from Kentik, by its :term:`FQDN`."""
+ """Remove a device from Kentik, by its FQDN."""
device_id = self.get_device_by_name(fqdn)["id"]
self.remove_device(device_id, archive=archive)
diff --git a/gso/services/librenms_client.py b/gso/services/librenms_client.py
index b514f000..0093e6f8 100644
--- a/gso/services/librenms_client.py
+++ b/gso/services/librenms_client.py
@@ -1,4 +1,4 @@
-"""The LibreNMS module interacts with the inventory management system of :term:`GAP`."""
+"""The LibreNMS module interacts with the inventory management system of GAP."""
import logging
from http import HTTPStatus
@@ -48,9 +48,14 @@ class LibreNMSClient:
def get_device(self, fqdn: str) -> dict[str, Any]:
"""Get an existing device from LibreNMS.
- :param str fqdn: The :term:`FQDN` of a device that is retrieved.
- :return dict[str, Any]: A :term:`JSON` formatted list of devices that match the queried :term:`FQDN`.
- :raises HTTPError: Raises an HTTP error 404 when the device is not found
+ Args:
+ fqdn: The FQDN of a device that is retrieved.
+
+ Returns:
+ A JSON formatted list of devices that match the queried FQDN.
+
+ Raises:
+ HTTPError: Raises an HTTP error 404 when the device is not found
"""
response = self._send_request("GET", f"/devices/{fqdn}")
response.raise_for_status()
@@ -60,8 +65,11 @@ class LibreNMSClient:
def device_exists(self, fqdn: str) -> bool:
"""Check whether a device exists in LibreNMS.
- :param str fqdn: The hostname that should be checked for.
- :return bool: Whether the device exists or not.
+ Args:
+ fqdn: The hostname that should be checked for.
+
+ Returns:
+ Whether the device exists or not.
"""
try:
device = self.get_device(fqdn)
@@ -75,9 +83,10 @@ class LibreNMSClient:
def add_device(self, fqdn: str, snmp_version: SNMPVersion) -> dict[str, Any]:
"""Add a new device to LibreNMS.
- :param str fqdn: The hostname of the newly added device.
- :param SNMPVersion snmp_version: The SNMP version of the new device, which decides the authentication parameters
- that LibreNMS should use to poll the device.
+ Args:
+ fqdn: The hostname of the newly added device.
+ snmp_version: The SNMP version of the new device, which decides the authentication parameters that LibreNMS
+ should use to poll the device.
"""
device_data = {
"display": fqdn,
@@ -95,9 +104,14 @@ class LibreNMSClient:
def remove_device(self, fqdn: str) -> dict[str, Any]:
"""Remove a device from LibreNMS.
- :param str fqdn: The :term:`FQDN` of the hostname that should get deleted.
- :return dict[str, Any]: A JSON representation of the device that got removed.
- :raises HTTPError: Raises an exception if the request did not succeed.
+ Args:
+ fqdn: The FQDN of the hostname that should get deleted.
+
+ Returns:
+ A JSON representation of the device that got removed.
+
+ Raises:
+ HTTPError: Raises an exception if the request did not succeed.
"""
device = self._send_request("DELETE", f"/devices/{fqdn}")
device.raise_for_status()
@@ -105,10 +119,13 @@ class LibreNMSClient:
return device.json()
def validate_device(self, fqdn: str) -> str | None:
- """Validate a device in LibreNMS by fetching the record match the queried :term:`FQDN` against its hostname.
+ """Validate a device in LibreNMS by fetching the record match the queried FQDN against its hostname.
+
+ Args:
+ fqdn: The FQDN of the host that is validated.
- :param str fqdn: The :term:`FQDN` of the host that is validated.
- :return list[str]: A list of errors, if empty the device is successfully validated.
+ Returns:
+ A list of errors, if empty the device is successfully validated.
"""
error = None
try:
diff --git a/gso/services/lso_client.py b/gso/services/lso_client.py
index f501ed70..6bd8e41d 100644
--- a/gso/services/lso_client.py
+++ b/gso/services/lso_client.py
@@ -1,6 +1,6 @@
-"""The :term:`LSO` client, which interacts with :term:`LSO` running externally.
+"""The LSO client, which interacts with LSO running externally.
-:term:`LSO` is responsible for executing Ansible playbooks, that deploy subscriptions.
+LSO is responsible for executing Ansible playbooks, that deploy subscriptions.
"""
import json
@@ -36,14 +36,12 @@ LSOState = State # FIXME: Use the above definition when python3.13 is released
def _send_request(parameters: dict, callback_route: str) -> None:
- """Send a request to :term:`LSO`. The callback address is derived using the process ID provided.
-
- :param parameters: JSON body for the request, which will almost always at least consist of a subscription object,
- and a boolean value to indicate a dry run.
- :type parameters: dict
- :param callback_route: The callback route that should be used to resume the workflow.
- :type callback_route: str
- :rtype: None
+ """Send a request to LSO. The callback address is derived using the process ID provided.
+
+ Args:
+ parameters: JSON body for the request, which will almost always at least consist of a subscription object, and a
+ boolean value to indicate a dry run.
+ callback_route: The callback route that should be used to resume the workflow.
"""
oss = settings.load_oss_params()
params = oss.PROVISIONING_PROXY
@@ -90,7 +88,7 @@ def _execute_playbook(
.. warning::
Note the fact that the collection of all hosts is a dictionary, and not a list of strings. Ansible expects each
- host to be a key-value pair. The key is the :term:`FQDN` of a host, and the value always ``null``.
+ host to be a key-value pair. The key is the FQDN of a host, and the value always ``null``.
The extra vars can be a simple dict consisting of key-value pairs, for example:
@@ -102,14 +100,13 @@ def _execute_playbook(
"verb": "deploy"
}
- :param str playbook_name: Filename of the playbook that is to be executed. It must be present on the remote system
- running the provisioning proxy, otherwise it will return an error.
- :param str callback_route: The endpoint at which :term:`GSO` expects a callback to continue the workflow executing
- this step.
- :param dict[str, Any] inventory: An inventory of machines at which the playbook is targeted. Must be in
- YAML-compatible format.
- :param dict[str, Any] extra_vars: Any extra variables that the playbook relies on. This can include a subscription
- object, a boolean value indicating a dry run, a commit comment, etc.
+ Args:
+ playbook_name: Filename of the playbook that is to be executed. It must be present on the remote system running
+ the provisioning proxy, otherwise it will return an error.
+ callback_route: The endpoint at which GSO expects a callback to continue the workflow executing this step.
+ inventory: An inventory of machines at which the playbook is targeted. Must be in YAML-compatible format.
+ extra_vars: Any extra variables that the playbook relies on. This can include a subscription object, a boolean
+ value indicating a dry run, a commit comment, etc.
"""
parameters = {
"playbook_name": playbook_name,
@@ -170,7 +167,7 @@ def _inventory_is_set(state: State) -> bool:
"""Validate whether the passed Ansible inventory is empty.
If the inventory is empty, which can happen in select cases, there should be no playbook run. This conditional will
- prevent from calling out to :term:`LSO` with an empty playbook, which would cause the Ansible runner process to
+ prevent from calling out to LSO with an empty playbook, which would cause the Ansible runner process to
hang. This in turn will result in a workflow step that is never called back to.
"""
if "inventory" not in state:
@@ -187,20 +184,21 @@ _inventory_is_not_empty = conditional(_inventory_is_set)
def lso_interaction(provisioning_step: Step) -> StepList:
- """Interact with the provisioning proxy :term:`LSO` using a callback step.
+ """Interact with the provisioning proxy LSO using a callback step.
An asynchronous interaction with the provisioning proxy. This is an external system that executes Ansible playbooks
to provision service subscriptions. If the playbook fails, this step will also fail, allowing for the user to retry
provisioning from the UI.
Optionally, the keys ``lso_result_title`` and ``lso_result_extra_label`` can be added to the state before running
- this interaction. They will be used to customise the input step that shows the outcome of the :term:`LSO`
+ this interaction. They will be used to customise the input step that shows the outcome of the LSO
interaction.
- :param provisioning_step: A workflow step that performs an operation remotely using the provisioning proxy.
- :type provisioning_step: :class:`Step`
- :return: A list of steps that is executed as part of the workflow.
- :rtype: :class:`StepList`
+ Args:
+ provisioning_step: A workflow step that performs an operation remotely using the provisioning proxy.
+
+ Returns:
+ A list of steps that is executed as part of the workflow.
"""
return (
begin
@@ -218,7 +216,7 @@ def lso_interaction(provisioning_step: Step) -> StepList:
def indifferent_lso_interaction(provisioning_step: Step) -> StepList:
- """Interact with the provisioning proxy :term:`LSO` using a callback step.
+ """Interact with the provisioning proxy LSO using a callback step.
This interaction is identical from the one described in ``lso_interaction()``, with one functional difference.
Whereas the ``lso_interaction()`` will make the workflow step fail on unsuccessful interaction, this step will not.
@@ -228,10 +226,11 @@ def indifferent_lso_interaction(provisioning_step: Step) -> StepList:
Using this interaction requires the operator to carefully evaluate the outcome of a playbook themselves. If a
playbook fails, this will not cause the workflow to fail.
- :param provisioning_step: A workflow step that performs an operation remotely using the provisioning proxy.
- :type provisioning_step: :class:`Step`
- :return: A list of steps that is executed as part of the workflow.
- :rtype: :class:`StepList`
+ Args:
+ provisioning_step: A workflow step that performs an operation remotely using the provisioning proxy.
+
+ Returns:
+ A list of steps that is executed as part of the workflow.
"""
return (
begin
@@ -249,15 +248,16 @@ def indifferent_lso_interaction(provisioning_step: Step) -> StepList:
def anonymous_lso_interaction(provisioning_step: Step, validation_step: Step = _evaluate_results) -> StepList:
- """Interact with the provisioning proxy :term:`LSO` without any user input.
+ """Interact with the provisioning proxy LSO without any user input.
- Similar to the indifferent :term:`LSO` interaction, there also is the anonymous interaction. Output is not ignored
+ Similar to the indifferent LSO interaction, there also is the anonymous interaction. Output is not ignored
but no input step is created to display the results.
A custom validation step may be given as input. This validation step should look inside the ``callback_result`` key
in the current state. By default, only the return code of the playbook execution is evaluated.
- :param Step provisioning_step: A workflow step to remotely provision a subscription.
- :param Step validation_step: An optional validation step which defaults to a step that evaluates the return code.
+ Args:
+ provisioning_step: A workflow step to remotely provision a subscription.
+ validation_step: An optional validation step which defaults to a step that evaluates the return code.
"""
return (
begin
diff --git a/gso/services/mailer.py b/gso/services/mailer.py
index c98ae7f8..b6344b83 100644
--- a/gso/services/mailer.py
+++ b/gso/services/mailer.py
@@ -12,8 +12,9 @@ def send_mail(subject: str, body: str) -> None:
Only supports STARTTLS, not SSL.
- :param str subject: The email subject.
- :param str body: The contents of the email message.
+ Args:
+ subject: The email subject.
+ body: The contents of the email message.
"""
email_params = load_oss_params().EMAIL
msg = EmailMessage()
diff --git a/gso/services/netbox_client.py b/gso/services/netbox_client.py
index 489f95ab..4b6560b9 100644
--- a/gso/services/netbox_client.py
+++ b/gso/services/netbox_client.py
@@ -111,7 +111,9 @@ class NetboxClient:
The type parameter can be 1000base-t, 10gbase-t, LAG, etc.
For more details on type definition have a look in choices.py in the netbox API implementation in module DCIM.
- Returns the new interface object as dict.
+
+ Returns:
+ the new interface object as dict.
"""
device = self.get_device_by_name(device_name)
@@ -214,7 +216,8 @@ class NetboxClient:
) -> Interfaces:
"""Assign a given interface to a LAG.
- Returns the interface object after assignment.
+ Returns:
+ the interface object after assignment.
"""
iface = self.get_interface_by_name_and_device(iface_name, device_name)
diff --git a/gso/services/partners.py b/gso/services/partners.py
index c8cfeb08..160cde09 100644
--- a/gso/services/partners.py
+++ b/gso/services/partners.py
@@ -1,4 +1,4 @@
-"""A module that returns the partners available in :term:`GSO`."""
+"""A module that returns the partners available in GSO."""
from datetime import datetime
from typing import Annotated, Any
@@ -115,8 +115,11 @@ def create_partner(
) -> dict:
"""Create a new partner and add it to the database using Pydantic schema for validation.
- :param partner_data: Partner data validated by Pydantic schema.
- :return: JSON representation of the created partner.
+ Args:
+ partner_data: Partner data validated by Pydantic schema.
+
+ Returns:
+ JSON representation of the created partner.
"""
new_partner = PartnerTable(**partner_data.model_dump())
db.session.add(new_partner)
diff --git a/gso/services/processes.py b/gso/services/processes.py
index f6470a8a..c33f17a2 100644
--- a/gso/services/processes.py
+++ b/gso/services/processes.py
@@ -13,8 +13,8 @@ from sqlalchemy import ScalarResult, or_, select
def count_incomplete_validate_products() -> int:
"""Count the number of incomplete validate_geant_products processes.
- :return: The count of incomplete 'validate_geant_products' processes.
- :rtype: int
+ Returns:
+ The count of incomplete 'validate_geant_products' processes.
"""
return ProcessTable.query.filter(
ProcessTable.workflow_name == "validate_geant_products",
diff --git a/gso/services/sharepoint.py b/gso/services/sharepoint.py
index d0efff0b..e846f78c 100644
--- a/gso/services/sharepoint.py
+++ b/gso/services/sharepoint.py
@@ -40,7 +40,8 @@ class SharePointClient:
def get_list_items(self, list_name: str) -> ListItemCollectionResponse | None:
"""Get list items from a given list in SharePoint.
- :param str list_name: The name of the list.
+ Args:
+ list_name: The name of the list.
"""
async def _get_list_items() -> ListItemCollectionResponse | None:
@@ -59,10 +60,12 @@ class SharePointClient:
def add_list_item(self, list_name: str, fields: dict[str, str]) -> str:
"""Add a new entry to a SharePoint list.
- :param str list_name: The name of the list.
- :param dict[str, str] fields: Any pre-filled fields in the list item. Can be left empty.
+ Args:
+ list_name: The name of the list.
+ fields: Any pre-filled fields in the list item. Can be left empty.
- :return str: The URL of the list in which a new item has been created.
+ Returns:
+ The URL of the list in which a new item has been created.
"""
async def _new_list_item() -> str:
diff --git a/gso/services/subscriptions.py b/gso/services/subscriptions.py
index add27960..9185265c 100644
--- a/gso/services/subscriptions.py
+++ b/gso/services/subscriptions.py
@@ -35,14 +35,15 @@ def get_subscriptions(
) -> list[SubscriptionType]:
"""Retrieve active subscriptions for a specific product type.
- :param list[ProductName] product_types: The types of the product for which to retrieve subscriptions.
- :param SubscriptionLifecycle lifecycles: The lifecycles that the products must be in.
- :param list[str] includes: List of fields to be included in the returned Subscription objects.
- :param list[str] excludes: List of fields to be excluded from the returned Subscription objects.
- :param UUIDstr partner_id: The customer id of subscriptions.
-
- :return: A list of Subscription objects that match the query.
- :rtype: list[Subscription]
+ Args:
+ product_types: The types of the product for which to retrieve subscriptions.
+ lifecycles: The lifecycles that the products must be in.
+ includes: List of fields to be included in the returned Subscription objects.
+ excludes: List of fields to be excluded from the returned Subscription objects.
+ partner_id: The customer id of subscriptions.
+
+ Returns:
+ A list of ``SubscriptionType`` objects that match the query.
"""
if not includes:
includes = [col.name for col in SubscriptionTable.__table__.columns]
@@ -73,11 +74,12 @@ def get_router_subscriptions(
) -> list[SubscriptionType]:
"""Retrieve subscriptions specifically for routers.
- :param includes: The fields to be included in the returned Subscription objects.
- :type includes: list[str]
+ Args:
+ includes: The fields to be included in the returned Subscription objects.
+ lifecycles: The subscription lifecycle states that should be included in the results.
- :return: A list of Subscription objects for routers.
- :rtype: list[Subscription]
+ Returns:
+ A list of Subscription objects for routers.
"""
return get_subscriptions(product_types=[ProductType.ROUTER], lifecycles=lifecycles, includes=includes)
@@ -85,11 +87,11 @@ def get_router_subscriptions(
def get_active_router_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
"""Retrieve active subscriptions specifically for routers.
- :param includes: The fields to be included in the returned Subscription objects.
- :type includes: list[str]
+ Args:
+ includes: The fields to be included in the returned Subscription objects.
- :return: A list of Subscription objects for routers.
- :rtype: list[Subscription]
+ Returns:
+ A list of Subscription objects for routers.
"""
return get_subscriptions(
product_types=[ProductType.ROUTER], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
@@ -99,8 +101,11 @@ def get_active_router_subscriptions(includes: list[str] | None = None) -> list[S
def get_provisioning_router_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
"""Retrieve provisioning subscriptions specifically for routers.
- :param list[str] includes: The fields to be included in the returned Subscription objects.
- :return list[Subscription]: A list of router Subscription objects.
+ Args:
+ includes: The fields to be included in the returned Subscription objects.
+
+ Returns:
+ A list of router Subscription objects.
"""
return get_subscriptions(
product_types=[ProductType.ROUTER], lifecycles=[SubscriptionLifecycle.PROVISIONING], includes=includes
@@ -110,11 +115,11 @@ def get_provisioning_router_subscriptions(includes: list[str] | None = None) ->
def get_active_switch_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
"""Retrieve active subscriptions specifically for switches.
- :param includes: The fields to be included in the returned Subscription objects.
- :type includes: list[str]
+ Args:
+ includes: The fields to be included in the returned Subscription objects.
- :return: A list of Subscription objects for switches.
- :rtype: list[Subscription]
+ Returns:
+ A list of Subscription objects for switches.
"""
return get_subscriptions(
product_types=[ProductType.SWITCH], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
@@ -124,11 +129,11 @@ def get_active_switch_subscriptions(includes: list[str] | None = None) -> list[S
def get_active_iptrunk_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
"""Retrieve active subscriptions specifically for IP trunks.
- :param includes: The fields to be included in the returned Subscription objects.
- :type includes: list[str]
+ Args:
+ includes: The fields to be included in the returned Subscription objects.
- :return: A list of Subscription objects for IP trunks.
- :rtype: list[Subscription]
+ Returns:
+ A list of Subscription objects for IP trunks.
"""
return get_subscriptions(
product_types=[ProductType.IP_TRUNK], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
@@ -138,8 +143,11 @@ def get_active_iptrunk_subscriptions(includes: list[str] | None = None) -> list[
def get_non_terminated_iptrunk_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
"""Retrieve all IP trunk subscriptions that are not terminated.
- :param list[Subscription] includes: Fields to be included in the returned Subscription objects.
- :return list[Subscription]: A list of IP trunk subscriptions.
+ Args:
+ includes: Fields to be included in the returned Subscription objects.
+
+ Returns:
+ A list of IP trunk subscriptions.
"""
return get_subscriptions(
product_types=[ProductType.IP_TRUNK],
@@ -157,11 +165,12 @@ def get_trunks_that_terminate_on_router(
terminate on this Router. The given lifecycle state dictates the state of trunk subscriptions that are counted as
terminating on this router.
- :param UUIDstr subscription_id: Subscription ID of a Router
- :param SubscriptionLifecycle lifecycle_state: Required lifecycle state of the IP trunk
+ Args:
+ subscription_id: Subscription ID of a Router
+ lifecycle_state: Required lifecycle state of the IP trunk
- :return: A list of IP trunk subscriptions
- :rtype: list[SubscriptionTable]
+ Returns:
+ A list of IP trunk subscriptions
"""
return (
query_in_use_by_subscriptions(UUID(subscription_id))
@@ -177,11 +186,11 @@ def get_trunks_that_terminate_on_router(
def get_product_id_by_name(product_name: ProductName) -> UUID:
"""Retrieve the UUID of a product by its name.
- :param product_name: The name of the product.
- :type product_name: ProductName
+ Args:
+ product_name: The name of the product.
- :return UUID: The UUID of the product.
- :rtype: UUID
+ Returns:
+ The UUID of the product.
"""
return ProductTable.query.filter_by(name=product_name).first().product_id
@@ -189,14 +198,12 @@ def get_product_id_by_name(product_name: ProductName) -> UUID:
def get_active_subscriptions_by_field_and_value(field_name: str, field_value: str) -> list[SubscriptionTable]:
"""Retrieve a list of active subscriptions based on a specified field and its value.
- :param field_name: The name of the field to filter by.
- :type field_name: str
-
- :param field_value: The value of the field to match.
- :type field_value: Any
+ Args:
+ field_name: The name of the field to filter by.
+ field_value: The value of the field to match.
- :return: A list of active Subscription objects that match the criteria.
- :rtype: List[SubscriptionTable]
+ Returns:
+ A list of active Subscription objects that match the criteria.
"""
return (
SubscriptionTable.query.join(ProductTable)
@@ -235,11 +242,11 @@ def get_active_insync_subscriptions() -> list[SubscriptionTable]:
def get_active_site_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
"""Retrieve active subscriptions specifically for sites.
- :param includes: The fields to be included in the returned Subscription objects.
- :type includes: list[str]
+ Args:
+ includes: The fields to be included in the returned Subscription objects.
- :return: A list of Subscription objects for sites.
- :rtype: list[Subscription]
+ Returns:
+ A list of Subscription objects for sites.
"""
return get_subscriptions(
product_types=[ProductType.SITE], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
@@ -249,11 +256,11 @@ def get_active_site_subscriptions(includes: list[str] | None = None) -> list[Sub
def get_active_edge_port_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
"""Retrieve active Edge Port subscriptions.
- :param includes: The fields to be included in the returned Subscription objects.
- :type includes: list[str]
+ Args:
+ includes: The fields to be included in the returned Subscription objects.
- :return: A list of Subscription objects for Edge Ports.
- :rtype: list[Subscription]
+ Returns:
+ A list of Subscription objects for Edge Ports.
"""
return get_subscriptions(
product_types=[ProductType.EDGE_PORT], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
@@ -263,8 +270,8 @@ def get_active_edge_port_subscriptions(includes: list[str] | None = None) -> lis
def get_site_by_name(site_name: str) -> Site:
"""Get a site by its name.
- :param site_name: The name of the site.
- :type site_name: str
+ Args:
+ site_name: The name of the site.
"""
subscription = get_active_subscriptions_by_field_and_value("site_name", site_name)
if not subscription:
@@ -277,8 +284,8 @@ def get_site_by_name(site_name: str) -> Site:
def get_all_active_sites() -> list[dict[str, Any]]:
"""Retrieve all active sites subscription together with instance values.
- :return: A list of active sites with their subscription IDs and site instances.
- :rtype: list[dict[str, Any]]
+ Returns:
+ A list of active sites with their subscription IDs and site instances.
"""
return [
{
@@ -315,9 +322,10 @@ def is_virtual_circuit_id_available(virtual_circuit_id: str) -> bool:
This function verifies if the specified virtual circuit ID is not already
present in the core database.
- :param virtual_circuit_id: The virtual circuit ID to check.
- :type virtual_circuit_id: str
- :return: True if the virtual circuit ID is unique (not found), False if it exists.
- :rtype: bool
+ Args:
+ virtual_circuit_id: The virtual circuit ID to check.
+
+ Returns:
+ True if the virtual circuit ID is unique (not found), False if it exists.
"""
return is_resource_type_value_unique("virtual_circuit_id", virtual_circuit_id)
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index 3557b4eb..852e152f 100644
--- a/gso/utils/helpers.py
+++ b/gso/utils/helpers.py
@@ -96,11 +96,11 @@ def available_service_lags_choices(router_id: UUID) -> Choice | None:
def get_router_vendor(router_id: UUID) -> Vendor:
"""Retrieve the vendor of a router.
- :param router_id: The UUID of the router.
- :type router_id: :class:`uuid.UUID`
+ Args:
+ router_id: The UUID of the router.
- :return: The vendor of the router.
- :rtype: Vendor:
+ Returns:
+ The vendor of the router.
"""
return Router.from_subscription(router_id).router.vendor
@@ -108,8 +108,11 @@ def get_router_vendor(router_id: UUID) -> Vendor:
def iso_from_ipv4(ipv4_address: IPv4AddressType) -> str:
"""Calculate an ISO address, based on an IPv4 address.
- :param IPv4Address ipv4_address: The address that's to be converted
- :returns: An ISO-formatted address.
+ Args:
+ ipv4_address: The address that's to be converted
+
+ Returns:
+ An ISO-formatted address.
"""
padded_octets = [f"{x:>03}" for x in str(ipv4_address).split(".")]
joined_octets = "".join(padded_octets)
@@ -134,12 +137,14 @@ def generate_inventory_for_routers(
Contains all active routers of a specific role. Optionally, routers can be excluded from the inventory.
- :param RouterRole router_role: The role of the routers to include in the inventory.
- :param list exclude_routers: List of routers to exclude from the inventory.
- :param Vendor router_vendor: The vendor of the routers to include in the inventory.
- :param bool include_provisioning_routers: Include routers that are in a ``PROVISIONING`` state.
- :return: A dictionary representing the inventory of active routers.
- :rtype: dict[str, Any]
+ Args:
+ router_role: The role of the routers to include in the inventory.
+ exclude_routers: List of routers to exclude from the inventory.
+ router_vendor: The vendor of the routers to include in the inventory.
+ include_provisioning_routers: Include routers that are in a ``PROVISIONING`` state.
+
+ Returns:
+ A dictionary representing the inventory of active routers.
"""
lifecycles = (
[SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE]
@@ -175,9 +180,12 @@ def calculate_recommended_minimum_links(iptrunk_number_of_members: int, iptrunk_
If the IP trunk speed is 400G, the recommended minimum number of links is the number of members minus 1.
Otherwise, the recommended minimum number of links is the number of members.
- :param int iptrunk_number_of_members: The number of members in the IP trunk.
- :param PhysicalPortCapacity iptrunk_speed: The speed of the IP trunk.
- :return: The recommended minimum number of links for the IP trunk.
+ Args:
+ iptrunk_number_of_members: The number of members in the IP trunk.
+ iptrunk_speed: The speed of the IP trunk.
+
+ Returns:
+ The recommended minimum number of links for the IP trunk.
"""
if iptrunk_speed == PhysicalPortCapacity.FOUR_HUNDRED_GIGABIT_PER_SECOND:
return iptrunk_number_of_members - 1
@@ -254,9 +262,12 @@ def partner_choice() -> Choice:
def validate_edge_port_number_of_members_based_on_lacp(*, number_of_members: int, enable_lacp: bool) -> None:
"""Validate the number of edge port members based on the LACP setting.
- :param number_of_members: The number of members to validate.
- :param enable_lacp: Whether LACP is enabled or not.
- :raises ValueError: If the number of members is greater than 1 and LACP is disabled.
+ Args:
+ number_of_members: The number of members to validate.
+ enable_lacp: Whether LACP is enabled or not.
+
+ Raises:
+ ValueError: If the number of members is greater than 1 and LACP is disabled.
"""
if number_of_members > 1 and not enable_lacp:
err_msg = "Number of members must be 1 if LACP is disabled."
@@ -270,10 +281,11 @@ def generate_unique_vc_id(max_attempts: int = 100) -> VC_ID | None:
checking its uniqueness before returning it. A maximum attempt limit is
set to prevent infinite loops in case the ID space is saturated.
- :param max_attempts: The maximum number of attempts to generate a unique ID.
- :type max_attempts: int
- :return: A unique VC_ID instance if successful, None if no unique ID is found.
- :rtype: Optional[VC_ID]
+ Args:
+ max_attempts: The maximum number of attempts to generate a unique ID.
+
+ Returns:
+ A unique VC_ID instance if successful, None if no unique ID is found.
"""
def create_vc_id() -> str:
diff --git a/gso/utils/shared_enums.py b/gso/utils/shared_enums.py
index a6050c9b..4272d588 100644
--- a/gso/utils/shared_enums.py
+++ b/gso/utils/shared_enums.py
@@ -20,7 +20,7 @@ class ConnectionStrategy(strEnum):
class SNMPVersion(StrEnum):
- """An enumerator for the two relevant versions of :term:`SNMP`: v2c and 3."""
+ """An enumerator for the two relevant versions of SNMP: v2c and 3."""
V2C = "v2c"
V3 = "v3"
diff --git a/gso/utils/types/interfaces.py b/gso/utils/types/interfaces.py
index 2251b49e..80ac4a5f 100644
--- a/gso/utils/types/interfaces.py
+++ b/gso/utils/types/interfaces.py
@@ -12,7 +12,7 @@ from typing_extensions import Doc
class LAGMember(BaseModel):
- """A :term:`LAG` member interface that consists of a name and description."""
+ """A LAG member interface that consists of a name and description."""
interface_name: str
interface_description: str | None = None
@@ -25,13 +25,13 @@ class LAGMember(BaseModel):
def validate_interface_names_are_unique(interfaces: list[LAGMember]) -> list[LAGMember]:
"""Verify if interfaces are unique.
- Raises a :class:`ValueError` if the interfaces are not unique.
+ Raises a ``ValueError`` if the interfaces are not unique.
- :param interfaces: The list of interfaces.
- :type interfaces: list[:class:`utils.types.LAGMember`]
+ Args:
+ interfaces: The list of interfaces.
- :return: The list of interfaces
- :rtype: list[:class:`utils.types.LAGMember`]
+ Returns:
+ The list of interfaces
"""
interface_names = [member.interface_name for member in interfaces]
if len(interface_names) != len(set(interface_names)):
@@ -48,9 +48,11 @@ def validate_juniper_phy_interface_name(interface_name: str) -> str:
another forward slash '/', and ends with a number between 0 and 99.
For example: 'xe-1/0/0'. This only applies to Juniper-brand hardware.
- :param str interface_name: Interface name to validate.
+ Args:
+ interface_name: Interface name to validate.
- :return str: The interface name if match was successful, otherwise it will throw a ValueError exception.
+ Returns:
+ The interface name if match was successful, otherwise it will throw a ValueError exception.
"""
pattern = re.compile(r"^(ge|et|xe)-1?[0-9]/[0-9]{1,2}/[0-9]{1,2}$")
if not bool(pattern.match(interface_name)):
@@ -62,7 +64,7 @@ def validate_juniper_phy_interface_name(interface_name: str) -> str:
def validate_juniper_ae_interface_name(interface_name: str) -> str:
- """Validate that the provided interface name matches the expected pattern for a :term:`LAG` interface.
+ """Validate that the provided interface name matches the expected pattern for a LAG interface.
Interface names must match the pattern 'ae' followed by a one- or two-digit number.
"""
@@ -80,12 +82,12 @@ LAGMemberList = Annotated[
AfterValidator(validate_unique_list),
AfterValidator(validate_interface_names_are_unique),
Len(min_length=0),
- Doc("A list of :term:`LAG` member interfaces."),
+ Doc("A list of LAG member interfaces."),
]
class JuniperLAGMember(LAGMember):
- """A Juniper-specific :term:`LAG` member interface."""
+ """A Juniper-specific LAG member interface."""
interface_name: JuniperPhyInterface
diff --git a/gso/utils/types/netbox_router.py b/gso/utils/types/netbox_router.py
index 8ea14aa4..c184c09b 100644
--- a/gso/utils/types/netbox_router.py
+++ b/gso/utils/types/netbox_router.py
@@ -13,13 +13,14 @@ from gso.utils.shared_enums import Vendor
def validate_router_in_netbox(subscription_id: UUIDstr) -> UUIDstr:
"""Verify if a device exists in Netbox.
- Raises a :class:`ValueError` if the device is not found.
+ Args:
+ subscription_id: The UUID of the router subscription.
- :param subscription_id: The :term:`UUID` of the router subscription.
- :type subscription_id: :class:`UUIDstr`
+ Returns:
+ The UUID of the router subscription.
- :return: The :term:`UUID` of the router subscription.
- :rtype: :class:`UUIDstr`
+ Raises:
+ ValueError: if the device is not found.
"""
router_type = Router.from_subscription(subscription_id)
if router_type.router.vendor == Vendor.NOKIA:
diff --git a/gso/utils/types/tt_number.py b/gso/utils/types/tt_number.py
index a9e1d05c..56c26d56 100644
--- a/gso/utils/types/tt_number.py
+++ b/gso/utils/types/tt_number.py
@@ -11,9 +11,11 @@ def validate_tt_number(tt_number: str) -> str:
This method checks if the input string starts with 'TT#' and is followed by exactly 16 digits.
- :param str tt_number: The TT number as string to validate
+ Args:
+ tt_number: The TT number as string to validate
- :return str: The TT number string if TT number match was successful, otherwise it will raise a ValueError.
+ Returns:
+ The TT number string if TT number match was successful, otherwise it will raise a ValueError.
"""
pattern = r"^TT#\d{16}$"
if not bool(re.match(pattern, tt_number)):
diff --git a/gso/utils/types/virtual_identifiers.py b/gso/utils/types/virtual_identifiers.py
index 2208ca74..093678e3 100644
--- a/gso/utils/types/virtual_identifiers.py
+++ b/gso/utils/types/virtual_identifiers.py
@@ -1,4 +1,4 @@
-"""Annotated types for virtual identifiers such as :term:`VLAN` ID or Virtual Circuit ID."""
+"""Annotated types for virtual identifiers such as VLAN ID or Virtual Circuit ID."""
from typing import Annotated
diff --git a/gso/workflows/edge_port/create_edge_port.py b/gso/workflows/edge_port/create_edge_port.py
index 18629e98..2005d7c2 100644
--- a/gso/workflows/edge_port/create_edge_port.py
+++ b/gso/workflows/edge_port/create_edge_port.py
@@ -168,10 +168,10 @@ def initialize_subscription(
@step("Reserve interfaces in NetBox")
def reserve_interfaces_in_netbox(subscription: EdgePortProvisioning) -> State:
- """Create the :term:`LAG` interfaces in NetBox and attach the LAG interfaces to the physical interfaces."""
+ """Create the LAG interfaces in NetBox and attach the LAG interfaces to the physical interfaces."""
nbclient = NetboxClient()
edge_port = subscription.edge_port
- # Create :term:`LAG` interfaces
+ # Create LAG interfaces
lag_interface: Interfaces = nbclient.create_interface(
iface_name=edge_port.edge_port_name,
interface_type="lag",
@@ -179,7 +179,7 @@ def reserve_interfaces_in_netbox(subscription: EdgePortProvisioning) -> State:
description=str(subscription.subscription_id),
enabled=True,
)
- # Attach physical interfaces to :term:`LAG`
+ # Attach physical interfaces to LAG
# Update interface description to subscription ID
# Reserve interfaces
for interface in edge_port.edge_port_ae_members:
@@ -261,7 +261,7 @@ def create_edge_port() -> StepList:
* Create and initialise the subscription object in the service database
* Deploy configuration on the new edge port, first as a dry run
- * allocate :term:`LAG` and :term:`LAG` members in the Netbox.
+ * allocate LAG and LAG members in the Netbox.
"""
return (
begin
diff --git a/gso/workflows/edge_port/create_imported_edge_port.py b/gso/workflows/edge_port/create_imported_edge_port.py
index 5aa24a28..867965c7 100644
--- a/gso/workflows/edge_port/create_imported_edge_port.py
+++ b/gso/workflows/edge_port/create_imported_edge_port.py
@@ -34,7 +34,7 @@ def create_subscription(partner: str) -> State:
def initial_input_form_generator() -> FormGenerator:
- """Generate a form that is filled in using information passed through the :term:`API` endpoint."""
+ """Generate a form that is filled in using information passed through the API endpoint."""
class ImportEdgePort(FormPage):
model_config = ConfigDict(title="Import Edge Port")
diff --git a/gso/workflows/edge_port/modify_edge_port.py b/gso/workflows/edge_port/modify_edge_port.py
index 67ff810b..bce56c74 100644
--- a/gso/workflows/edge_port/modify_edge_port.py
+++ b/gso/workflows/edge_port/modify_edge_port.py
@@ -167,7 +167,7 @@ def update_interfaces_in_netbox(
# Free removed interfaces
for removed_member in removed_ae_members:
nbclient.free_interface(subscription.edge_port.node.router_fqdn, removed_member["interface_name"])
- # Attach physical interfaces to :term:`LAG`
+ # Attach physical interfaces to LAG
# Update interface description to subscription ID
# Reserve interfaces
for member in subscription.edge_port.edge_port_ae_members:
@@ -234,7 +234,7 @@ def update_edge_port_real(
@step("Allocate/Deallocate interfaces in NetBox")
def allocate_interfaces_in_netbox(subscription: EdgePort, previous_ae_members: list[dict]) -> None:
- """Allocate the new interfaces in NetBox and detach the old ones from the :term:`LAG`."""
+ """Allocate the new interfaces in NetBox and detach the old ones from the LAG."""
nbclient = NetboxClient()
for member in subscription.edge_port.edge_port_ae_members:
if any(member.interface_name == prev_member["interface_name"] for prev_member in previous_ae_members):
@@ -260,7 +260,7 @@ def modify_edge_port() -> StepList:
* Modify the subscription object in the service database
* Modify configuration on the new edge port, first as a dry run
- * Change :term:`LAG` and :term:`LAG` members in the Netbox.
+ * Change LAG and LAG members in the Netbox.
"""
capacity_has_changed = conditional(lambda state: state["capacity_has_changed"])
return (
diff --git a/gso/workflows/edge_port/terminate_edge_port.py b/gso/workflows/edge_port/terminate_edge_port.py
index 72d014c3..ffaa023e 100644
--- a/gso/workflows/edge_port/terminate_edge_port.py
+++ b/gso/workflows/edge_port/terminate_edge_port.py
@@ -18,7 +18,7 @@ from gso.utils.types.tt_number import TTNumber
def initial_input_form_generator() -> FormGenerator:
- """Let the operator decide whether to delete configuration on the router, and clear up :term:`IPAM` resources."""
+ """Let the operator decide whether to delete configuration on the router, and clear up IPAM resources."""
class TerminateForm(FormPage):
tt_number: TTNumber
@@ -65,7 +65,7 @@ def remove_edge_port_real(subscription: dict[str, Any], tt_number: str, process_
@step("Netbox Clean Up")
def netbox_clean_up(subscription: EdgePort) -> None:
- """Update Netbox to remove the edge port :term:`LAG` interface and all the :term:`LAG` members."""
+ """Update Netbox to remove the edge port LAG interface and all the LAG members."""
nbclient = NetboxClient()
for member in subscription.edge_port.edge_port_ae_members:
diff --git a/gso/workflows/iptrunk/create_imported_iptrunk.py b/gso/workflows/iptrunk/create_imported_iptrunk.py
index 4d754375..5cf9c5f8 100644
--- a/gso/workflows/iptrunk/create_imported_iptrunk.py
+++ b/gso/workflows/iptrunk/create_imported_iptrunk.py
@@ -24,7 +24,7 @@ from gso.utils.types.interfaces import LAGMember, LAGMemberList, PhysicalPortCap
def initial_input_form_generator() -> FormGenerator:
- """Take all information passed to this workflow by the :term:`API` endpoint that was called."""
+ """Take all information passed to this workflow by the API endpoint that was called."""
class CreateIptrunkForm(FormPage):
model_config = ConfigDict(title="Import Iptrunk")
@@ -123,7 +123,7 @@ def update_ipam_stub_for_subscription(
iptrunk_ipv4_network: ipaddress.IPv4Network,
iptrunk_ipv6_network: ipaddress.IPv6Network,
) -> State:
- """Update :term:`IPAM` information in the subscription."""
+ """Update IPAM information in the subscription."""
subscription.iptrunk.iptrunk_ipv4_network = iptrunk_ipv4_network
subscription.iptrunk.iptrunk_ipv6_network = iptrunk_ipv6_network
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index ef111977..f8b54c61 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/gso/workflows/iptrunk/create_iptrunk.py
@@ -434,7 +434,7 @@ def check_ip_trunk_connectivity(subscription: IptrunkInactive) -> LSOState:
@step("[DRY RUN] Provision IP trunk ISIS interface")
def provision_ip_trunk_isis_iface_dry(subscription: IptrunkInactive, process_id: UUIDstr, tt_number: str) -> LSOState:
- """Perform a dry run of deploying :term:`ISIS` configuration."""
+ """Perform a dry run of deploying ISIS configuration."""
extra_vars = {
"wfo_trunk_json": json.loads(json_dumps(subscription)),
"dry_run": True,
@@ -460,7 +460,7 @@ def provision_ip_trunk_isis_iface_dry(subscription: IptrunkInactive, process_id:
@step("[FOR REAL] Provision IP trunk ISIS interface")
def provision_ip_trunk_isis_iface_real(subscription: IptrunkInactive, process_id: UUIDstr, tt_number: str) -> LSOState:
- """Deploy :term:`ISIS` configuration on both sides."""
+ """Deploy ISIS configuration on both sides."""
extra_vars = {
"wfo_trunk_json": json.loads(json_dumps(subscription)),
"dry_run": False,
@@ -486,7 +486,7 @@ def provision_ip_trunk_isis_iface_real(subscription: IptrunkInactive, process_id
@step("Check ISIS adjacency")
def check_ip_trunk_isis(subscription: IptrunkInactive) -> LSOState:
- """Run an Ansible playbook to confirm :term:`ISIS` adjacency."""
+ """Run an Ansible playbook to confirm ISIS adjacency."""
extra_vars = {"wfo_ip_trunk_json": json.loads(json_dumps(subscription)), "check": "isis"}
return {
@@ -498,7 +498,7 @@ def check_ip_trunk_isis(subscription: IptrunkInactive) -> LSOState:
@step("Register DNS records for both sides of the trunk")
def register_dns_records(subscription: IptrunkInactive) -> State:
- """Register :term:`DNS` records for both sides of the newly created IPtrunk."""
+ """Register DNS records for both sides of the newly created IPtrunk."""
for index, side in enumerate(subscription.iptrunk.iptrunk_sides):
fqdn = f"{side.iptrunk_side_ae_iface}-0.{side.iptrunk_side_node.router_fqdn}"
if not (subscription.iptrunk.iptrunk_ipv4_network and subscription.iptrunk.iptrunk_ipv6_network):
@@ -514,11 +514,11 @@ def register_dns_records(subscription: IptrunkInactive) -> State:
@step("NextBox integration")
def reserve_interfaces_in_netbox(subscription: IptrunkInactive) -> State:
- """Create the :term:`LAG` interfaces in NetBox and attach the LAG interfaces to the physical interfaces."""
+ """Create the LAG interfaces in NetBox and attach the LAG interfaces to the physical interfaces."""
nbclient = NetboxClient()
for trunk_side in subscription.iptrunk.iptrunk_sides:
if get_router_vendor(trunk_side.iptrunk_side_node.owner_subscription_id) == Vendor.NOKIA:
- # Create :term:`LAG` interfaces
+ # Create LAG interfaces
lag_interface: Interfaces = nbclient.create_interface(
iface_name=trunk_side.iptrunk_side_ae_iface, # type: ignore[arg-type]
interface_type="lag",
@@ -526,7 +526,7 @@ def reserve_interfaces_in_netbox(subscription: IptrunkInactive) -> State:
description=str(subscription.subscription_id),
enabled=True,
)
- # Attach physical interfaces to :term:`LAG`
+ # Attach physical interfaces to LAG
# Update interface description to subscription ID
# Reserve interfaces
for interface in trunk_side.iptrunk_side_ae_members:
@@ -558,13 +558,13 @@ def _allocate_interfaces_in_netbox(iptrunk_side: IptrunkSideBlockInactive) -> No
@step("Allocate interfaces in Netbox for side A")
def netbox_allocate_side_a_interfaces(subscription: IptrunkInactive) -> None:
- """Allocate the :term:`LAG` interfaces for the Nokia router on side A."""
+ """Allocate the LAG interfaces for the Nokia router on side A."""
_allocate_interfaces_in_netbox(subscription.iptrunk.iptrunk_sides[0])
@step("Allocate interfaces in Netbox for side B")
def netbox_allocate_side_b_interfaces(subscription: IptrunkInactive) -> None:
- """Allocate the :term:`LAG` interfaces for the Nokia router on side B."""
+ """Allocate the LAG interfaces for the Nokia router on side B."""
_allocate_interfaces_in_netbox(subscription.iptrunk.iptrunk_sides[1])
@@ -597,8 +597,8 @@ def create_iptrunk() -> StepList:
* Reserve interfaces in Netbox
* Deploy configuration on the two sides of the trunk, first as a dry run
* Check connectivity on the new trunk
- * Deploy the new :term:`ISIS` metric on the trunk, first as a dry run
- * Verify :term:`ISIS` adjacency
+ * Deploy the new ISIS metric on the trunk, first as a dry run
+ * Verify ISIS adjacency
* Allocate the interfaces in Netbox
* Set the subscription to active in the database
"""
diff --git a/gso/workflows/iptrunk/deploy_twamp.py b/gso/workflows/iptrunk/deploy_twamp.py
index 66b1f1c1..f5712b27 100644
--- a/gso/workflows/iptrunk/deploy_twamp.py
+++ b/gso/workflows/iptrunk/deploy_twamp.py
@@ -114,9 +114,9 @@ def check_twamp_status(subscription: Iptrunk) -> LSOState:
target=Target.MODIFY,
)
def deploy_twamp() -> StepList:
- """Deploy a :term:`TWAMP` session on an IP trunk.
+ """Deploy a TWAMP session on an IP trunk.
- * Run the :term:`TWAMP` playbook, including an initial dry run
+ * Run the TWAMP playbook, including an initial dry run
"""
return (
begin
diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index b101c9c2..5684dd2d 100644
--- a/gso/workflows/iptrunk/migrate_iptrunk.py
+++ b/gso/workflows/iptrunk/migrate_iptrunk.py
@@ -175,7 +175,7 @@ def netbox_reserve_interfaces(
description=str(subscription.subscription_id),
enabled=True,
)
- # Attach physical interfaces to :term:`LAG`
+ # Attach physical interfaces to LAG
# Reserve interfaces
for interface in new_lag_member_interfaces:
nbclient.attach_interface_to_lag(
@@ -254,7 +254,7 @@ def check_ip_trunk_optical_levels_post(
def check_ip_trunk_lldp(
subscription: Iptrunk, new_node: Router, new_lag_member_interfaces: list[dict], replace_index: int
) -> LSOState:
- """Check :term:`LLDP` on the new trunk endpoints."""
+ """Check LLDP on the new trunk endpoints."""
extra_vars = {
"wfo_ip_trunk_json": json.loads(json_dumps(subscription)),
"new_node": json.loads(json_dumps(new_node)),
@@ -491,7 +491,7 @@ def update_remaining_side_bfd_real(
@step("Check BFD session over trunk")
def check_ip_trunk_bfd(subscription: Iptrunk, new_node: Router, replace_index: int) -> LSOState:
- """Check :term:`BFD` session across the new trunk."""
+ """Check BFD session across the new trunk."""
extra_vars = {
"wfo_ip_trunk_json": json.loads(json_dumps(subscription)),
"new_node": json.loads(json_dumps(new_node)),
@@ -553,7 +553,7 @@ def deploy_new_isis(
process_id: UUIDstr,
tt_number: str,
) -> LSOState:
- """Deploy :term:`ISIS` configuration."""
+ """Deploy ISIS configuration."""
extra_vars = {
"wfo_trunk_json": json.loads(json_dumps(subscription)),
"new_node": json.loads(json_dumps(new_node)),
@@ -584,7 +584,7 @@ def deploy_new_isis(
@step("Check ISIS adjacency")
def check_ip_trunk_isis(subscription: Iptrunk, replace_index: int) -> LSOState:
- """Run an Ansible playbook to confirm :term:`ISIS` adjacency."""
+ """Run an Ansible playbook to confirm ISIS adjacency."""
extra_vars = {"wfo_ip_trunk_json": json.loads(json_dumps(subscription)), "check": "isis"}
return {
@@ -602,7 +602,7 @@ def check_ip_trunk_isis(subscription: Iptrunk, replace_index: int) -> LSOState:
@inputstep("Wait for confirmation", assignee=Assignee.SYSTEM)
def confirm_continue_restore_isis() -> FormGenerator:
- """Wait for an operator to confirm that the old :term:`ISIS` metric should be restored."""
+ """Wait for an operator to confirm that the old ISIS metric should be restored."""
class ProvisioningResultPage(FormPage):
model_config = ConfigDict(title="Please confirm before continuing")
@@ -621,7 +621,7 @@ def restore_isis_metric(
tt_number: str,
old_isis_metric: int,
) -> LSOState:
- """Restore the :term:`ISIS` metric to its original value."""
+ """Restore the ISIS metric to its original value."""
subscription.iptrunk.iptrunk_isis_metric = old_isis_metric
extra_vars = {
"wfo_trunk_json": json.loads(json_dumps(subscription)),
@@ -727,9 +727,9 @@ def delete_old_config_real(
@step("Update IP records in IPAM")
def update_ipam(subscription: Iptrunk, replace_index: int, new_node: Router, new_lag_interface: str) -> State:
- """Update :term:`IPAM` resources.
+ """Update IPAM resources.
- Move the :term:`DNS` record pointing to the old side of the trunk, to the new side.
+ Move the DNS record pointing to the old side of the trunk, to the new side.
"""
v4_addr = subscription.iptrunk.iptrunk_ipv4_network[replace_index]
# IPv6 networks start with an unused address we need to skip past.
@@ -778,7 +778,7 @@ def update_subscription_model(
@step("Netbox: Remove old LAG interface")
def netbox_remove_old_interfaces(old_side_data: dict) -> State:
- """Remove the old :term:`LAG` interface from Netbox, only relevant if the old side is a Nokia router."""
+ """Remove the old LAG interface from Netbox, only relevant if the old side is a Nokia router."""
nbclient = NetboxClient()
for iface in old_side_data["iptrunk_side_ae_members"]:
@@ -797,7 +797,7 @@ def netbox_remove_old_interfaces(old_side_data: dict) -> State:
@step("Netbox: Allocate new LAG member interfaces")
def netbox_allocate_new_interfaces(subscription: Iptrunk, replace_index: int) -> State:
- """Allocate the new :term:`LAG` interface in Netbox. Only relevant if the new router is a Nokia."""
+ """Allocate the new LAG interface in Netbox. Only relevant if the new router is a Nokia."""
nbclient = NetboxClient()
new_side = subscription.iptrunk.iptrunk_sides[replace_index]
@@ -835,15 +835,15 @@ def migrate_iptrunk() -> StepList:
"""Migrate an IP trunk.
* Reserve new interfaces in Netbox
- * Set the :term:`ISIS` metric of the current trunk to an arbitrarily high value to drain all traffic
+ * Set the ISIS metric of the current trunk to an arbitrarily high value to drain all traffic
* Disable - but do not delete - the old configuration on the routers, first as a dry run
* Deploy the new configuration on the routers, first as a dry run
* Wait for operator confirmation that the physical fiber has been moved before continuing
- * Deploy a new :term:`ISIS` interface between routers A and C
- * Wait for operator confirmation that :term:`ISIS` is behaving as expected
- * Restore the old :term:`ISIS` metric on the new trunk
+ * Deploy a new ISIS interface between routers A and C
+ * Wait for operator confirmation that ISIS is behaving as expected
+ * Restore the old ISIS metric on the new trunk
* Delete the old configuration from the routers, first as a dry run
- * Reflect the changes made in :term:`IPAM`
+ * Reflect the changes made in IPAM
* Update the subscription model in the database
* Update the reserved interfaces in Netbox
"""
diff --git a/gso/workflows/iptrunk/modify_trunk_interface.py b/gso/workflows/iptrunk/modify_trunk_interface.py
index e76d9d2e..9c1be5cd 100644
--- a/gso/workflows/iptrunk/modify_trunk_interface.py
+++ b/gso/workflows/iptrunk/modify_trunk_interface.py
@@ -381,7 +381,7 @@ def _netbox_update_interfaces(
for member in removed_ae_members:
nbclient.free_interface(side_block.iptrunk_side_node.router_fqdn, member["interface_name"])
- # Attach physical interfaces to :term:`LAG`
+ # Attach physical interfaces to LAG
# Update interface description to subscription ID
# Reserve interfaces
for interface in side_block.iptrunk_side_ae_members:
@@ -448,18 +448,18 @@ def _netbox_allocate_interfaces(side_block: IptrunkSideBlock, previous_ae_member
@step("Netbox: Allocate side A interfaces")
def allocate_interfaces_in_netbox_side_a(subscription: Iptrunk, previous_ae_members: list[list[dict]]) -> None:
- """Allocate the :term:`LAG` interfaces on side A in Netbox.
+ """Allocate the LAG interfaces on side A in Netbox.
- Attach the :term:`LAG` interface to the physical interface detach old one from the :term:`LAG`.
+ Attach the LAG interface to the physical interface detach old one from the LAG.
"""
_netbox_allocate_interfaces(subscription.iptrunk.iptrunk_sides[0], previous_ae_members[0])
@step("Netbox: Allocate side B interfaces")
def allocate_interfaces_in_netbox_side_b(subscription: Iptrunk, previous_ae_members: list[list[dict]]) -> None:
- """Allocate the :term:`LAG` interface on side B in Netbox.
+ """Allocate the LAG interface on side B in Netbox.
- Attach the :term:`LAG` interface to the physical interface detach old one from the :term:`LAG`.
+ Attach the LAG interface to the physical interface detach old one from the LAG.
"""
_netbox_allocate_interfaces(subscription.iptrunk.iptrunk_sides[1], previous_ae_members[1])
diff --git a/gso/workflows/iptrunk/validate_iptrunk.py b/gso/workflows/iptrunk/validate_iptrunk.py
index f66612f5..b8e87928 100644
--- a/gso/workflows/iptrunk/validate_iptrunk.py
+++ b/gso/workflows/iptrunk/validate_iptrunk.py
@@ -38,9 +38,10 @@ def validate_router_config(subscription: Iptrunk) -> LSOState:
@step("Verify IPAM resources for LAG interfaces")
def verify_ipam_records(subscription: Iptrunk) -> None:
- """Validate the :term:`IPAM` resources for the :term:`LAG` interfaces.
+ """Validate the IPAM resources for the LAG interfaces.
- Raises an :class:`orchestrator.utils.errors.ProcessFailureError` if :term:`IPAM` is configured incorrectly.
+ Raises:
+ ProcessFailureError: if IPAM is configured incorrectly.
"""
ipam_errors = []
ipam_v4_network = infoblox.find_network_by_cidr(subscription.iptrunk.iptrunk_ipv4_network)
@@ -158,7 +159,7 @@ def verify_iptrunk_config(subscription: Iptrunk) -> LSOState:
@step("Check ISIS configuration")
def check_ip_trunk_isis(subscription: Iptrunk) -> LSOState:
- """Run an Ansible playbook to check for any :term:`ISIS` configuration drift."""
+ """Run an Ansible playbook to check for any ISIS configuration drift."""
return {
"playbook_name": "gap_ansible/playbooks/iptrunks.yaml",
"inventory": {
@@ -207,10 +208,10 @@ def verify_twamp_config(subscription: Iptrunk) -> LSOState:
def validate_iptrunk() -> StepList:
"""Validate an existing, active IP Trunk subscription.
- * Verify that the :term:`LAG` interfaces are correctly configured in :term:`IPAM`.
+ * Verify that the LAG interfaces are correctly configured in IPAM.
* Check correct configuration of interfaces in NetBox.
* Verify the configuration on both sides of the trunk is intact.
- * Check the :term:`ISIS` metric of the trunk.
+ * Check the ISIS metric of the trunk.
* Verify that TWAMP configuration is correct.
If a trunk has a Juniper router on both sides, it is considered legacy and does not require validation.
diff --git a/gso/workflows/l2_circuit/create_imported_layer_2_circuit.py b/gso/workflows/l2_circuit/create_imported_layer_2_circuit.py
index 1a94e4ef..a0fc39cb 100644
--- a/gso/workflows/l2_circuit/create_imported_layer_2_circuit.py
+++ b/gso/workflows/l2_circuit/create_imported_layer_2_circuit.py
@@ -28,7 +28,7 @@ from gso.utils.types.virtual_identifiers import VC_ID, VLAN_ID
def initial_input_form_generator() -> FormGenerator:
- """Generate a form that can be pre-filled using an :term:`API` endpoint."""
+ """Generate a form that can be pre-filled using an API endpoint."""
class ServiceBindingPortInput(BaseModel):
edge_port: UUIDstr
@@ -52,7 +52,7 @@ def initial_input_form_generator() -> FormGenerator:
@model_validator(mode="after")
def tagged_layer_2_circuit_has_vlan_bounds(self) -> Self:
- """If a Layer 2 Circuit is tagged, it must have a :term:`VLAN` range set."""
+ """If a Layer 2 Circuit is tagged, it must have a VLAN range set."""
if self.layer_2_circuit_type == Layer2CircuitType.TAGGED and (
self.vlan_range_lower_bound is None or self.vlan_range_upper_bound is None
):
diff --git a/gso/workflows/l3_core_service/create_imported_l3_core_service.py b/gso/workflows/l3_core_service/create_imported_l3_core_service.py
index 0ff4c30f..dd77021b 100644
--- a/gso/workflows/l3_core_service/create_imported_l3_core_service.py
+++ b/gso/workflows/l3_core_service/create_imported_l3_core_service.py
@@ -26,7 +26,7 @@ from gso.utils.types.virtual_identifiers import VLAN_ID
def initial_input_form_generator() -> FormGenerator:
- """Take all information passed to this workflow by the :term:`API` endpoint that was called."""
+ """Take all information passed to this workflow by the API endpoint that was called."""
class BFDSettingsModel(BaseModel):
bfd_enabled: bool = False
diff --git a/gso/workflows/l3_core_service/create_l3_core_service.py b/gso/workflows/l3_core_service/create_l3_core_service.py
index f23a2fcf..70686fb6 100644
--- a/gso/workflows/l3_core_service/create_l3_core_service.py
+++ b/gso/workflows/l3_core_service/create_l3_core_service.py
@@ -247,7 +247,7 @@ def check_sbp_functionality(subscription: dict[str, Any], edge_port_fqdn_list: l
def deploy_bgp_peers_dry(
subscription: dict[str, Any], edge_port_fqdn_list: list[str], tt_number: str, process_id: UUIDstr, partner_name: str
) -> LSOState:
- """Perform a dry run of deploying :term:`BGP` peers."""
+ """Perform a dry run of deploying BGP peers."""
extra_vars = {
"subscription": subscription,
"partner_name": partner_name,
@@ -269,7 +269,7 @@ def deploy_bgp_peers_dry(
def deploy_bgp_peers_real(
subscription: dict[str, Any], edge_port_fqdn_list: list[str], tt_number: str, process_id: UUIDstr, partner_name: str
) -> LSOState:
- """Deploy :term:`BGP` peers."""
+ """Deploy BGP peers."""
extra_vars = {
"subscription": subscription,
"partner_name": partner_name,
@@ -289,7 +289,7 @@ def deploy_bgp_peers_real(
@step("Check BGP peers")
def check_bgp_peers(subscription: dict[str, Any], edge_port_fqdn_list: list[str]) -> LSOState:
- """Check correct deployment of :term:`BGP` peers."""
+ """Check correct deployment of BGP peers."""
extra_vars = {"subscription": subscription, "verb": "check", "object": "bgp"}
return {
@@ -301,7 +301,7 @@ def check_bgp_peers(subscription: dict[str, Any], edge_port_fqdn_list: list[str]
@step("Update Infoblox")
def update_dns_records(subscription: L3CoreService) -> State:
- """Update :term:`DNS` records in Infoblox."""
+ """Update DNS records in Infoblox."""
# TODO: implement
return {"subscription": subscription}
@@ -316,8 +316,8 @@ def create_l3_core_service() -> StepList:
* Create subscription object in the service database
* Deploy service binding ports
- * Deploy :term:`BGP` peers
- * Update :term:`DNS` records
+ * Deploy BGP peers
+ * Update DNS records
* Set the subscription in a provisioning state in the database
"""
return (
diff --git a/gso/workflows/l3_core_service/modify_l3_core_service.py b/gso/workflows/l3_core_service/modify_l3_core_service.py
index e5f1f127..630418f7 100644
--- a/gso/workflows/l3_core_service/modify_l3_core_service.py
+++ b/gso/workflows/l3_core_service/modify_l3_core_service.py
@@ -246,7 +246,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@step("Clean up removed Edge Ports")
def remove_old_sbp_blocks(subscription: L3CoreService, removed_access_ports: list[UUIDstr]) -> State:
- """Remove old :term:`SBP` product blocks from the GÉANT IP subscription."""
+ """Remove old SBP product blocks from the GÉANT IP subscription."""
subscription.l3_core_service.ap_list = [
ap
for ap in subscription.l3_core_service.ap_list
@@ -291,7 +291,7 @@ def modify_existing_sbp_blocks(subscription: L3CoreService, modified_sbp_list: l
@step("Instantiate new Service Binding Ports")
def create_new_sbp_blocks(subscription: L3CoreService, added_service_binding_ports: list[dict[str, Any]]) -> State:
- """Add new two :term:`SBP` to the L3 Core Service subscription."""
+ """Add new two SBP to the L3 Core Service subscription."""
for sbp_input in added_service_binding_ports:
edge_port = EdgePort.from_subscription(sbp_input["edge_port_id"])
bgp_session_list = [
diff --git a/gso/workflows/lan_switch_interconnect/create_imported_lan_switch_interconnect.py b/gso/workflows/lan_switch_interconnect/create_imported_lan_switch_interconnect.py
index 1439c2fe..ae26125f 100644
--- a/gso/workflows/lan_switch_interconnect/create_imported_lan_switch_interconnect.py
+++ b/gso/workflows/lan_switch_interconnect/create_imported_lan_switch_interconnect.py
@@ -1,4 +1,4 @@
-"""Import an existing :term:`LAN` Switch Interconnect into the subscription database."""
+"""Import an existing LAN Switch Interconnect into the subscription database."""
from uuid import uuid4
@@ -88,7 +88,7 @@ def initialize_subscription(
"Create Imported LAN Switch Interconnect", initial_input_form=_initial_input_form_generator, target=Target.CREATE
)
def create_imported_lan_switch_interconnect() -> StepList:
- """Create an imported :term:`LAN` Switch Interconnect without provisioning it."""
+ """Create an imported LAN Switch Interconnect without provisioning it."""
return (
begin
>> create_subscription
diff --git a/gso/workflows/lan_switch_interconnect/import_lan_switch_interconnect.py b/gso/workflows/lan_switch_interconnect/import_lan_switch_interconnect.py
index e0f247aa..0de05997 100644
--- a/gso/workflows/lan_switch_interconnect/import_lan_switch_interconnect.py
+++ b/gso/workflows/lan_switch_interconnect/import_lan_switch_interconnect.py
@@ -1,4 +1,4 @@
-"""Import an existing :term:`LAN` Switch Interconnect into the service database."""
+"""Import an existing LAN Switch Interconnect into the service database."""
from orchestrator import step, workflow
from orchestrator.targets import Target
diff --git a/gso/workflows/lan_switch_interconnect/terminate_lan_switch_interconnect.py b/gso/workflows/lan_switch_interconnect/terminate_lan_switch_interconnect.py
index 9df06c9a..ac9aef3f 100644
--- a/gso/workflows/lan_switch_interconnect/terminate_lan_switch_interconnect.py
+++ b/gso/workflows/lan_switch_interconnect/terminate_lan_switch_interconnect.py
@@ -37,7 +37,7 @@ def _input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
target=Target.TERMINATE,
)
def terminate_lan_switch_interconnect() -> StepList:
- """Terminate a :term:`LAN` Switch Interconnect."""
+ """Terminate a LAN Switch Interconnect."""
return (
begin
>> store_process_subscription(Target.TERMINATE)
diff --git a/gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py b/gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py
index f2a623de..9e6cfdd3 100644
--- a/gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py
+++ b/gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py
@@ -1,4 +1,4 @@
-"""Validation workflow for :term:`LAN` Switch Interconnect subscription objects."""
+"""Validation workflow for LAN Switch Interconnect subscription objects."""
from typing import Any
@@ -29,7 +29,7 @@ def validate_config(subscription: dict[str, Any]) -> LSOState:
"Validate LAN Switch Interconnect", target=Target.SYSTEM, initial_input_form=(wrap_modify_initial_input_form(None))
)
def validate_lan_switch_interconnect() -> StepList:
- """Validate an existing :term:`LAN` Switch Interconnect."""
+ """Validate an existing LAN Switch Interconnect."""
return (
begin
>> store_process_subscription(Target.SYSTEM)
diff --git a/gso/workflows/office_router/create_imported_office_router.py b/gso/workflows/office_router/create_imported_office_router.py
index 8b830634..aeb0b994 100644
--- a/gso/workflows/office_router/create_imported_office_router.py
+++ b/gso/workflows/office_router/create_imported_office_router.py
@@ -31,7 +31,7 @@ def create_subscription(partner: str) -> State:
def initial_input_form_generator() -> FormGenerator:
- """Generate a form that is filled in using information passed through the :term:`API` endpoint."""
+ """Generate a form that is filled in using information passed through the API endpoint."""
class ImportOfficeRouter(FormPage):
model_config = ConfigDict(title="Import an office router")
diff --git a/gso/workflows/opengear/create_imported_opengear.py b/gso/workflows/opengear/create_imported_opengear.py
index ce4bd3cf..64060fb2 100644
--- a/gso/workflows/opengear/create_imported_opengear.py
+++ b/gso/workflows/opengear/create_imported_opengear.py
@@ -29,7 +29,7 @@ def create_subscription(partner: str) -> State:
def initial_input_form_generator() -> FormGenerator:
- """Generate a form that is filled in using information passed through the :term:`API` endpoint."""
+ """Generate a form that is filled in using information passed through the API endpoint."""
class ImportOpengear(FormPage):
model_config = ConfigDict(title="Import Opengear")
diff --git a/gso/workflows/router/create_imported_router.py b/gso/workflows/router/create_imported_router.py
index cb729eb2..a12f7121 100644
--- a/gso/workflows/router/create_imported_router.py
+++ b/gso/workflows/router/create_imported_router.py
@@ -29,7 +29,7 @@ def create_subscription(partner: str) -> State:
def initial_input_form_generator() -> FormGenerator:
- """Generate a form that is filled in using information passed through the :term:`API` endpoint."""
+ """Generate a form that is filled in using information passed through the API endpoint."""
class ImportRouter(FormPage):
model_config = ConfigDict(title="Import Router")
diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index f0595fe8..cf2d4a26 100644
--- a/gso/workflows/router/create_router.py
+++ b/gso/workflows/router/create_router.py
@@ -183,9 +183,10 @@ def create_netbox_device(subscription: RouterInactive) -> State:
@step("Verify IPAM resources for loopback interface")
def verify_ipam_loopback(subscription: RouterInactive) -> None:
- """Validate the :term:`IPAM` resources for the loopback interface.
+ """Validate the IPAM resources for the loopback interface.
- Raises an :class:`orchestrator.utils.errors.ProcessFailureError` if :term:`IPAM` is configured incorrectly.
+ Raises:
+ ProcessFailureError: If IPAM is configured incorrectly.
"""
host_record = infoblox.find_host_by_fqdn(f"lo0.{subscription.router.router_fqdn}")
if not host_record or str(subscription.subscription_id) not in host_record.comment:
@@ -289,9 +290,9 @@ def create_router() -> StepList:
"""Create a new router in the service database.
* Create and initialise the subscription object in the service database
- * Allocate :term:`IPAM` resources for the loopback interface
+ * Allocate IPAM resources for the loopback interface
* Deploy configuration on the new router, first as a dry run
- * Validate :term:`IPAM` resources
+ * Validate IPAM resources
* Create a new device in Netbox
"""
router_is_nokia = conditional(lambda state: state["vendor"] == Vendor.NOKIA)
diff --git a/gso/workflows/router/import_router.py b/gso/workflows/router/import_router.py
index e9d9c4e4..1421ff0f 100644
--- a/gso/workflows/router/import_router.py
+++ b/gso/workflows/router/import_router.py
@@ -1,4 +1,4 @@
-"""A modification workflow for setting a new :term:`ISIS` metric for an IP trunk."""
+"""A modification workflow for setting a new ISIS metric for an IP trunk."""
from orchestrator.targets import Target
from orchestrator.types import State, UUIDstr
diff --git a/gso/workflows/router/update_ibgp_mesh.py b/gso/workflows/router/update_ibgp_mesh.py
index 6cee29fa..1b43273c 100644
--- a/gso/workflows/router/update_ibgp_mesh.py
+++ b/gso/workflows/router/update_ibgp_mesh.py
@@ -208,7 +208,7 @@ def prompt_radius_login() -> FormGenerator:
@step("Update subscription model")
def update_subscription_model(subscription: Router) -> State:
- """Update the database model, such that it should not be reached via :term:`OOB` access anymore."""
+ """Update the database model, such that it should not be reached via OOB access anymore."""
subscription.router.router_access_via_ts = False
return {"subscription": subscription}
diff --git a/gso/workflows/router/validate_router.py b/gso/workflows/router/validate_router.py
index 125db317..b029db47 100644
--- a/gso/workflows/router/validate_router.py
+++ b/gso/workflows/router/validate_router.py
@@ -30,9 +30,10 @@ def prepare_state(subscription_id: UUIDstr) -> State:
@step("Verify IPAM resources for loopback interface")
def verify_ipam_loopback(subscription: Router) -> None:
- """Validate the :term:`IPAM` resources for the loopback interface.
+ """Validate the IPAM resources for the loopback interface.
- Raises an :class:`orchestrator.utils.errors.ProcessFailureError` if :term:`IPAM` is configured incorrectly.
+ Raises:
+ ProcessFailureError: If IPAM is configured incorrectly.
"""
host_record = infoblox.find_host_by_fqdn(f"lo0.{subscription.router.router_fqdn}")
if not host_record or str(subscription.subscription_id) not in host_record.comment:
@@ -171,7 +172,7 @@ def verify_base_config(subscription: dict[str, Any]) -> LSOState:
def validate_router() -> StepList:
"""Validate an existing, active Router subscription.
- * Verify that the loopback interface is correctly configured in :term:`IPAM`.
+ * Verify that the loopback interface is correctly configured in IPAM.
* Verify that the router is correctly configured in Netbox.
* Verify that the router is correctly configured in LibreNMS.
* Redeploy base config to verify the configuration is intact.
diff --git a/gso/workflows/site/create_imported_site.py b/gso/workflows/site/create_imported_site.py
index 8a9b3d9e..b1190863 100644
--- a/gso/workflows/site/create_imported_site.py
+++ b/gso/workflows/site/create_imported_site.py
@@ -33,7 +33,7 @@ def create_subscription(partner: str) -> State:
def generate_initial_input_form() -> FormGenerator:
- """Generate a form that is filled in using information passed through the :term:`API` endpoint."""
+ """Generate a form that is filled in using information passed through the API endpoint."""
class ImportSite(FormPage, BaseSiteValidatorModel):
model_config = ConfigDict(title="Import Site")
diff --git a/gso/workflows/super_pop_switch/create_imported_super_pop_switch.py b/gso/workflows/super_pop_switch/create_imported_super_pop_switch.py
index da99dd68..d657c46a 100644
--- a/gso/workflows/super_pop_switch/create_imported_super_pop_switch.py
+++ b/gso/workflows/super_pop_switch/create_imported_super_pop_switch.py
@@ -32,7 +32,7 @@ def create_subscription(partner: str) -> State:
def initial_input_form_generator() -> FormGenerator:
- """Generate a form that is filled in using information passed through the :term:`API` endpoint."""
+ """Generate a form that is filled in using information passed through the API endpoint."""
class ImportSuperPopSwitch(FormPage):
model_config = ConfigDict(title="Import a Super PoP switch")
diff --git a/gso/workflows/super_pop_switch/import_super_pop_switch.py b/gso/workflows/super_pop_switch/import_super_pop_switch.py
index 6372f69a..63c7148f 100644
--- a/gso/workflows/super_pop_switch/import_super_pop_switch.py
+++ b/gso/workflows/super_pop_switch/import_super_pop_switch.py
@@ -1,4 +1,4 @@
-"""A modification workflow for migrating an ImportedSuperPopSwitch to a SuperPopSwitch subscription."""
+"""A modification workflow for migrating an ImportedSuperPoPSwitch to a SuperPopSwitch subscription."""
from orchestrator.targets import Target
from orchestrator.types import State, UUIDstr
diff --git a/gso/workflows/switch/terminate_switch.py b/gso/workflows/switch/terminate_switch.py
index da74b82c..705237f7 100644
--- a/gso/workflows/switch/terminate_switch.py
+++ b/gso/workflows/switch/terminate_switch.py
@@ -42,7 +42,7 @@ def remove_device_from_netbox(subscription: Switch) -> None:
@step("Remove switch from IPAM")
def remove_device_from_ipam(subscription: Switch) -> None:
- """Remove the switch from :term:`IPAM`."""
+ """Remove the switch from IPAM."""
delete_host_by_fqdn(subscription.switch.fqdn)
diff --git a/pyproject.toml b/pyproject.toml
index 2ae64df6..c31bcaef 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -96,6 +96,9 @@ select = [
"YTT"
]
+[tool.ruff.lint.pydocstyle]
+convention = "google"
+
[tool.ruff.lint.flake8-tidy-imports]
ban-relative-imports = "all"
diff --git a/test/conftest.py b/test/conftest.py
index ce44b074..da558c8a 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -161,8 +161,11 @@ def db_uri():
def run_migrations(db_uri: str) -> None:
"""Configure the alembic migration and run the migration on the database.
- :param str db_uri: The database uri configuration to run the migration on.
- :return: None
+ Args:
+ db_uri: The database uri configuration to run the migration on.
+
+ Returns:
+ None
"""
path = Path(__file__).resolve().parent
app_settings.DATABASE_URI = db_uri
@@ -183,7 +186,8 @@ def run_migrations(db_uri: str) -> None:
def _database(db_uri):
"""Create database and run migrations and cleanup after wards.
- :param db_uri: The database uri configuration to run the migration on.
+ Args:
+ db_uri: The database uri configuration to run the migration on.
"""
db.update(Database(db_uri))
url = make_url(db_uri)
@@ -236,7 +240,8 @@ def _db_session(_database):
- Each test runs in isolation with a pristine database state.
- Avoids the overhead of recreating the database schema or re-seeding data between tests.
- :param _database: A fixture reference that initializes the database.
+ Args:
+ _database: A fixture reference that initializes the database.
"""
with contextlib.closing(db.wrapped_database.engine.connect()) as test_connection:
# Create a new session factory for this context.
diff --git a/test/fixtures/common_fixtures.py b/test/fixtures/common_fixtures.py
index aeac5a99..2cc5b32a 100644
--- a/test/fixtures/common_fixtures.py
+++ b/test/fixtures/common_fixtures.py
@@ -24,7 +24,8 @@ def create_subscription_for_mapping(
values: a dictionary of keys from the sub_map and their corresponding test values
kwargs: The rest of the arguments
- Returns: The conforming subscription.
+ Returns:
+ The conforming subscription.
"""
def build_instance(name, value_mapping):
diff --git a/test/workflows/__init__.py b/test/workflows/__init__.py
index 0b5c9d2b..0607621d 100644
--- a/test/workflows/__init__.py
+++ b/test/workflows/__init__.py
@@ -157,10 +157,8 @@ class WorkflowInstanceForTests(LazyWorkflowInstance):
This can be as simple as merely importing a workflow function. However, if it concerns a workflow generating
function, that function will be called with or without arguments as specified.
- Returns
- -------
+ Returns:
A workflow function.
-
"""
self.workflow.name = self.name
return self.workflow
diff --git a/test/workflows/edge_port/test_create_edge_port.py b/test/workflows/edge_port/test_create_edge_port.py
index 84c87b21..8066c5e0 100644
--- a/test/workflows/edge_port/test_create_edge_port.py
+++ b/test/workflows/edge_port/test_create_edge_port.py
@@ -116,7 +116,7 @@ def test_edge_port_creation_with_invalid_input(
test_client,
):
product_id = get_product_id_by_name(ProductName.EDGE_PORT)
- # If the number of members is greater than 1 then :term:`LACP` must be enabled.
+ # If the number of members is greater than 1 then LACP must be enabled.
input_form_wizard_data[0]["enable_lacp"] = False
initial_data = [{"product": product_id}, *input_form_wizard_data]
diff --git a/test/workflows/edge_port/test_validate_edge_port.py b/test/workflows/edge_port/test_validate_edge_port.py
index 3aff8081..37f9e9dd 100644
--- a/test/workflows/edge_port/test_validate_edge_port.py
+++ b/test/workflows/edge_port/test_validate_edge_port.py
@@ -58,5 +58,5 @@ def test_validate_edge_port_success(
subscription = EdgePort.from_subscription(subscription_id)
assert subscription.status == "active"
assert mock_execute_playbook.call_count == 1
- # One time for getting the :term:`LAG` and two times for getting the interfaces
+ # One time for getting the LAG and two times for getting the interfaces
assert mock_get_interface_by_name_and_device.call_count == 3
diff --git a/utils/__init__.py b/utils/__init__.py
index 80dce5bf..838af45e 100644
--- a/utils/__init__.py
+++ b/utils/__init__.py
@@ -1 +1 @@
-"""Utilities that can be used alongside :term:`GSO`."""
+"""Utilities that can be used alongside GSO."""
diff --git a/utils/netboxcli.py b/utils/netboxcli.py
index 2cf2c1d7..3c51b00d 100644
--- a/utils/netboxcli.py
+++ b/utils/netboxcli.py
@@ -23,7 +23,7 @@ def convert_to_table(data: List[dict[str, Any]], fields: List[str]) -> pd.DataFr
@click.group()
def cli() -> None:
- """Instantiate a new :term:`CLI`."""
+ """Instantiate a new CLI."""
@cli.group()
@@ -116,7 +116,7 @@ def list() -> None: # noqa: A001
help="Interface speed to list interfaces (default 1000=1G)",
)
def interfaces(fqdn: str, speed: str) -> None:
- """List all interfaces that belong to a given :term:`FQDN`."""
+ """List all interfaces that belong to a given FQDN."""
click.echo(f"Listing all interfaces for: device with fqdn={fqdn}, speed={speed}")
interface_list = NetboxClient().get_interfaces_by_device(fqdn, speed)
display_fields = [
@@ -227,7 +227,7 @@ def deallocate_interface(fqdn: str, iface: str) -> None:
@click.option("--lag", help="LAG name to attach physical interface to")
@click.option("--iface", help="Interface name to attach to LAG")
def attach_interface_to_lag(fqdn: str, lag: str, iface: str) -> None:
- """Attach an interface to a :term:`LAG`."""
+ """Attach an interface to a LAG."""
click.echo(f"Attaching LAG to physical interface: device={fqdn}, LAG name={lag}, interface name={iface}")
attached_iface = NetboxClient().attach_interface_to_lag(fqdn, lag, iface)
click.echo(attached_iface)
@@ -238,7 +238,7 @@ def attach_interface_to_lag(fqdn: str, lag: str, iface: str) -> None:
@click.option("--lag", help="LAG name to detach from physical interface")
@click.option("--iface", help="Interface name to detach LAG from")
def detach_interface_from_lag(fqdn: str, lag: str, iface: str) -> None:
- """Detach an interface from a :term:`LAG`."""
+ """Detach an interface from a LAG."""
click.echo(f"Detaching LAG from physical interface: device={fqdn}, LAG name={lag}, interface name={iface}")
NetboxClient().detach_interfaces_from_lag(fqdn, lag)
click.echo(f"Detached LAG from physical interface: device={fqdn}, LAG name={lag}, interface name={iface}")
--
GitLab