diff --git a/gso/workflows/site/create_site.py b/gso/workflows/site/create_site.py index 33aea63ef0ae725e78d5aff3115a682205a1e129..1ca2cd85858f6e3ee92da1f4a7a2b66e20f8e696 100644 --- a/gso/workflows/site/create_site.py +++ b/gso/workflows/site/create_site.py @@ -1,123 +1,123 @@ -import ipaddress -import re -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 pydantic import validator - -from gso.products.product_blocks import site as site_pb -from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate -from gso.products.product_types import site -from gso.services.crm 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[valid-type] - site_name: str - site_city: str - site_country: str - site_country_code: str - site_latitude: LatitudeCoordinate - site_longitude: LongitudeCoordinate - 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_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.") - - @validator("site_name", allow_reuse=True) - def site_name_must_be_valid(cls, site_name: str) -> str | NoReturn: - # Accept 3 chapital letters and only one ditigt after that terminate with $ - pattern = re.compile(r"^[A-Z]{3}[0-9]?$") - if not bool(pattern.match(site_name)): - raise ValueError(f"Enter a valid site name similar looks like AMS, AMS1or LON9. Get: {site_name}") - - return site_name - - 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: LatitudeCoordinate, - site_longitude: LongitudeCoordinate, - 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_latitude - subscription.site.site_longitude = site_longitude - 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 - ) +import ipaddress +import re +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 pydantic import validator + +from gso.products.product_blocks import site as site_pb +from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate +from gso.products.product_types import site +from gso.services.crm 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[valid-type] + site_name: str + site_city: str + site_country: str + site_country_code: str + site_latitude: LatitudeCoordinate + site_longitude: LongitudeCoordinate + 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_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.") + + @validator("site_name", allow_reuse=True) + def site_name_must_be_valid(cls, site_name: str) -> str | NoReturn: + # Accept 3 chapital letters and only one ditigt after capital letters. + pattern = re.compile(r"^[A-Z]{3}[0-9]?$") + if not bool(pattern.match(site_name)): + raise ValueError(f"Enter a valid site name similar looks like AMS, AMS1or LON9. Get: {site_name}") + + return site_name + + 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: LatitudeCoordinate, + site_longitude: LongitudeCoordinate, + 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_latitude + subscription.site.site_longitude = site_longitude + 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 + )