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

resolve more linting errors, disable some pedantic rules

parent 0cf6c7c3
No related branches found
No related tags found
1 merge request!111Feature/ruff everything party hat emoji
"""The main entrypoint for :term:`GSO`, and the different ways in which it can be run."""
import typer import typer
from orchestrator import OrchestratorCore, app_settings from orchestrator import OrchestratorCore, app_settings
from orchestrator.cli.main import app as cli_app from orchestrator.cli.main import app as cli_app
...@@ -9,16 +11,19 @@ from gso.cli import netbox ...@@ -9,16 +11,19 @@ from gso.cli import netbox
def init_gso_app() -> OrchestratorCore: def init_gso_app() -> OrchestratorCore:
"""Initialise the :term:`GSO` app."""
app = OrchestratorCore(base_settings=app_settings) app = OrchestratorCore(base_settings=app_settings)
app.include_router(api_router, prefix="/api") app.include_router(api_router, prefix="/api")
return app return app
def init_worker_app() -> OrchestratorCore: def init_worker_app() -> OrchestratorCore:
"""Initialise a :term:`GSO` instance as Celery worker."""
return OrchestratorCore(base_settings=app_settings) return OrchestratorCore(base_settings=app_settings)
def init_cli_app() -> typer.Typer: def init_cli_app() -> typer.Typer:
"""Initialise :term:`GSO` as a CLI application."""
from gso.cli import import_sites from gso.cli import import_sites
cli_app.add_typer(import_sites.app, name="import_sites") cli_app.add_typer(import_sites.app, name="import_sites")
......
"""Initialisation class for the :term:`GSO` :term:`API`."""
from fastapi import APIRouter from fastapi import APIRouter
from gso.api.v1 import router as router_v1 from gso.api.v1 import router as router_v1
......
"""Version 1 of the :term:`GSO` :term:`API`."""
from fastapi import APIRouter from fastapi import APIRouter
from gso.api.v1.imports import router as imports_router from gso.api.v1.imports import router as imports_router
......
""":term:`GSO` :term:`API` endpoints that import different types of existing services."""
import ipaddress import ipaddress
from typing import Any from typing import Any
from uuid import UUID from uuid import UUID
...@@ -26,11 +27,15 @@ router = APIRouter(prefix="/imports", tags=["Imports"], dependencies=[Depends(op ...@@ -26,11 +27,15 @@ router = APIRouter(prefix="/imports", tags=["Imports"], dependencies=[Depends(op
class ImportResponseModel(BaseModel): class ImportResponseModel(BaseModel):
"""The model of a response given when services are imported using the :term:`API`."""
pid: UUID pid: UUID
detail: str detail: str
class SiteImportModel(BaseModel): class SiteImportModel(BaseModel):
"""The required input for importing an existing :class:`gso.products.product_types.site`."""
site_name: str site_name: str
site_city: str site_city: str
site_country: str site_country: str
...@@ -45,17 +50,20 @@ class SiteImportModel(BaseModel): ...@@ -45,17 +50,20 @@ class SiteImportModel(BaseModel):
@validator("site_ts_address", allow_reuse=True) @validator("site_ts_address", allow_reuse=True)
def validate_ts_address(cls, site_ts_address: str) -> str: def validate_ts_address(cls, site_ts_address: str) -> str:
"""A terminal server address must be valid."""
validate_site_fields_is_unique("site_ts_address", site_ts_address) validate_site_fields_is_unique("site_ts_address", site_ts_address)
validate_ipv4_or_ipv6(site_ts_address) validate_ipv4_or_ipv6(site_ts_address)
return site_ts_address return site_ts_address
@validator("site_country_code", allow_reuse=True) @validator("site_country_code", allow_reuse=True)
def country_code_must_exist(cls, country_code: str) -> str: def country_code_must_exist(cls, country_code: str) -> str:
"""A country code must exist."""
validate_country_code(country_code) validate_country_code(country_code)
return country_code return country_code
@validator("site_internal_id", "site_bgp_community_id", allow_reuse=True) @validator("site_internal_id", "site_bgp_community_id", allow_reuse=True)
def validate_unique_fields(cls, value: str, field: ModelField) -> str | int: def validate_unique_fields(cls, value: str, field: ModelField) -> str | int:
"""Validate that the internal side ID and :term:`BGP` community IDs are unique."""
return validate_site_fields_is_unique(field.name, value) return validate_site_fields_is_unique(field.name, value)
@validator("site_name", allow_reuse=True) @validator("site_name", allow_reuse=True)
...@@ -71,6 +79,8 @@ class SiteImportModel(BaseModel): ...@@ -71,6 +79,8 @@ class SiteImportModel(BaseModel):
class RouterImportModel(BaseModel): class RouterImportModel(BaseModel):
"""Required fields for importing an existing :class:`gso.product.product_types.router`."""
customer: str customer: str
router_site: str router_site: str
hostname: str hostname: str
...@@ -87,6 +97,8 @@ class RouterImportModel(BaseModel): ...@@ -87,6 +97,8 @@ class RouterImportModel(BaseModel):
class IptrunkImportModel(BaseModel): class IptrunkImportModel(BaseModel):
"""Required fields for importing an existing :class:`gso.products.product_types.iptrunk`."""
customer: str customer: str
geant_s_sid: str geant_s_sid: str
iptrunk_type: IptrunkType iptrunk_type: IptrunkType
...@@ -114,6 +126,7 @@ class IptrunkImportModel(BaseModel): ...@@ -114,6 +126,7 @@ class IptrunkImportModel(BaseModel):
@validator("customer") @validator("customer")
def check_if_customer_exists(cls, value: str) -> str: def check_if_customer_exists(cls, value: str) -> str:
"""The customer must exist."""
try: try:
get_customer_by_name(value) get_customer_by_name(value)
except CustomerNotFoundError as e: except CustomerNotFoundError as e:
...@@ -124,6 +137,7 @@ class IptrunkImportModel(BaseModel): ...@@ -124,6 +137,7 @@ class IptrunkImportModel(BaseModel):
@validator("side_a_node_id", "side_b_node_id") @validator("side_a_node_id", "side_b_node_id")
def check_if_router_side_is_available(cls, value: str) -> str: def check_if_router_side_is_available(cls, value: str) -> str:
"""Both sides of the trunk must exist in :term:`GSO`."""
if value not in cls._get_active_routers(): if value not in cls._get_active_routers():
msg = f"Router {value} not found" msg = f"Router {value} not found"
raise ValueError(msg) raise ValueError(msg)
...@@ -132,6 +146,7 @@ class IptrunkImportModel(BaseModel): ...@@ -132,6 +146,7 @@ class IptrunkImportModel(BaseModel):
@validator("side_a_ae_members", "side_b_ae_members") @validator("side_a_ae_members", "side_b_ae_members")
def check_side_uniqueness(cls, value: list[str]) -> list[str]: def check_side_uniqueness(cls, value: list[str]) -> list[str]:
""":term:`LAG` members must be unique."""
if len(value) != len(set(value)): if len(value) != len(set(value)):
msg = "Items must be unique" msg = "Items must be unique"
raise ValueError(msg) raise ValueError(msg)
...@@ -140,6 +155,7 @@ class IptrunkImportModel(BaseModel): ...@@ -140,6 +155,7 @@ class IptrunkImportModel(BaseModel):
@root_validator @root_validator
def check_members(cls, values: dict[str, Any]) -> dict[str, Any]: def check_members(cls, values: dict[str, Any]) -> dict[str, Any]:
"""Amount of :term:`LAG` members has to match on side A and B, and meet the minimum requirement."""
min_links = values["iptrunk_minimum_links"] min_links = values["iptrunk_minimum_links"]
side_a_members = values.get("side_a_ae_members", []) side_a_members = values.get("side_a_ae_members", [])
side_b_members = values.get("side_b_ae_members", []) side_b_members = values.get("side_b_ae_members", [])
...@@ -167,7 +183,7 @@ def _start_process(process_name: str, data: dict) -> UUID: ...@@ -167,7 +183,7 @@ def _start_process(process_name: str, data: dict) -> UUID:
detail="Failed to start the process.", detail="Failed to start the process.",
) )
process = processes._get_process(pid) process = processes._get_process(pid) # noqa: SLF001
if process.last_status == "failed": if process.last_status == "failed":
raise HTTPException( raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
......
""":term:`API` endpoint for fetching different types of subscriptions."""
from typing import Any from typing import Any
from fastapi import Depends, status from fastapi import Depends, status
......
"""The :term:`CLI` of :term:`GSO`."""
""":term:`CLI` command for importing sites."""
import typer import typer
app: typer.Typer = typer.Typer() app: typer.Typer = typer.Typer()
......
"""A :term:`CLI` for interacting with Netbox."""
import typer import typer
from pynetbox import RequestError from pynetbox import RequestError
......
...@@ -13,6 +13,8 @@ from gso.products.product_types.site import Site ...@@ -13,6 +13,8 @@ from gso.products.product_types.site import Site
class ProductType(strEnum): class ProductType(strEnum):
"""An enumerator of available products in :term:`GSO`."""
SITE = "Site" SITE = "Site"
ROUTER = "Router" ROUTER = "Router"
IP_TRUNK = "IP trunk" IP_TRUNK = "IP trunk"
......
...@@ -27,15 +27,18 @@ class PhyPortCapacity(strEnum): ...@@ -27,15 +27,18 @@ class PhyPortCapacity(strEnum):
class IptrunkType(strEnum): class IptrunkType(strEnum):
"""Types of IP trunks. Can be dark fiber or a leased line."""
DARK_FIBER = "Dark_fiber" DARK_FIBER = "Dark_fiber"
LEASED = "Leased" LEASED = "Leased"
T = TypeVar("T", covariant=True) T_co = TypeVar("T_co", covariant=True)
class LAGMemberList(UniqueConstrainedList[T_co]):
"""A list of :term:`LAG` member interfaces."""
class LAGMemberList(UniqueConstrainedList[T]): # type: ignore[type-var]
pass
class IptrunkInterfaceBlockInactive( class IptrunkInterfaceBlockInactive(
...@@ -43,22 +46,30 @@ class IptrunkInterfaceBlockInactive( ...@@ -43,22 +46,30 @@ class IptrunkInterfaceBlockInactive(
lifecycle=[SubscriptionLifecycle.INITIAL], lifecycle=[SubscriptionLifecycle.INITIAL],
product_block_name="IptrunkInterfaceBlock", product_block_name="IptrunkInterfaceBlock",
): ):
"""An inactive IP trunk interface."""
# TODO: add validation for interface names, making the type a constrained string # TODO: add validation for interface names, making the type a constrained string
interface_name: str | None = None interface_name: str | None = None
interface_description: str | None = None interface_description: str | None = None
class IptrunkInterfaceBlockProvisioning(IptrunkInterfaceBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]): class IptrunkInterfaceBlockProvisioning(IptrunkInterfaceBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
"""An IP trunk interface that is being provisioned."""
interface_name: str interface_name: str
interface_description: str interface_description: str
class IptrunkInterfaceBlock(IptrunkInterfaceBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]): class IptrunkInterfaceBlock(IptrunkInterfaceBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
"""An active IP trunk interface."""
interface_name: str interface_name: str
interface_description: str interface_description: str
class IptrunkSides(UniqueConstrainedList[T]): # type: ignore[type-var] class IptrunkSides(UniqueConstrainedList[T_co]):
"""A list of IP trunk interfaces that make up one side of a link."""
min_items = 2 min_items = 2
max_items = 2 max_items = 2
...@@ -68,6 +79,8 @@ class IptrunkSideBlockInactive( ...@@ -68,6 +79,8 @@ class IptrunkSideBlockInactive(
lifecycle=[SubscriptionLifecycle.INITIAL], lifecycle=[SubscriptionLifecycle.INITIAL],
product_block_name="IptrunkSideBlock", product_block_name="IptrunkSideBlock",
): ):
"""An inactive IP trunk side."""
iptrunk_side_node: RouterBlockInactive iptrunk_side_node: RouterBlockInactive
iptrunk_side_ae_iface: str | None = None iptrunk_side_ae_iface: str | None = None
iptrunk_side_ae_geant_a_sid: str | None = None iptrunk_side_ae_geant_a_sid: str | None = None
...@@ -75,6 +88,8 @@ class IptrunkSideBlockInactive( ...@@ -75,6 +88,8 @@ class IptrunkSideBlockInactive(
class IptrunkSideBlockProvisioning(IptrunkSideBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]): class IptrunkSideBlockProvisioning(IptrunkSideBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
"""An IP trunk side that is being provisioned."""
iptrunk_side_node: RouterBlockProvisioning iptrunk_side_node: RouterBlockProvisioning
iptrunk_side_ae_iface: str | None = None iptrunk_side_ae_iface: str | None = None
iptrunk_side_ae_geant_a_sid: str | None = None iptrunk_side_ae_geant_a_sid: str | None = None
...@@ -82,6 +97,8 @@ class IptrunkSideBlockProvisioning(IptrunkSideBlockInactive, lifecycle=[Subscrip ...@@ -82,6 +97,8 @@ class IptrunkSideBlockProvisioning(IptrunkSideBlockInactive, lifecycle=[Subscrip
class IptrunkSideBlock(IptrunkSideBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]): class IptrunkSideBlock(IptrunkSideBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
"""An active IP trunk side."""
iptrunk_side_node: RouterBlock iptrunk_side_node: RouterBlock
iptrunk_side_ae_iface: str | None = None iptrunk_side_ae_iface: str | None = None
iptrunk_side_ae_geant_a_sid: str | None = None iptrunk_side_ae_geant_a_sid: str | None = None
......
...@@ -60,6 +60,7 @@ class RouterBlockInactive( ...@@ -60,6 +60,7 @@ class RouterBlockInactive(
def generate_fqdn(hostname: str, site_name: str, country_code: str) -> str: def generate_fqdn(hostname: str, site_name: str, country_code: str) -> str:
"""Generate an :term:`FQDN` from a hostname, site name, and a country code."""
return f"{hostname}.{site_name.lower()}.{country_code.lower()}.geant.net" return f"{hostname}.{site_name.lower()}.{country_code.lower()}.geant.net"
......
...@@ -127,9 +127,7 @@ class NetboxClient: ...@@ -127,9 +127,7 @@ class NetboxClient:
"""Create a new device type in Netbox.""" """Create a new device type in Netbox."""
# First get manufacturer id # First get manufacturer id
manufacturer_id = int(self.netbox.dcim.manufacturers.get(name=manufacturer).id) manufacturer_id = int(self.netbox.dcim.manufacturers.get(name=manufacturer).id)
device_type = DeviceType( device_type = DeviceType(manufacturer=manufacturer_id, model=model, slug=slug)
**{"manufacturer": manufacturer_id, "model": model, "slug": slug}, # type: ignore[arg-type]
)
return self.netbox.dcim.device_types.create(dict(device_type)) return self.netbox.dcim.device_types.create(dict(device_type))
def create_device_role(self, name: str, slug: str) -> DeviceRole: def create_device_role(self, name: str, slug: str) -> DeviceRole:
......
...@@ -61,7 +61,6 @@ select = [ ...@@ -61,7 +61,6 @@ select = [
"F", "F",
"FA", "FA",
"FBT", "FBT",
"FIX",
"FLY", "FLY",
"FURB", "FURB",
"G", "G",
......
...@@ -8,6 +8,7 @@ from gso.services.netbox_client import NetboxClient ...@@ -8,6 +8,7 @@ from gso.services.netbox_client import NetboxClient
def convert_to_table(data: list[dict[str, Any]], fields: list[str]) -> pd.DataFrame: def convert_to_table(data: list[dict[str, Any]], fields: list[str]) -> pd.DataFrame:
"""Convert raw data into a Pandas data table."""
if not data: if not data:
msg = "No data is available for your request" msg = "No data is available for your request"
raise ValueError(msg) raise ValueError(msg)
...@@ -33,6 +34,7 @@ def create() -> None: ...@@ -33,6 +34,7 @@ def create() -> None:
@click.option("--fqdn", prompt="Enter device name", help="Device name") @click.option("--fqdn", prompt="Enter device name", help="Device name")
@click.option("--model", default="vmx", help="Device model") @click.option("--model", default="vmx", help="Device model")
def device(fqdn: str, model: str) -> None: def device(fqdn: str, model: str) -> None:
"""Create a new device in Netbox."""
click.echo(f"Creating device: fqdn={fqdn}, model={model}") click.echo(f"Creating device: fqdn={fqdn}, model={model}")
new_device = NetboxClient().create_device(fqdn, model) new_device = NetboxClient().create_device(fqdn, model)
click.echo(new_device) click.echo(new_device)
...@@ -44,6 +46,7 @@ def device(fqdn: str, model: str) -> None: ...@@ -44,6 +46,7 @@ def device(fqdn: str, model: str) -> None:
@click.option("--speed", default="1000", help="Interface speed , default is 1000") @click.option("--speed", default="1000", help="Interface speed , default is 1000")
@click.option("--fqdn", help="Device where to create interface") @click.option("--fqdn", help="Device where to create interface")
def interface(name: str, type: str, speed: str, fqdn: str) -> None: def interface(name: str, type: str, speed: str, fqdn: str) -> None:
"""Create a new interface in Netbox."""
click.echo(f"Creating interface: name={name}, speed={speed}, fqdn={fqdn}") click.echo(f"Creating interface: name={name}, speed={speed}, fqdn={fqdn}")
new_interface = NetboxClient().create_interface(name, type, speed, fqdn) new_interface = NetboxClient().create_interface(name, type, speed, fqdn)
click.echo(new_interface) click.echo(new_interface)
...@@ -53,6 +56,7 @@ def interface(name: str, type: str, speed: str, fqdn: str) -> None: ...@@ -53,6 +56,7 @@ def interface(name: str, type: str, speed: str, fqdn: str) -> None:
@click.option("--name", help="Manufacturer name") @click.option("--name", help="Manufacturer name")
@click.option("--slug", help="Short name for manufacturer") @click.option("--slug", help="Short name for manufacturer")
def manufacturer(name: str, slug: str) -> None: def manufacturer(name: str, slug: str) -> None:
"""Add a new manufacturer to Netbox."""
click.echo(f"Creating manufacturer: name={name}") click.echo(f"Creating manufacturer: name={name}")
manufacturer = NetboxClient().create_device_manufacturer(name, slug) manufacturer = NetboxClient().create_device_manufacturer(name, slug)
click.echo(manufacturer) click.echo(manufacturer)
...@@ -63,6 +67,7 @@ def manufacturer(name: str, slug: str) -> None: ...@@ -63,6 +67,7 @@ def manufacturer(name: str, slug: str) -> None:
@click.option("--model", help="Model for device") @click.option("--model", help="Model for device")
@click.option("--slug", help="Short name for manufacturer") @click.option("--slug", help="Short name for manufacturer")
def device_type(manufacturer: str, model: str, slug: str) -> None: def device_type(manufacturer: str, model: str, slug: str) -> None:
"""Create a new device type in Netbox."""
click.echo(f"Creating device type: manufacturer={manufacturer} model = {model}") click.echo(f"Creating device type: manufacturer={manufacturer} model = {model}")
device_type = NetboxClient().create_device_type(manufacturer, model, slug) device_type = NetboxClient().create_device_type(manufacturer, model, slug)
click.echo(device_type) click.echo(device_type)
...@@ -72,6 +77,7 @@ def device_type(manufacturer: str, model: str, slug: str) -> None: ...@@ -72,6 +77,7 @@ def device_type(manufacturer: str, model: str, slug: str) -> None:
@click.option("--name", help="Name for device role") @click.option("--name", help="Name for device role")
@click.option("--slug", help="Short name for device role") @click.option("--slug", help="Short name for device role")
def device_role(name: str, slug: str) -> None: def device_role(name: str, slug: str) -> None:
"""Create a new device role in Netbox."""
click.echo(f"Creating device role: name={name}") click.echo(f"Creating device role: name={name}")
device_role = NetboxClient().create_device_role(name, slug) device_role = NetboxClient().create_device_role(name, slug)
click.echo(device_role) click.echo(device_role)
...@@ -81,6 +87,7 @@ def device_role(name: str, slug: str) -> None: ...@@ -81,6 +87,7 @@ def device_role(name: str, slug: str) -> None:
@click.option("--name", help="Name for device site") @click.option("--name", help="Name for device site")
@click.option("--slug", help="Short name for device site") @click.option("--slug", help="Short name for device site")
def device_site(name: str, slug: str) -> None: def device_site(name: str, slug: str) -> None:
"""Create a new device site in Netbox."""
click.echo(f"Creating device site: name={name}") click.echo(f"Creating device site: name={name}")
device_site = NetboxClient().create_device_site(name, slug) device_site = NetboxClient().create_device_site(name, slug)
click.echo(device_site) click.echo(device_site)
...@@ -97,7 +104,7 @@ create.add_command(device_site) ...@@ -97,7 +104,7 @@ create.add_command(device_site)
# Define list commands here # Define list commands here
@cli.group() @cli.group()
def list() -> None: def list() -> None:
pass """Definitions of all listing commands."""
@list.command() @list.command()
...@@ -108,6 +115,7 @@ def list() -> None: ...@@ -108,6 +115,7 @@ def list() -> None:
help="Interface speed to list interfaces (default 1000=1G)", help="Interface speed to list interfaces (default 1000=1G)",
) )
def interfaces(fqdn: str, speed: str) -> None: def interfaces(fqdn: str, speed: str) -> None:
"""List all interfaces that belong to a given :term:`FQDN`."""
click.echo(f"Listing all interfaces for: device with fqdn={fqdn}, speed={speed}") click.echo(f"Listing all interfaces for: device with fqdn={fqdn}, speed={speed}")
interface_list = NetboxClient().get_interfaces_by_device(fqdn, speed) interface_list = NetboxClient().get_interfaces_by_device(fqdn, speed)
display_fields = [ display_fields = [
...@@ -128,12 +136,11 @@ def interfaces(fqdn: str, speed: str) -> None: ...@@ -128,12 +136,11 @@ def interfaces(fqdn: str, speed: str) -> None:
@list.command() @list.command()
def devices() -> None: def devices() -> None:
"""List all devices in Netbox."""
click.echo("Listing all devices:") click.echo("Listing all devices:")
device_list = NetboxClient().get_all_devices() device_list = NetboxClient().get_all_devices()
display_fields = ["name", "device_type"] display_fields = ["name", "device_type"]
devices = [] devices = [dict(device) for device in device_list]
for device in device_list:
devices.append(dict(device))
table = convert_to_table(devices, display_fields) table = convert_to_table(devices, display_fields)
click.echo(table) click.echo(table)
...@@ -146,12 +153,13 @@ list.add_command(devices) ...@@ -146,12 +153,13 @@ list.add_command(devices)
# Define delete commands here # Define delete commands here
@cli.group() @cli.group()
def delete() -> None: def delete() -> None:
pass """Definitions of delete commands."""
@delete.command() # type: ignore[no-redef] @delete.command() # type: ignore[no-redef]
@click.option("--fqdn", help="Name of device to delete") @click.option("--fqdn", help="Name of device to delete")
def device(fqdn: str) -> None: def device(fqdn: str) -> None:
"""Delete a device from Netbox."""
click.echo(f"Deleting device: device={fqdn}") click.echo(f"Deleting device: device={fqdn}")
NetboxClient().delete_device(fqdn) NetboxClient().delete_device(fqdn)
...@@ -160,6 +168,7 @@ def device(fqdn: str) -> None: ...@@ -160,6 +168,7 @@ def device(fqdn: str) -> None:
@click.option("--fqdn", help="Device name from where to get interface to delete") @click.option("--fqdn", help="Device name from where to get interface to delete")
@click.option("--iface", help="Name of interface name to delete") @click.option("--iface", help="Name of interface name to delete")
def interface(fqdn: str, iface: str) -> None: def interface(fqdn: str, iface: str) -> None:
"""Delete an interface from Netbox."""
click.echo(f"Deleting interface: device={fqdn}, interface name={iface}") click.echo(f"Deleting interface: device={fqdn}, interface name={iface}")
NetboxClient().delete_interface(fqdn, iface) NetboxClient().delete_interface(fqdn, iface)
...@@ -171,13 +180,14 @@ delete.add_command(interface) ...@@ -171,13 +180,14 @@ delete.add_command(interface)
# The action command # The action command
@cli.group() @cli.group()
def action() -> None: def action() -> None:
pass """Available actions."""
@action.command() @action.command()
@click.option("--fqdn", help="Device name from where to get interface to edit") @click.option("--fqdn", help="Device name from where to get interface to edit")
@click.option("--iface", help="Interface name to edit") @click.option("--iface", help="Interface name to edit")
def reserve_interface(fqdn: str, iface: str) -> None: def reserve_interface(fqdn: str, iface: str) -> None:
"""Reserve an available interface in Netbox."""
click.echo(f"Reserving interface: device ={fqdn}, interface name={iface}") click.echo(f"Reserving interface: device ={fqdn}, interface name={iface}")
reserved_iface = NetboxClient().reserve_interface(fqdn, iface) reserved_iface = NetboxClient().reserve_interface(fqdn, iface)
click.echo(reserved_iface) click.echo(reserved_iface)
...@@ -187,6 +197,7 @@ def reserve_interface(fqdn: str, iface: str) -> None: ...@@ -187,6 +197,7 @@ def reserve_interface(fqdn: str, iface: str) -> None:
@click.option("--fqdn", help="Device name from where to get interface to edit") @click.option("--fqdn", help="Device name from where to get interface to edit")
@click.option("--iface", help="Interface name to edit") @click.option("--iface", help="Interface name to edit")
def free_interface(fqdn: str, iface: str) -> None: def free_interface(fqdn: str, iface: str) -> None:
"""Mark a taken interface in Netbox as free."""
click.echo(f"Freeing interface: device={fqdn}, interface name={iface}") click.echo(f"Freeing interface: device={fqdn}, interface name={iface}")
freed_iface = NetboxClient().free_interface(fqdn, iface) freed_iface = NetboxClient().free_interface(fqdn, iface)
click.echo(freed_iface) click.echo(freed_iface)
...@@ -196,6 +207,7 @@ def free_interface(fqdn: str, iface: str) -> None: ...@@ -196,6 +207,7 @@ def free_interface(fqdn: str, iface: str) -> None:
@click.option("--fqdn", help="Device name from where to get interface to edit") @click.option("--fqdn", help="Device name from where to get interface to edit")
@click.option("--iface", help="Interface name to edit") @click.option("--iface", help="Interface name to edit")
def allocate_interface(fqdn: str, iface: str) -> None: def allocate_interface(fqdn: str, iface: str) -> None:
"""Allocate a new interface in Netbox."""
click.echo(f"Allocating interface: device={fqdn}, interface name={iface}") click.echo(f"Allocating interface: device={fqdn}, interface name={iface}")
allocated_iface = NetboxClient().allocate_interface(fqdn, iface) allocated_iface = NetboxClient().allocate_interface(fqdn, iface)
click.echo(allocated_iface) click.echo(allocated_iface)
...@@ -205,6 +217,7 @@ def allocate_interface(fqdn: str, iface: str) -> None: ...@@ -205,6 +217,7 @@ def allocate_interface(fqdn: str, iface: str) -> None:
@click.option("--fqdn", help="Device name from where to get interface to edit") @click.option("--fqdn", help="Device name from where to get interface to edit")
@click.option("--iface", help="Interface name to edit") @click.option("--iface", help="Interface name to edit")
def deallocate_interface(fqdn: str, iface: str) -> None: def deallocate_interface(fqdn: str, iface: str) -> None:
"""Deallocate an existing interface in Netbox."""
click.echo(f"Deallocating interface: device={fqdn}, interface name={iface}") click.echo(f"Deallocating interface: device={fqdn}, interface name={iface}")
deallocated_iface = NetboxClient().free_interface(fqdn, iface) deallocated_iface = NetboxClient().free_interface(fqdn, iface)
click.echo(deallocated_iface) click.echo(deallocated_iface)
...@@ -215,6 +228,7 @@ def deallocate_interface(fqdn: str, iface: str) -> None: ...@@ -215,6 +228,7 @@ def deallocate_interface(fqdn: str, iface: str) -> None:
@click.option("--lag", help="LAG name to attach physical interface to") @click.option("--lag", help="LAG name to attach physical interface to")
@click.option("--iface", help="Interface name to attach to LAG") @click.option("--iface", help="Interface name to attach to LAG")
def attach_interface_to_lag(fqdn: str, lag: str, iface: str) -> None: def attach_interface_to_lag(fqdn: str, lag: str, iface: str) -> None:
"""Attach an interface to a :term:`LAG`."""
click.echo(f"Attaching LAG to physical interface: device={fqdn}, LAG name={lag}, interface name={iface}") 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) attached_iface = NetboxClient().attach_interface_to_lag(fqdn, lag, iface)
click.echo(attached_iface) click.echo(attached_iface)
...@@ -225,6 +239,7 @@ def attach_interface_to_lag(fqdn: str, lag: str, iface: str) -> None: ...@@ -225,6 +239,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("--lag", help="LAG name to detach from physical interface")
@click.option("--iface", help="Interface name to detach LAG from") @click.option("--iface", help="Interface name to detach LAG from")
def detach_interface_from_lag(fqdn: str, lag: str, iface: str) -> None: def detach_interface_from_lag(fqdn: str, lag: str, iface: str) -> None:
"""Detach an interface from a :term:`LAG`."""
click.echo(f"Detaching LAG from physical interface: device={fqdn}, LAG name={lag}, interface name={iface}") click.echo(f"Detaching LAG from physical interface: device={fqdn}, LAG name={lag}, interface name={iface}")
NetboxClient().detach_interfaces_from_lag(fqdn, lag) NetboxClient().detach_interfaces_from_lag(fqdn, lag)
click.echo(f"Detached LAG from physical interface: device={fqdn}, LAG name={lag}, interface name={iface}") click.echo(f"Detached LAG from physical interface: device={fqdn}, LAG name={lag}, interface name={iface}")
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment