Skip to content
Snippets Groups Projects
Commit 1a8ed1f3 authored by Karel van Klink's avatar Karel van Klink :smiley_cat: Committed by JORGE SASIAIN
Browse files

add docstrings to the tasks

parent 6b89b478
No related branches found
No related tags found
1 merge request!108NAT-315: Restore ISIS should be optional in migrate iptrunk
This commit is part of merge request !108. Comments created here will be created in the context of that merge request.
"""Task workflows that are either started by an :term:`API` endpoint, or by one of the set schedules."""
"""A creation workflow for adding an existing IP trunk to the service database."""
import ipaddress
from orchestrator import workflow
......@@ -27,6 +29,7 @@ def _generate_routers() -> dict[str, str]:
def initial_input_form_generator() -> FormGenerator:
"""Take all information passed to this workflow by the :term:`API` endpoint that was called."""
routers = _generate_routers()
router_enum = Choice("Select a router", zip(routers.keys(), routers.items(), strict=True)) # type: ignore[arg-type]
......@@ -61,6 +64,7 @@ def initial_input_form_generator() -> FormGenerator:
@step("Create a new subscription")
def create_subscription(customer: str) -> State:
"""Create a new subscription in the service database."""
customer_id = get_customer_by_name(customer)["id"]
product_id = subscriptions.get_product_id_by_name(ProductType.IP_TRUNK)
subscription = IptrunkInactive.from_product_id(product_id, customer_id)
......@@ -77,6 +81,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."""
subscription.iptrunk.iptrunk_ipv4_network = iptrunk_ipv4_network
subscription.iptrunk.iptrunk_ipv6_network = iptrunk_ipv6_network
......@@ -89,6 +94,7 @@ def update_ipam_stub_for_subscription(
target=Target.SYSTEM,
)
def import_iptrunk() -> StepList:
"""Import an IP trunk without provisioning it."""
return (
init
>> create_subscription
......
"""A creation workflow that adds an existing router to the service database."""
import ipaddress
from uuid import UUID
......@@ -10,7 +12,7 @@ from orchestrator.workflows.steps import resync, set_status, store_process_subsc
from gso.products import ProductType
from gso.products.product_blocks import router as router_pb
from gso.products.product_blocks.router import PortNumber, RouterRole, RouterVendor
from gso.products.product_blocks.router import PortNumber, RouterRole, RouterVendor, generate_fqdn
from gso.products.product_types import router
from gso.products.product_types.router import RouterInactive
from gso.products.product_types.site import Site
......@@ -35,6 +37,7 @@ def _get_site_by_name(site_name: str) -> Site:
@step("Create subscription")
def create_subscription(customer: str) -> State:
"""Create a new subscription object."""
customer_id = get_customer_by_name(customer)["id"]
product_id: UUID = subscriptions.get_product_id_by_name(ProductType.ROUTER)
subscription = RouterInactive.from_product_id(product_id, customer_id)
......@@ -46,6 +49,7 @@ def create_subscription(customer: str) -> State:
def initial_input_form_generator() -> FormGenerator:
"""Generate a form that is filled in using information passed through the :term:`API` endpoint."""
class ImportRouter(FormPage):
class Config:
title = "Import Router"
......@@ -85,14 +89,12 @@ def initialize_subscription(
router_ias_lt_ipv4_network: ipaddress.IPv4Network | None = None,
router_ias_lt_ipv6_network: ipaddress.IPv6Network | None = None,
) -> State:
"""Initialise the router subscription using input data."""
subscription.router.router_ts_port = ts_port
subscription.router.router_vendor = router_vendor
subscription.router.router_site = _get_site_by_name(router_site).site
fqdn = (
f"{hostname}.{subscription.router.router_site.site_name.lower()}."
f"{subscription.router.router_site.site_country_code.lower()}"
".geant.net"
)
router_site_obj = _get_site_by_name(router_site).site
subscription.router.router_site = router_site_obj
fqdn = generate_fqdn(hostname, router_site_obj.site_name, router_site_obj.site_country_code)
subscription.router.router_fqdn = fqdn
subscription.router.router_role = router_role
subscription.router.router_access_via_ts = True
......@@ -116,6 +118,7 @@ def initialize_subscription(
target=Target.SYSTEM,
)
def import_router() -> StepList:
"""Import a router without provisioning it."""
return (
init
>> create_subscription
......
"""A creation workflow for importing an existing site."""
from uuid import UUID
from orchestrator.forms import FormPage
......@@ -16,6 +18,10 @@ from gso.workflows.site.create_site import initialize_subscription
@step("Create subscription")
def create_subscription(customer: str) -> State:
"""Create a new subscription object in the service database.
FIXME: all attributes passed by the input form appear to be unused
"""
customer_id = get_customer_by_name(customer)["id"]
product_id: UUID = subscriptions.get_product_id_by_name(ProductType.SITE)
subscription = SiteInactive.from_product_id(product_id, customer_id)
......@@ -27,6 +33,7 @@ def create_subscription(customer: str) -> State:
def generate_initial_input_form() -> FormGenerator:
"""Generate a form that is filled in using information passed through the :term:`API` endpoint."""
class ImportSite(FormPage):
class Config:
title = "Import Site"
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment