Skip to content
Snippets Groups Projects
Commit 0eda1167 authored by Hakan Calim's avatar Hakan Calim
Browse files

NAT-328: fixed documentation issue from vale

parent 5f3148d1
No related branches found
No related tags found
No related merge requests found
Pipeline #84354 passed
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
)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment