diff --git a/gso/auth/oidc.py b/gso/auth/oidc.py
index b65de5428bddc6abf734ec9ebeadd47fc1b2e296..1a10f6ca94964da0015bb63717db734e41208203 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 c57d80bd43ba0b10305ea78e10103e494ab70bcd..14a45ad05810386d27abc66cc68129f668e0930c 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 968abeb94a1145de0c923cdc8d27dd2030a55df7..fb535946527dd0a487e75c52d3f6856b9f4eda34 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 2a57eb69b1ea5d200f2ac96ad6f8169f203093ef..8113654ae2304be9ee1d265630925508b8eb1be3 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 3ec09b2a9efbd11616f46bfb246f7611ffedb948..b3400e0f5274957f9d0570a42e2420519fd01a9d 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 7978edd6283dae01eb03ac24302ea394d851126b..e68911dd1c3b93502ee79d5015e380860b542c72 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 add0724bee631819c1049b37be3b8132e91e8b8b..129b94778d5a170e6869099aa4434629e82ce335 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 945efc5a14ffb6f8e58b60aa375b91165f19616b..a3128a72347e90d8f5eb86d00509e441b97f2b01 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 261caa48669fd604c4182a5bb878a39b24647301..40968dfea10be3dc5ab4204fe7dd258c3a8f2194 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 bff50dd3018afd3f2bdee771cd0ca9405ae44b47..0a16b4f43a27b385b9b3dc051feaf4fcf626ad78 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 14a4db847deaf6f65bd0d849f51ddabd8debf53d..8fd88b31718c5cf255462a3af45be586ca0e6497 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 e90d665599bd64256abc5cef4ab09959f9a52b80..f4ee9c4030171932fef89e2c1c0d7e2a3511de25 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 7f64c3f5b10d4cc46b812d0a7b67b28f2b323020..5ffc2ef98cb1a10f22e27b9d52d61e941fb2715a 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 2c9cd2ce37f5029c2393bccbccb786cc0b84a6b1..6fd7467c95808ee42438f2af74a1e0ba05a1609e 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 e4d7231ea23b0ad704a46c9e7db344c6b4f54768..11c8e3364ddc94582dc637302d042e57c1fe2060 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 872f37b59042b2534de234d26241cd2792c55c3e..63f161921a030a8787d881ac4a02b2032b9c342a 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 790807ad058c6dece6eb1cab7dd055ae9bac96fd..f6bdb87769022cf8f3d4063e160d761eb888894a 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 0f83fe3714ac68bc605b77c498d56b51412d7873..51fb620b611f715ea256a3d13d2a73ee6f77ba95 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 7567f4ea8014bbd49fa66b6204839d2596cccb30..ffb6e3c64c4aa643c9e1ac930bdc1bf1a09bbd8c 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 b688ecbe4459270efc1ae788d1f2824968027977..508af69ec81c7c53cbfd9abe09724b8eb154d837 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 3139eab4fa71180b5a8eca35616b3155192cab4d..cad3fb436961519534cf403db7daa386f8764bbb 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 a958aca58e074ce4de9e32c18849ca9ebf8fc048..cb9c30932c271489974d5facf795e09e362352e0 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 b514f0009c814fcc0e698b07fbb8e86cb27e3150..0093e6f81f123fc5d6ea900e83a8ed4b9fa8b340 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 f501ed7023a52d418dedf0c88200ada8096b92f8..6bd8e41d2fdc79605b5845a2a202e7ddfcc7352f 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 c98ae7f8b108a82b7de4425b4b4fecfca13907f7..b6344b830f90a22e8237241207a1d6e861a21466 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 489f95ab9f8898bc85db8ff863929435f289bd30..4b6560b9cf78fb07ffde0e5b2604620c35844990 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 c8cfeb084b4daa4136d1326ffd90f68de2399f59..160cde0969cfa6e6bad43794716655c2df7f58c7 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 f6470a8aee5586d85bbb98204b4b734052fec4cc..c33f17a285b57513ab7dd7bfe5cf1925a80bb87e 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 d0efff0b05a75799ae05c23e0b25cdab0e5db0a4..e846f78c6a92228d75a4a5e2325753ce6f8268e1 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 add2796027d3a279b5f157f3b42187c9fd930811..9185265c09d31a4c6a010336123a407a969404b1 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 3557b4eb8e278a04a714db253c6b951a056fc100..852e152f21f00605e55b2283ad1ca06c6f37e8c3 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 a6050c9bc72de8f7554d41785b41c1e91d38a483..4272d588d31416cdd8525c82bc2460f7f14c0327 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 2251b49e998375a49d9b30c258fda6330f50585c..80ac4a5f84a804ed1ea97f8d8fade083ce5743a7 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 8ea14aa43391818219ba1d3020b2e62453229b07..c184c09bebba0583afd5628b55587d9d05f4121d 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 a9e1d05c6ad9563047764d56461afe846b99be88..56c26d56888f4cb423fe5cd8b31e1f1b224e8f3e 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 2208ca742f298bae64f1217737c673c0fd2131b3..093678e38ff37bee29a3199514b63214d57f6b3b 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 18629e98f26d188b257ad5217f42704c14da991c..2005d7c2601ec59b85c2ff05695f833695288e33 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 5aa24a28dbd02652adb9f1366e3837e9d296adf1..867965c741f2f96b7ae13df2b9b952b45f0056a0 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 67ff810b7d54b1fb8cb33a48373f47953f0f7a99..bce56c745475e4a6e2cd43adb7dc3da77be9cbf5 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 72d014c31d05b2bc3325650ff6acdea271eeffd7..ffaa023ef065000b9275e7f0d75da8443d7f3723 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 4d75437504ac6536cf56c8b2f5231bea0a937db8..5cf9c5f8bfad0b2ff9062e3ec3e058c24b6d9671 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 ef111977ae2a43366a25b3c3fbf66a124fbe75dd..f8b54c61f43cd303335c1f3eb9fa2057cfee4cc9 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 66b1f1c1e66ceb4dc525d6312195df941553d981..f5712b271bacec1c331a7ab7adb6eb0001e34736 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 b101c9c27519051ddd438a540521efed628cd204..5684dd2d10dc042032e1523e54e4b5657ba247c2 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 e76d9d2e3aebf811648e6e0384d1285ee8aba788..9c1be5cd451dadb4a039c869c3f9f67cf32836de 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 f66612f56c4fcfae5676b13c24f4b755d6b5965b..b8e879280e0da209f901a9cdf3d0350da306cfa0 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 1a94e4ef2b59e9f986d8ed182ea95c393f367a3a..a0fc39cb3afffb338b939996fb339e8d8659430e 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 0ff4c30f9664b6d5124d2fb144ca002e8027e7a0..dd77021bbd511fe6dec291279680e36cf618678d 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 f23a2fcfb2260c42dc930cd6390c05874e89242b..70686fb664e82d6c6aeeed3784e95418c82fb2ee 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 e5f1f127bf25f8c623cbd37782a8d80b4ffee78c..630418f7897079b50c587dfeb61c4463d521b784 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 1439c2fe7572c7fe89cf0e2c9fb4fcc4164b5de2..ae26125f37beb8b876a1edbc691e7c37f9787fed 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 e0f247aa28d0578ec4e5898e32f316b22c449286..0de05997aad2dc5ffd930c98f6f271174add6b82 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 9df06c9a08b046e6c8891b84ff88252b2fc0605a..ac9aef3f6d75d868555f8b23d6a9ea3c03b1a71f 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 f2a623dedb2a75716ec989b0c29795b4217541a2..9e6cfdd36d5d58c54e7b5319ed129932b068093c 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 8b8306343f6dcc29ae97ef7526c8d399e0b6e444..aeb0b99412fe685636d652437813b3e0c4e59f3d 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 ce4bd3cf039e7323ce861c11feea41f87e647947..64060fb224a19e22344cc3bc88e07a52ba84a288 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 cb729eb276c6a4734dd31c98ddd92bd1240b61cd..a12f7121f4ebf116566ef91fc5e969eb57575ba6 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 f0595fe872b071faf96987f256c31306fef26f86..cf2d4a26216538fa49251d769aa97ec5bc3ffec0 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 e9d9c4e47a04fc525c259217a5c16704c8e77751..1421ff0faac2638d15e15bea867865ea887e16ec 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 6cee29fa2774dce8add71dcabb68c94689e1b41a..1b43273ccff72b24c68ec10f59d12394d106a011 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 125db3171ee52f34029bf6d8b7e01c26034b4212..b029db47ab9233f1f3624628852f2692abbf5aec 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 8a9b3d9eaf3160e47a810a56f5c95d1421f36130..b11908638cae008d14a4d1c31b693b174c80db89 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 da99dd68249318c5f2931d6196e08c0ae6140737..d657c46a5ad6d1839aead976710b019932f5ac26 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 6372f69a70eb38da415703efdc73c1bcb6efe32e..63c7148f978a7d11eb9b97f0d3bfabb9dc71b30c 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 da74b82cfa231f28b6c812cfcd852222bf343d73..705237f7bd903658fb5fdb79696555b3e72b1491 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 2ae64df63bb2609ee3196b6d31605957a5b78a0d..c31bcaefc059e0800ece9d21387068d25f74f887 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 ce44b074025e788e8820ef5a5c8de10b7fb87c8a..da558c8a693e9a71cd2f30ff5832eac42b49684c 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 aeac5a991c3afda7b32c7da7ff9b5e223f630aab..2cc5b32a9d1e91dbc71128a088c247f2a9d46e94 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 0b5c9d2b4ea5d04e63f9fb2426ac4555f095b6b3..0607621d10c0de9066b244ba256b69cabed73104 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 84c87b21cd08b0e9b2a0fb73d6f3e5f4a9b6171a..8066c5e0217668853f7243c1d1cdfe16b4441228 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 3aff8081a8a4986751997db723cba7c293250c5c..37f9e9dd8522932f1e4c7989a42413b0ca66c6be 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 80dce5bfc818b089358bcdaf5e169ec8ff80eb3d..838af45ee878166a0b100ed56581c347dd0b6a58 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 2cf2c1d7bf0c79c3221a13a497cfc499b3813852..3c51b00d7c8f1454c66d41a8e72b90e019da35b1 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}")