Add input validation to modification workflows that contain GA- and GS-IDs

gso/services/subscriptions.py

@@ -355,7 +355,7 @@ def generate_unique_ga_id() -> str:
     """Generate a unique GA ID using the ga_id_seq database sequence.
 
     Returns:
-        str: A unique GA ID in the format `GA<number>`.
+        str: A unique GA ID in the format `GA-<number>`.
 
     Raises:
         ValueError: If there is an error generating the ID.

gso/workflows/edge_port/modify_edge_port.py

 """Modify an existing edge port subscription."""
 
+from functools import partial
 from typing import Annotated, Any, Self
 from uuid import uuid4
 
@@ -10,7 +11,7 @@ from orchestrator.targets import Target
 from orchestrator.workflow import StepList, begin, conditional, done, step
 from orchestrator.workflows.steps import resync, store_process_subscription, unsync
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
-from pydantic import AfterValidator, ConfigDict, model_validator
+from pydantic import AfterValidator, ConfigDict, Field, model_validator
 from pydantic_forms.types import FormGenerator, State, UUIDstr
 from pydantic_forms.validators import ReadOnlyField, validate_unique_list
 
@@ -26,6 +27,7 @@ from gso.utils.helpers import (
 )
 from gso.utils.types.interfaces import LAGMember, PhysicalPortCapacity
 from gso.utils.types.tt_number import TTNumber
+from gso.utils.types.unique_field import validate_field_is_unique
 
 
 def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -43,7 +45,12 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         minimum_links: int | None = subscription.edge_port.minimum_links or None
         mac_address: str | None = subscription.edge_port.mac_address or None
         ignore_if_down: bool = subscription.edge_port.ignore_if_down
-        ga_id: str | None = subscription.edge_port.ga_id or None
+        ga_id: (
+            Annotated[
+                str, AfterValidator(partial(validate_field_is_unique, subscription_id)), Field(pattern=r"^GA-\d{5}$")
+            ]
+            | None
+        ) = subscription.edge_port.ga_id or None
 
         @model_validator(mode="after")
         def validate_number_of_members(self) -> Self:

gso/workflows/iptrunk/create_imported_iptrunk.py

@@ -114,7 +114,7 @@ def initialize_subscription(
         subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
         subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
     ])
-    subscription.description = f"IP trunk {side_names[0]} {side_names[1]}, gs_id:{gs_id}"
+    subscription.description = f"IP trunk {side_names[0]} {side_names[1]}, {gs_id}"
     return {"subscription": subscription}
 
 

gso/workflows/iptrunk/create_iptrunk.py

@@ -361,7 +361,7 @@ def initialize_subscription(
             IptrunkInterfaceBlockInactive.new(subscription_id=uuid4(), **member),
         )
     side_names = sorted([side_a.router_site.site_name, side_b.router_site.site_name])
-    subscription.description = f"IP trunk {side_names[0]} {side_names[1]}, gs_id:{gs_id}"
+    subscription.description = f"IP trunk {side_names[0]} {side_names[1]}, {gs_id}"
 
     return {"subscription": subscription}
 

gso/workflows/iptrunk/migrate_iptrunk.py

@@ -769,7 +769,7 @@ def update_subscription_model(
         subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
         subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
     ])
-    subscription.description = f"IP trunk {side_names[0]} {side_names[1]}, gs_id:{subscription.iptrunk.gs_id}"
+    subscription.description = f"IP trunk {side_names[0]} {side_names[1]}, {subscription.iptrunk.gs_id}"
 
     return {"subscription": subscription}
 

gso/workflows/iptrunk/modify_trunk_interface.py

@@ -7,6 +7,7 @@ necessary modifications will be applied.
 """
 
 import json
+from functools import partial
 from typing import Annotated
 from uuid import UUID, uuid4
 
@@ -18,7 +19,7 @@ from orchestrator.utils.json import json_dumps
 from orchestrator.workflow import StepList, begin, conditional, done, step, workflow
 from orchestrator.workflows.steps import resync, store_process_subscription, unsync
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
-from pydantic import ConfigDict
+from pydantic import AfterValidator, ConfigDict, Field
 from pydantic_forms.validators import Label, ReadOnlyField
 
 from gso.products.product_blocks.iptrunk import (
@@ -39,6 +40,7 @@ from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import JuniperLAGMember, LAGMember, LAGMemberList, PhysicalPortCapacity
 from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType
 from gso.utils.types.tt_number import TTNumber
+from gso.utils.types.unique_field import validate_field_is_unique
 from gso.workflows.iptrunk.migrate_iptrunk import check_ip_trunk_optical_levels_pre
 from gso.workflows.iptrunk.validate_iptrunk import check_ip_trunk_isis
 
@@ -86,7 +88,12 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
 
     class ModifyIptrunkForm(FormPage):
         tt_number: TTNumber
-        gs_id: str | None = subscription.iptrunk.gs_id
+        gs_id: (
+            Annotated[
+                str, AfterValidator(partial(validate_field_is_unique, subscription_id)), Field(pattern=r"^GS-\d{5}$")
+            ]
+            | None
+        ) = subscription.iptrunk.gs_id
         iptrunk_description: str | None = subscription.iptrunk.iptrunk_description
         iptrunk_type: IptrunkType = subscription.iptrunk.iptrunk_type
         warning_label: Label = (
@@ -302,7 +309,7 @@ def modify_iptrunk_subscription(
         subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
         subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
     ])
-    subscription.description = f"IP trunk {side_names[0]} {side_names[1]}, gs_id:{gs_id}"
+    subscription.description = f"IP trunk {side_names[0]} {side_names[1]}, {gs_id}"
 
     return {
         "subscription": subscription,

gso/workflows/l2_circuit/create_imported_layer_2_circuit.py

@@ -23,6 +23,7 @@ from gso.products.product_types.layer_2_circuit import (
 from gso.services.partners import get_partner_by_name
 from gso.services.subscriptions import get_product_id_by_name
 from gso.utils.shared_enums import SBPType
+from gso.utils.types.geant_ids import IMPORTED_GS_ID
 from gso.utils.types.interfaces import BandwidthString
 from gso.utils.types.virtual_identifiers import VC_ID, VLAN_ID
 
@@ -39,7 +40,7 @@ def initial_input_form_generator() -> FormGenerator:
 
         service_type: Layer2CircuitServiceType
         partner: str
-        gs_id: str
+        gs_id: IMPORTED_GS_ID
         vc_id: VC_ID
         layer_2_circuit_side_a: ServiceBindingPortInput
         layer_2_circuit_side_b: ServiceBindingPortInput

gso/workflows/l3_core_service/create_imported_l3_core_service.py

@@ -21,6 +21,7 @@ from gso.products.product_types.l3_core_service import ImportedL3CoreServiceInac
 from gso.services.partners import get_partner_by_name
 from gso.services.subscriptions import get_product_id_by_name
 from gso.utils.shared_enums import SBPType
+from gso.utils.types.geant_ids import IMPORTED_GS_ID
 from gso.utils.types.ip_address import IPAddress, IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
 from gso.utils.types.virtual_identifiers import VLAN_ID
 
@@ -50,7 +51,7 @@ def initial_input_form_generator() -> FormGenerator:
     class ServiceBindingPort(BaseModel):
         edge_port: UUIDstr
         ap_type: str
-        gs_id: str
+        gs_id: IMPORTED_GS_ID
         sbp_type: SBPType = SBPType.L3
         is_tagged: bool = False
         vlan_id: VLAN_ID

test/workflows/edge_port/test_modify_edge_port.py

 from unittest.mock import patch
 
 import pytest
+from pydantic_forms.exceptions import FormValidationError
 
 from gso.products.product_types.edge_port import EdgePort
 from gso.utils.types.interfaces import PhysicalPortCapacity
@@ -37,6 +38,22 @@ def input_form_wizard_data(request, faker, edge_port_subscription_factory, partn
     ]
 
 
+@pytest.mark.workflow()
+@pytest.mark.parametrize("invalid_ga_id", ["GS-11111", "GA-1234", "GA_12345", "GA-100000"])
+def test_modify_edge_port_with_invalid_ga_id(
+    input_form_wizard_data, faker, invalid_ga_id, iptrunk_side_subscription_factory, iptrunk_subscription_factory
+):
+    input_data = input_form_wizard_data
+    input_data[1]["ga_id"] = invalid_ga_id
+    iptrunk_subscription_factory(
+        iptrunk_sides=[iptrunk_side_subscription_factory(ga_id="GA-11111"), iptrunk_side_subscription_factory()]
+    )
+
+    #  Run workflow
+    with pytest.raises(FormValidationError):
+        run_workflow("modify_edge_port", input_data)
+
+
 @pytest.mark.workflow()
 @patch("gso.services.lso_client._send_request")
 @patch("gso.services.netbox_client.NetboxClient.get_available_interfaces")

test/workflows/iptrunk/test_create_iptrunk.py

@@ -151,9 +151,7 @@ def test_successful_iptrunk_creation_with_standard_lso_result(
     ])
     assert subscription.status == "provisioning"
     assert subscription.iptrunk.gs_id is not None
-    assert subscription.description == (
-        f"IP trunk {sorted_sides[0]} {sorted_sides[1]}, gs_id:{subscription.iptrunk.gs_id}"
-    )
+    assert subscription.description == f"IP trunk {sorted_sides[0]} {sorted_sides[1]}, {subscription.iptrunk.gs_id}"
 
     assert mock_execute_playbook.call_count == 6
     #  We search for 6 hosts in total, 2 in a /31 and 4 in a /126

test/workflows/iptrunk/test_modify_trunk_interface.py

 from unittest.mock import patch
 
 import pytest
+from pydantic_forms.exceptions import FormValidationError
 
 from gso.products.product_blocks.iptrunk import IptrunkType
 from gso.products.product_types.iptrunk import Iptrunk
@@ -164,7 +165,7 @@ def test_iptrunk_modify_trunk_interface_success(
         subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
         subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
     ])
-    assert subscription.description == f"IP trunk {side_names[0]} {side_names[1]}, gs_id:{new_sid}"
+    assert subscription.description == f"IP trunk {side_names[0]} {side_names[1]}, {new_sid}"
     assert subscription.iptrunk.gs_id == input_form_iptrunk_data[1]["gs_id"]
     assert subscription.iptrunk.iptrunk_description == input_form_iptrunk_data[1]["iptrunk_description"]
     assert subscription.iptrunk.iptrunk_type == input_form_iptrunk_data[1]["iptrunk_type"]
@@ -192,3 +193,19 @@ def test_iptrunk_modify_trunk_interface_success(
             member.interface_description
             == _find_interface_by_name(new_side_b_ae_members, member.interface_name).interface_description
         )
+
+
+@pytest.mark.workflow()
+@pytest.mark.parametrize("invalid_ga_id", ["GA-11111", "GS-1234", "GS_12345", "GS-100000"])
+def test_modify_iptrunk_interface_with_invalid_ga_id(
+    input_form_iptrunk_data, faker, invalid_ga_id, iptrunk_side_subscription_factory, iptrunk_subscription_factory
+):
+    input_data = input_form_iptrunk_data
+    input_data[3]["side_a_ga_id"] = invalid_ga_id
+    iptrunk_subscription_factory(
+        iptrunk_sides=[iptrunk_side_subscription_factory(ga_id="GA-11111"), iptrunk_side_subscription_factory()]
+    )
+
+    #  Run workflow
+    with pytest.raises(FormValidationError):
+        run_workflow("modify_edge_port", input_data)