create_site.py

import ipaddress
from typing import NoReturn

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
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 CreateSiteForm(FormPage):
        class Config:
            title = product_name

        customer: customer_selector()  # type: ignore
        site_name: str
        site_city: str
        site_country: str
        site_country_code: str
        site_latitude: str
        site_longitude: str
        site_bgp_community_id: int
        site_internal_id: int
        site_tier: site_pb.SiteTier
        site_ts_address: str

        @validator("site_country_code", allow_reuse=True)
        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)
        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)):
                return latitude

            raise ValueError("Entered latitude is not a valid value, must be between -90.0° and 90.0°.")

        @validator("site_longitude", allow_reuse=True)
        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)):
                return longitude

            raise ValueError("Entered longitude is not a valid value, must be between -180.0° and 180.0°.")

        @validator("site_ts_address", allow_reuse=True)
        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.")

    user_input = yield CreateSiteForm

    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,
    site_latitude: SnmpCoordinate,
    site_longitude: SnmpCoordinate,
    site_bgp_community_id: int,
    site_internal_id: int,
    site_ts_address: str,
    site_tier: site_pb.SiteTier,
) -> State:
    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(
    "Create Site",
    initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
    target=Target.CREATE,
)
def create_site() -> StepList:
    return (
        init
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )