Skip to content
Snippets Groups Projects
Verified Commit 96171bb2 authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

Remove Sphinx term statements

These are not needed anymore, as MkDocs will pick up on them automatically
parent 7d18e6dd
No related branches found
No related tags found
1 merge request!316Replace Sphinx with MkDocs
Showing
with 56 additions and 55 deletions
# noqa: INP001
"""Generate the code reference pages.
Source: https://mkdocstrings.github.io/recipes/
......
# noqa: INP001
"""Generate pages to forward workflow URLs.
This will redirect `/workflow/workflow_name` to `reference/gso/workflow/module/workflow_name`. This is required since
......@@ -21,18 +22,17 @@ for path in sorted(src.rglob("*.py")):
elif parts[-1] == "__main__":
continue
if len(parts) == 4 and "workflows" in parts:
if len(parts) == 4 and "workflows" in parts: # noqa: PLR2004
redirect_map["workflow/" + parts[3] + "/index.md"] = "https://docs.gap.geant.org/reference/" + "/".join(parts)
with open(root / "wf_redirects.yaml", 'w') as redirect_file:
with Path.open(root / "wf_redirects.yaml", "w") as redirect_file:
file_content = {
"plugins":
[
"search",
{"gen-files": {"scripts": ["scripts/gen_ref_pages.py"]}},
{"redirects": {"redirect_maps": redirect_map}},
{"literate-nav": {"nav_file": "SUMMARY.md"}},
"mkdocstrings",
]
"plugins": [
"search",
{"gen-files": {"scripts": ["scripts/gen_ref_pages.py"]}},
{"redirects": {"redirect_maps": redirect_map}},
{"literate-nav": {"nav_file": "SUMMARY.md"}},
"mkdocstrings",
]
}
yaml.dump(file_content, redirect_file)
"""The main entrypoint for :term:`GSO`, and the different ways in which it can be run."""
"""The main entrypoint for GSO, and the different ways in which it can be run."""
import os
......@@ -24,13 +24,13 @@ SCALAR_OVERRIDES.update(GSO_SCALAR_OVERRIDES)
def gso_initialise_celery(celery: Celery) -> None:
"""Initialise the :term:`Celery` app."""
"""Initialise the Celery app."""
initialise_celery(celery)
celery.conf.task_routes = {}
def init_gso_app() -> OrchestratorCore:
"""Initialise the :term:`GSO` app."""
"""Initialise the GSO app."""
app = OrchestratorCore(base_settings=app_settings)
app.register_authentication(oidc_instance)
app.register_authorization(opa_instance)
......@@ -55,7 +55,7 @@ def init_gso_app() -> OrchestratorCore:
def init_cli_app() -> typer.Typer:
"""Initialise :term:`GSO` as a CLI application."""
"""Initialise GSO as a CLI application."""
from gso.cli import imports, netbox # noqa: PLC0415
cli_app.add_typer(imports.app, name="import-cli")
......
"""Initialisation class for the :term:`GSO` :term:`API`."""
"""Initialisation class for the GSO API."""
from fastapi import APIRouter
......
"""Version 1 of the :term:`GSO` :term:`API`."""
"""Version 1 of the GSO API."""
from fastapi import APIRouter
......
""":term:`API` endpoint for fetching different types of subscriptions."""
"""API endpoint for fetching different types of subscriptions."""
from typing import Any
......
"""The :term:`CLI` of :term:`GSO`."""
"""The CLI of GSO."""
""":term:`CLI` commands for importing data to coreDB."""
"""CLI commands for importing data to coreDB."""
import csv
import ipaddress
......@@ -153,7 +153,7 @@ class IptrunkImportModel(BaseModel):
@field_validator("side_a_node_id", "side_b_node_id")
def check_if_router_side_is_available(cls, value: str) -> str:
"""Both sides of the trunk must exist in :term:`GSO`."""
"""Both sides of the trunk must exist in GSO."""
if value not in cls._get_active_routers():
msg = f"Router {value} not found"
raise ValueError(msg)
......@@ -162,7 +162,7 @@ class IptrunkImportModel(BaseModel):
@model_validator(mode="after")
def check_members(self) -> Self:
"""Amount of :term:`LAG` members has to match on side A and B, and meet the minimum requirement."""
"""Amount of LAG members has to match on side A and B, and meet the minimum requirement."""
len_a = len(self.side_a_ae_members)
len_b = len(self.side_b_ae_members)
......@@ -218,7 +218,7 @@ class EdgePortImportModel(BaseModel):
@field_validator("node")
def validate_node(cls, value: str) -> str:
"""Check if the node is an active PE router in :term:`GSO`."""
"""Check if the node is an active PE router in GSO."""
pe_routers = {
str(router.subscription_id)
for router in get_active_subscriptions_by_field_and_value("router_role", RouterRole.PE)
......@@ -231,7 +231,7 @@ class EdgePortImportModel(BaseModel):
@model_validator(mode="after")
def check_members(self) -> Self:
"""Amount of :term:`LAG` members has to match and meet the minimum requirement."""
"""Amount of LAG members has to match and meet the minimum requirement."""
if len(self.ae_members) < self.minimum_links:
msg = f"Number of members should be at least {self.minimum_links} (edge_port_minimum_links)"
raise ValueError(msg)
......@@ -657,7 +657,7 @@ def import_partners(file_path: str = typer.Argument(..., help="Path to the CSV f
@app.command()
def import_l3_core_service(filepath: str = common_filepath_option) -> None:
"""Import L3 Core Services into :term:`GSO`."""
"""Import L3 Core Services into GSO."""
successfully_imported_data = []
l3_core_service_list = _read_data(Path(filepath))
......@@ -702,7 +702,7 @@ def import_l3_core_service(filepath: str = common_filepath_option) -> None:
@app.command()
def import_lan_switch_interconnect(filepath: str = common_filepath_option) -> None:
"""Import :term:`LAN` Switch Interconnect services into :term:`GSO`."""
"""Import LAN Switch Interconnect services into GSO."""
_generic_import_product(
Path(filepath),
ProductType.IMPORTED_LAN_SWITCH_INTERCONNECT,
......
"""A :term:`CLI` for interacting with Netbox."""
"""A CLI for interacting with Netbox."""
import typer
from pynetbox import RequestError
......
"""The main module that runs :term:`GSO`."""
"""The main module that runs GSO."""
from gso import init_cli_app, init_gso_app
......
"""Module that updates the domain model of :term:`GSO`. Should contain all types of subscriptions.
"""Module that updates the domain model of GSO. Should contain all types of subscriptions.
.. warning::
Whenever a new product is added, this should be reflected in the `ProductType` enumerator.
......@@ -24,7 +24,7 @@ from gso.products.product_types.vrf import VRF
class ProductName(strEnum):
"""An enumerator of available product names in :term:`GSO`."""
"""An enumerator of available product names in GSO."""
IP_TRUNK = "IP trunk"
ROUTER = "Router"
......@@ -63,7 +63,7 @@ class ProductName(strEnum):
class ProductType(strEnum):
"""An enumerator of available product types in :term:`GSO`."""
"""An enumerator of available product types in GSO."""
IP_TRUNK = Iptrunk.__name__
ROUTER = Router.__name__
......
......@@ -134,7 +134,7 @@ class IptrunkBlock(IptrunkBlockProvisioning, lifecycle=[SubscriptionLifecycle.AC
iptrunk_speed: PhysicalPortCapacity
#: The minimum amount of links the trunk should consist of.
iptrunk_minimum_links: int
#: The :term:`ISIS` metric of this link
#: The ISIS metric of this link
iptrunk_isis_metric: int
#: The IPv4 network used for this trunk.
iptrunk_ipv4_network: ipaddress.IPv4Network
......
"""Pop :term:`VLAN` product block that has all parameters of a subscription throughout its lifecycle."""
"""PoP VLAN product block that has all parameters of a subscription throughout its lifecycle."""
from ipaddress import IPv4Network, IPv6Network
from typing import Annotated, TypeVar
......@@ -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 :term:`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 :term:`VLAN` port that is being provisioned."""
"""A Pop VLAN port that is being provisioned."""
port_name: str | None
port_description: str | None
......
......@@ -55,7 +55,7 @@ class RouterBlockProvisioning(RouterBlockInactive, lifecycle=[SubscriptionLifecy
class RouterBlock(RouterBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
"""A router that's currently deployed in the network."""
#: :term:`FQDN` of a router.
#: FQDN of a router.
router_fqdn: str
#: The port of the terminal server that this router is connected to. Used to offer out of band access.
router_ts_port: PortNumber
......@@ -65,7 +65,7 @@ class RouterBlock(RouterBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTI
router_lo_ipv4_address: IPv4AddressType
#: The IPv6 loopback address of the router.
router_lo_ipv6_address: IPv6AddressType
#: The :term:`ISO` :term:`NET` of the router, used for :term:`ISIS` support.
#: 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`.
router_role: RouterRole
......
......@@ -58,23 +58,23 @@ class SiteBlockProvisioning(SiteBlockInactive, lifecycle=[SubscriptionLifecycle.
class SiteBlock(SiteBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
"""A site that's currently available for routers and services to be hosted at."""
#: The name of the site, that will dictate part of the :term:`FQDN` of routers that are hosted at this site. For
#: The name of the site, that will dictate part of the FQDN of routers that are hosted at this site. For
#: example: ``router.X.Y.geant.net``, where X denotes the name of the site.
site_name: SiteName
#: The city at which the site is located.
site_city: str
#: The country in which the site is located.
site_country: str
#: The code of the corresponding country. This is also used for the :term:`FQDN`, following the example given for
#: The code of the corresponding country. This is also used for the FQDN, following the example given for
#: the site name, the country code would end up in the Y position.
site_country_code: str
#: The latitude of the site, used for :term:`SNMP` purposes.
#: The latitude of the site, used for SNMP purposes.
site_latitude: LatitudeCoordinate
#: Similar to the latitude, the longitude of a site.
site_longitude: LongitudeCoordinate
#: The internal ID used within GÉANT to denote a site.
site_internal_id: int
#: The :term:`BGP` community ID of a site, used to advertise routes learned at this site.
#: 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`.
site_tier: SiteTier
......
"""Tasks that are scheduled to run periodically in :term:`GSO`."""
"""Tasks that are scheduled to run periodically in GSO."""
"""Definition of the decorator that schedules tasks in :term:`GSO` that are to run periodically."""
"""Definition of the decorator that schedules tasks in GSO that are to run periodically."""
import inspect
from collections.abc import Callable
......
"""Scheduled task that validates all products and inactive subscriptions in :term:`GSO`."""
"""Scheduled task that validates all products and inactive subscriptions in GSO."""
from orchestrator.services.processes import start_process
......
"""The Infoblox service that allocates :term:`IPAM` resources used in :term:`GSO` products."""
"""The Infoblox service that allocates IPAM resources used in GSO products."""
import ipaddress
from logging import getLogger
......@@ -23,7 +23,7 @@ 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 :term:`IPAM` parameters.
:return: A tuple that has an Infoblox ``Connector`` instance, and IPAM parameters.
:rtype: tuple[:class:`Connector`, IPAMParams]
"""
oss = load_oss_params().IPAM
......@@ -37,7 +37,7 @@ def _setup_connection() -> tuple[connector.Connector, IPAMParams]:
return connector.Connector(options), oss
def _allocate_network( # noqa: PLR0917
def _allocate_network(
conn: connector.Connector,
dns_view: str,
network_view: str,
......@@ -55,7 +55,7 @@ def _allocate_network( # noqa: PLR0917
: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
:term:`CIDR` notation.
CIDR notation.
:param str comment: Optionally, a comment can be added to the network allocation.
"""
for container in [ipaddress.ip_network(con) for con in containers]:
......@@ -92,7 +92,7 @@ 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 :term:`OSS`
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.
:param service_type: The service type for which the network is allocated.
......
......@@ -53,7 +53,7 @@ class Site(pydantic.BaseModel):
class NetboxClient:
"""Implement all methods to communicate with the Netbox :term:`API`."""
"""Implement all methods to communicate with the Netbox API."""
def __init__(self) -> None:
"""Instantiate a new Netbox client."""
......@@ -212,16 +212,16 @@ class NetboxClient:
iface_name: str,
description: str | None = None,
) -> Interfaces:
"""Assign a given interface to a :term:`LAG`.
"""Assign a given interface to a LAG.
Returns the interface object after assignment.
"""
iface = self.get_interface_by_name_and_device(iface_name, device_name)
# Get :term:`LAG`
# Get LAG
lag = self.get_interface_by_name_and_device(lag_name, device_name)
# Assign interface to :term:`LAG`, ensuring it does not already belong to a :term:`LAG`.
# Assign interface to LAG, ensuring it does not already belong to a LAG.
if iface.lag:
msg = f"The interface: {iface_name} on device: {device_name} already belongs to a LAG: {iface.lag.name}."
raise WorkflowStateError(msg)
......@@ -278,7 +278,7 @@ class NetboxClient:
return interface
def detach_interfaces_from_lag(self, device_name: str, lag_name: str) -> None:
"""Detach all interfaces from a :term:`LAG`."""
"""Detach all interfaces from a LAG."""
device = self.get_device_by_name(device_name)
lag = self.netbox.dcim.interfaces.get(device_id=device.id, name=lag_name)
for interface in self.netbox.dcim.interfaces.filter(
......@@ -295,7 +295,7 @@ class NetboxClient:
router_name = Router.from_subscription(router_id).router.router_fqdn
device = self.get_device_by_name(router_name)
# Get the existing :term:`LAG` interfaces for the device
# Get the existing LAG interfaces for the device
lag_interface_names = [
interface["name"] for interface in self.netbox.dcim.interfaces.filter(device=device.name, type="lag")
]
......@@ -307,7 +307,7 @@ class NetboxClient:
return [lag for lag in all_feasible_lags if lag not in lag_interface_names]
def get_available_lags(self, router_id: UUID) -> list[str]:
"""Return all available :term:`LAG` not assigned to a device."""
"""Return all available LAG not assigned to a device."""
return self.get_available_lags_in_range(router_id, FEASIBLE_IP_TRUNK_LAG_RANGE)
def get_available_services_lags(self, router_id: UUID) -> list[str]:
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment