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

add docstrings to the tasks

parent 1527f73d
No related branches found
No related tags found
1 merge request!111Feature/ruff everything party hat emoji
"""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.
Finish editing this message first!
Please register or to comment