Feature/nat 328 site names should be validated

gso/workflows/site/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 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.")
-
-    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
+    )