Newer
Older
Karel van Klink
committed
from typing import NoReturn
Karel van Klink
committed
import pycountry
from orchestrator.forms import FormPage
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
from orchestrator.workflow import StepList, done, init, step, workflow
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
from orchestrator.workflows.utils import wrap_create_initial_input_form
Karel van Klink
committed
from products.product_blocks.site import SnmpCoordinate
from pydantic import validator
from gso.products.product_blocks import site as site_pb
from gso.products.product_types import site
from gso.workflows.utils import customer_selector
def initial_input_form_generator(product_name: str) -> FormGenerator: # noqa: C901
class Config:
title = product_name
customer: customer_selector() # type: ignore
site_latitude: str
site_longitude: str
site_bgp_community_id: int
site_internal_id: int
site_tier: site_pb.SiteTier
@validator("site_country_code", allow_reuse=True)
Karel van Klink
committed
def country_code_must_exist(cls, country_code: str) -> str | NoReturn:
try:
_ = pycountry.countries.lookup(country_code)
return country_code
except LookupError:
raise ValueError("Invalid or non-existent country code, it must be in ISO 3166-1 alpha-2 format.")
@validator("site_latitude", allow_reuse=True)
Karel van Klink
committed
def latitude_must_be_valid(cls, latitude: str) -> str | NoReturn:
def _is_valid_latitude(degree: float) -> bool:
return -90 <= degree <= 90
if _is_valid_latitude(float(latitude)):
Karel van Klink
committed
raise ValueError("Entered latitude is not a valid value, must be between -90.0° and 90.0°.")
@validator("site_longitude", allow_reuse=True)
Karel van Klink
committed
def longitude_must_be_valid(cls, longitude: str) -> str | NoReturn:
def _is_valid_longitude(degree: float) -> bool:
return -180 <= degree <= 180
if _is_valid_longitude(float(longitude)):
Karel van Klink
committed
raise ValueError("Entered longitude is not a valid value, must be between -180.0° and 180.0°.")
@validator("site_ts_address", allow_reuse=True)
Karel van Klink
committed
def ts_address_must_be_valid(cls, ts_address: str) -> str | NoReturn:
try:
ipaddress.ip_address(ts_address)
return ts_address
except ValueError:
raise ValueError("Enter a valid IPv4 or v6 address.")
return user_input.dict()
@step("Create subscription")
def create_subscription(product: UUIDstr, customer: UUIDstr) -> State:
subscription = site.SiteInactive.from_product_id(product, customer)
return {
"subscription": subscription,
"subscription_id": subscription.subscription_id,
}
@step("Initialize subscription")
def initialize_subscription(
subscription: site.SiteInactive,
site_name: str,
site_city: str,
site_country: str,
site_country_code: str,
Karel van Klink
committed
site_latitude: SnmpCoordinate,
site_longitude: SnmpCoordinate,
site_tier: site_pb.SiteTier,
subscription.site.site_name = site_name
subscription.site.site_city = site_city
subscription.site.site_country = site_country
subscription.site.site_country_code = site_country_code
subscription.site.site_latitude = site_longitude
subscription.site.site_longitude = site_latitude
subscription.site.site_bgp_community_id = site_bgp_community_id
subscription.site.site_internal_id = site_internal_id
subscription.site.site_tier = site_tier
subscription.site.site_ts_address = site_ts_address
subscription.description = f"Site in {site_city}, {site_country}"
subscription = site.SiteProvisioning.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
return {"subscription": subscription}
@workflow(
initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
init
>> create_subscription
>> store_process_subscription(Target.CREATE)
>> initialize_subscription
>> set_status(SubscriptionLifecycle.ACTIVE)
>> resync
>> done