add validators to site creation workflow

gso/products/product_blocks/site.py

 """The product block that describes a site subscription."""
-
+import re
 from typing import Optional
 
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle, strEnum
+from pydantic import ConstrainedStr
 
 
 class SiteTier(strEnum):
@@ -19,6 +20,14 @@ class SiteTier(strEnum):
     TIER4 = 4
 
 
+class SnmpCoordinate(ConstrainedStr):
+    """An SNMP coordinate, modeled as a constrained string.
+
+    The coordinate must match the format of `1.35`, `-12.3456`, etc.
+    """
+    regex = re.compile(r"^-?\d{1,2}\.\d+$")
+
+
 class SiteBlockInactive(ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="SiteBlock"):
     """A site that's currently inactive, see {class}`SiteBlock`."""
 
@@ -26,8 +35,8 @@ class SiteBlockInactive(ProductBlockModel, lifecycle=[SubscriptionLifecycle.INIT
     site_city: Optional[str] = None
     site_country: Optional[str] = None
     site_country_code: Optional[str] = None
-    site_latitude: Optional[float] = None
-    site_longitude: Optional[float] = None
+    site_latitude: Optional[SnmpCoordinate] = None
+    site_longitude: Optional[SnmpCoordinate] = None
     site_internal_id: Optional[int] = None
     site_bgp_community_id: Optional[int] = None
     site_tier: Optional[SiteTier] = None
@@ -41,8 +50,8 @@ class SiteBlockProvisioning(SiteBlockInactive, lifecycle=[SubscriptionLifecycle.
     site_city: Optional[str] = None
     site_country: Optional[str] = None
     site_country_code: Optional[str] = None
-    site_latitude: Optional[float] = None
-    site_longitude: Optional[float] = None
+    site_latitude: Optional[SnmpCoordinate] = None
+    site_longitude: Optional[SnmpCoordinate] = None
     site_internal_id: Optional[int] = None
     site_bgp_community_id: Optional[int] = None
     site_tier: Optional[SiteTier] = None
@@ -62,9 +71,9 @@ class SiteBlock(SiteBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE])
     site_country_code: str
     """The code of the corresponding country. This is also used for the {term}`FQDN`, following the example given for
     the site name, the country code would end up in the Y position."""
-    site_latitude: float
+    site_latitude: SnmpCoordinate
     """The latitude of the site, used for {term}`SNMP` purposes."""
-    site_longitude: float
+    site_longitude: SnmpCoordinate
     """Similar to the latitude, the longitude of a site."""
     site_internal_id: int
     """The internal ID used within GÉANT to denote a site."""

gso/workflows/site/create_site.py

+import ipaddress
+
 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
+import pycountry
 
 from gso.products.product_blocks import site as site_pb
 from gso.products.product_types import site
@@ -20,13 +24,48 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         site_city: str
         site_country: str
         site_country_code: str
-        site_latitude: float
-        site_longitude: float
+        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):
+            try:
+                _ = pycountry.countries.lookup(country_code)
+                #  Lookup succeeded, the country code is valid.
+                return country_code
+            except LookupError:
+                #  Lookup failed, the country code is not valid.
+                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):
+            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°.")
+
+        @validator("site_longitude", allow_reuse=True)
+        def longitude_must_be_valid(cls, longitude):
+            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°.")
+
+        @validator("site_ts_address", allow_reuse=True)
+        def ts_address_must_be_valid(cls, ts_address):
+            try:
+                ipaddress.ip_address(ts_address)
+                #  The address is valid
+                return ts_address
+            except ValueError:
+                raise ValueError("Enter a valid IPv4 or v6 address.")
+
     user_input = yield CreateSiteForm
 
     return user_input.dict()

setup.py

@@ -12,5 +12,6 @@ setup(
         "orchestrator-core==1.2.2",
         "pydantic",
         "requests",
+        "pycountry",
     ],
 )