add validators to router creation workflow, also making IAS connection optional

gso/products/product_blocks/router.py

@@ -4,6 +4,7 @@ from typing import Optional
 
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle, strEnum
+from pydantic import ConstrainedInt
 
 from gso.products.product_blocks.site import SiteBlock, SiteBlockInactive, SiteBlockProvisioning
 
@@ -28,13 +29,25 @@ class RouterRole(strEnum):
     """AMT router."""
 
 
+class PortNumber(ConstrainedInt):
+    """Constrained integer for valid port numbers.
+
+    The range from 49152 to 65535 is marked as ephemeral, and can therefore not be selected for permanent allocation.
+    """
+
+    gt = 0
+    """The lower bound of the valid port number range."""
+    le = 49151
+    """As mentioned earlier, the ephemeral port range should not be chosen, and is therefore not available."""
+
+
 class RouterBlockInactive(
     ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="RouterBlock"
 ):
     """A router that's being currently inactive. See {class}`RouterBlock`."""
 
     router_fqdn: Optional[str] = None
-    router_ts_port: Optional[int] = None
+    router_ts_port: Optional[PortNumber] = None
     router_access_via_ts: Optional[bool] = None
     router_lo_ipv4_address: Optional[ipaddress.IPv4Address] = None
     router_lo_ipv6_address: Optional[ipaddress.IPv6Address] = None
@@ -52,7 +65,7 @@ class RouterBlockProvisioning(RouterBlockInactive, lifecycle=[SubscriptionLifecy
     """A router that's being provisioned. See {class}`RouterBlock`."""
 
     router_fqdn: str
-    router_ts_port: int
+    router_ts_port: PortNumber
     router_access_via_ts: Optional[bool] = None
     router_lo_ipv4_address: Optional[ipaddress.IPv4Address] = None
     router_lo_ipv6_address: Optional[ipaddress.IPv6Address] = None
@@ -71,9 +84,8 @@ class RouterBlock(RouterBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTI
 
     router_fqdn: str
     """{term}`FQDN` of a router."""
-    router_ts_port: int
-    """The port of the terminal server that this router is connected to. Used for the same reason as mentioned
-    previously."""
+    router_ts_port: PortNumber
+    """The port of the terminal server that this router is connected to. Used to provide out of band access."""
     router_access_via_ts: bool
     """Whether this router should be accessed through the terminal server, or through its loopback address."""
     router_lo_ipv4_address: ipaddress.IPv4Address

gso/products/product_blocks/site.py

@@ -25,6 +25,7 @@ class SnmpCoordinate(ConstrainedStr):
 
     The coordinate must match the format of `1.35`, `-12.3456`, etc.
     """
+
     regex = re.compile(r"^-?\d{1,2}\.\d+$")
 
 

gso/workflows/router/create_router.py

-import ipaddress
 import re
+from typing import Optional
 
 # noinspection PyProtectedMember
 from orchestrator.forms import FormPage
@@ -9,6 +9,7 @@ from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUID
 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.router import PortNumber
 
 from gso.products.product_blocks import router as router_pb
 from gso.products.product_types import router
@@ -16,10 +17,10 @@ from gso.products.product_types.router import RouterInactive, RouterProvisioning
 from gso.products.product_types.site import Site
 from gso.services import ipam, provisioning_proxy, subscriptions
 from gso.services.provisioning_proxy import pp_interaction
-from gso.workflows.utils import customer_selector
+from gso.workflows.utils import customer_selector, iso_from_ipv4
 
 
-def site_selector() -> Choice:
+def _site_selector() -> Choice:
     site_subscriptions = {}
     for site_id, site_description in subscriptions.get_active_site_subscriptions(
         fields=["subscription_id", "description"]
@@ -36,12 +37,12 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
             title = product_name
 
         customer: customer_selector()  # type: ignore
-        router_site: site_selector()  # type: ignore
+        router_site: _site_selector()  # type: ignore
         hostname: str
-        ts_port: int
+        ts_port: PortNumber
         router_vendor: router_pb.RouterVendor
         router_role: router_pb.RouterRole
-        is_ias_connected: bool
+        is_ias_connected: Optional[bool]
 
     user_input = yield CreateRouterForm
 
@@ -58,13 +59,6 @@ def create_subscription(product: UUIDstr, customer: UUIDstr) -> State:
     }
 
 
-def iso_from_ipv4(ipv4_address: ipaddress.IPv4Address) -> str:
-    padded_octets = [f"{x:>03}" for x in str(ipv4_address).split(".")]
-    joined_octets = "".join(padded_octets)
-    re_split = ".".join(re.findall("....", joined_octets))
-    return ".".join(["49.51e5.0001", re_split, "00"])
-
-
 @step("Get information from IPAM")
 def get_info_from_ipam(subscription: RouterProvisioning, is_ias_connected: bool) -> State:
     lo0_alias = re.sub(".geant.net", "", subscription.router.router_fqdn)
@@ -93,7 +87,7 @@ def get_info_from_ipam(subscription: RouterProvisioning, is_ias_connected: bool)
 def initialize_subscription(
     subscription: router.RouterInactive,
     hostname: str,
-    ts_port: int,
+    ts_port: PortNumber,
     router_vendor: router_pb.RouterVendor,
     router_site: str,
     router_role: router_pb.RouterRole,

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 products.product_blocks.site import SnmpCoordinate
 from pydantic import validator
-import pycountry
 
 from gso.products.product_blocks import site as site_pb
 from gso.products.product_types import site
@@ -32,7 +34,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         site_ts_address: str
 
         @validator("site_country_code", allow_reuse=True)
-        def country_code_must_exist(cls, country_code):
+        def country_code_must_exist(cls, country_code: str) -> str | NoReturn:
             try:
                 _ = pycountry.countries.lookup(country_code)
                 #  Lookup succeeded, the country code is valid.
@@ -42,23 +44,23 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
                 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):
+        def latitude_must_be_valid(cls, latitude: str) -> str | NoReturn:
             if -90 <= float(latitude) <= 90:
                 #  Check whether the value is a valid degree of latitude.
                 return latitude
-            else:
-                raise ValueError("Entered latitude is not a valid value, must be between -90.0° and 90.0°.")
+
+            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):
+        def longitude_must_be_valid(cls, longitude: str) -> str | NoReturn:
             if -180 <= float(longitude) <= 180:
                 #  Check whether the value is a valid degree of longitude.
                 return longitude
-            else:
-                raise ValueError("Entered longitude is not a valid value, must be between -180.0° and 180.0°.")
+
+            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):
+        def ts_address_must_be_valid(cls, ts_address: str) -> str | NoReturn:
             try:
                 ipaddress.ip_address(ts_address)
                 #  The address is valid
@@ -88,8 +90,8 @@ def initialize_subscription(
     site_city: str,
     site_country: str,
     site_country_code: str,
-    site_latitude: float,
-    site_longitude: float,
+    site_latitude: SnmpCoordinate,
+    site_longitude: SnmpCoordinate,
     site_bgp_community_id: int,
     site_internal_id: int,
     site_ts_address: str,

gso/workflows/utils.py

+import re
+from ipaddress import IPv4Address
+
 from orchestrator.forms.validators import Choice
 
 from gso.services.crm import all_customers
@@ -9,3 +12,15 @@ def customer_selector() -> Choice:
         customers[customer["id"]] = customer["name"]
 
     return Choice("Select a customer", zip(customers.keys(), customers.items()))  # type: ignore
+
+
+def iso_from_ipv4(ipv4_address: IPv4Address) -> str:
+    """Calculate an ISO address, based on an IPv4 address.
+
+    :param IPv4Address ipv4_address: The address that is to be converted
+    :returns: An ISO-formatted address.
+    """
+    padded_octets = [f"{x:>03}" for x in str(ipv4_address).split(".")]
+    joined_octets = "".join(padded_octets)
+    re_split = ".".join(re.findall("....", joined_octets))
+    return ".".join(["49.51e5.0001", re_split, "00"])