fix tests

gso/__init__.py

@@ -28,7 +28,7 @@ def init_worker_app() -> OrchestratorCore:
 
 def init_cli_app() -> typer.Typer:
     """Initialise :term:`GSO` as a CLI application."""
-    from gso.cli import imports, netbox  # noqa: PLC0415
+    from gso.cli import imports, netbox
 
     cli_app.add_typer(imports.app, name="import-cli")
     cli_app.add_typer(netbox.app, name="netbox-cli")

gso/api/v1/imports.py

@@ -12,11 +12,10 @@ from pydantic import BaseModel, field_validator, model_validator
 from gso.auth.security import opa_security_default
 from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
 from gso.products.product_blocks.router import RouterRole
-from gso.products.product_blocks.site import SiteTier
 from gso.services import subscriptions
 from gso.services.partners import PartnerNotFoundError, get_partner_by_name
 from gso.utils.helpers import BaseSiteValidatorModel, LAGMember
-from gso.utils.shared_enums import PortNumber, Vendor
+from gso.utils.shared_enums import FancyIPV4Address, FancyIPV6Address, PortNumber, Vendor
 
 router = APIRouter(prefix="/imports", tags=["Imports"], dependencies=[Depends(opa_security_default)])
 
@@ -31,16 +30,6 @@ class ImportResponseModel(BaseModel):
 class SiteImportModel(BaseSiteValidatorModel):
     """The required input for importing an existing :class:`gso.products.product_types.site`."""
 
-    site_name: str
-    site_city: str
-    site_country: str
-    site_country_code: str
-    site_latitude: float
-    site_longitude: float
-    site_bgp_community_id: int
-    site_internal_id: int
-    site_tier: SiteTier
-    site_ts_address: str
     partner: str
 
 
@@ -53,8 +42,8 @@ class RouterImportModel(BaseModel):
     ts_port: int
     router_vendor: Vendor
     router_role: RouterRole
-    router_lo_ipv4_address: ipaddress.IPv4Address
-    router_lo_ipv6_address: ipaddress.IPv6Address
+    router_lo_ipv4_address: FancyIPV4Address
+    router_lo_ipv6_address: FancyIPV6Address
     router_lo_iso_address: str
 
 
@@ -140,7 +129,7 @@ class SuperPopSwitchImportModel(BaseModel):
     super_pop_switch_site: str
     hostname: str
     super_pop_switch_ts_port: PortNumber
-    super_pop_switch_mgmt_ipv4_address: ipaddress.IPv4Address
+    super_pop_switch_mgmt_ipv4_address: FancyIPV4Address
 
 
 class OfficeRouterImportModel(BaseModel):
@@ -150,8 +139,8 @@ class OfficeRouterImportModel(BaseModel):
     office_router_site: str
     office_router_fqdn: str
     office_router_ts_port: PortNumber
-    office_router_lo_ipv4_address: ipaddress.IPv4Address
-    office_router_lo_ipv6_address: ipaddress.IPv6Address
+    office_router_lo_ipv4_address: FancyIPV4Address
+    office_router_lo_ipv6_address: FancyIPV6Address
 
 
 def _start_process(process_name: str, data: dict) -> UUID:

gso/api/v1/subscriptions.py

 """:term:`API` endpoint for fetching different types of subscriptions."""
-
 from typing import Any
 
 from fastapi import Depends, Response, status
@@ -20,6 +19,17 @@ router = APIRouter(
 )
 
 
+# class MySubscriptionDomainModelSchema(SubscriptionDomainModelSchema):
+#     model_config = ConfigDict(
+#         extra="allow",
+#         json_encoders={
+#             # datetime: lambda dt: dt.timestamp(),
+#             ipaddress.IPv4Address: lambda v: 1/0,
+#             ipaddress.IPv6Address: lambda v: str(v),
+#         }
+#     )
+
+
 @router.get(
     "/routers",
     status_code=status.HTTP_200_OK,

gso/monkeypatches.py

@@ -3,9 +3,12 @@
 This adjustment is typically done to extend or modify the functionality of the original
 oauth2_lib package to meet specific requirements of the gso application.
 """
+from datetime import datetime
+from ipaddress import IPv4Address, IPv6Address
 
 import oauth2_lib.fastapi
 import oauth2_lib.settings
+from pydantic import BaseModel
 
 from gso.auth.oidc_policy_helper import HTTPX_SSL_CONTEXT, OIDCUser, OIDCUserModel, opa_decision
 from gso.auth.settings import oauth2lib_settings
@@ -15,3 +18,9 @@ oauth2_lib.fastapi.OIDCUserModel = OIDCUserModel  # type: ignore[assignment, mis
 oauth2_lib.fastapi.opa_decision = opa_decision  # type: ignore[assignment]
 oauth2_lib.fastapi.HTTPX_SSL_CONTEXT = HTTPX_SSL_CONTEXT
 oauth2_lib.settings.oauth2lib_settings = oauth2lib_settings  # type: ignore[assignment]
+
+BaseModel.model_config["json_encoders"] = {
+    datetime: lambda dt: dt.timestamp(),
+    IPv4Address: lambda v: str(v),
+    IPv6Address: lambda v: str(v),
+}

gso/oss-params-example.json

@@ -45,7 +45,7 @@
     },
     "LT_IAS": {
       "V4": {"containers": ["10.255.255.0/24"], "networks": [], "mask": 31},
-      "V6": {"containers": ["dead:beef:cc::/48"], "networks": [], "mask": 126},
+      "V6": {"containers": [ "2001:798:1::/48"], "networks": [], "mask": 126},
       "domain_name": ".geantip",
       "dns_view": "default",
       "network_view": "default"
@@ -73,6 +73,7 @@
   "PROVISIONING_PROXY": {
     "scheme": "https",
     "api_base": "localhost:44444",
+    "auth": "Bearer <token>",
     "api_version": 1123
   },
   "CELERY": {

gso/products/product_blocks/iptrunk.py

 """IP trunk product block that has all parameters of a subscription throughout its lifecycle."""
 
 import ipaddress
-from typing import Annotated, TypeVar
+from typing import Annotated
 
 from annotated_types import Len
 from orchestrator.domain.base import ProductBlockModel, T
@@ -35,8 +35,8 @@ class IptrunkType(strEnum):
     LEASED = "Leased"
 
 
-# A list of :term:`LAG` member interfaces.
 LAGMemberList = Annotated[list[T], AfterValidator(validate_unique_list), Len(min_length=0, max_length=None)]
+IptrunkSides = Annotated[list[T], AfterValidator(validate_unique_list), Len(min_length=2, max_length=2)]
 
 
 class IptrunkInterfaceBlockInactive(
@@ -64,16 +64,6 @@ class IptrunkInterfaceBlock(IptrunkInterfaceBlockProvisioning, lifecycle=[Subscr
     interface_description: str | None = None
 
 
-def validate_unique_list(value):
-    if len(value) != len(set(value)):
-        raise ValueError("List items must be unique")
-    return value
-
-
-T_co = TypeVar("T_co", covariant=True)
-IptrunkSides = Annotated[list[T_co], AfterValidator(validate_unique_list), Len(min_length=2, max_length=2)]
-
-
 class IptrunkSideBlockInactive(
     ProductBlockModel,
     lifecycle=[SubscriptionLifecycle.INITIAL],

gso/products/product_blocks/office_router.py

 """Product block for :class:`office router` products."""
 
-import ipaddress
 
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle
@@ -10,7 +9,7 @@ from gso.products.product_blocks.site import (
     SiteBlockInactive,
     SiteBlockProvisioning,
 )
-from gso.utils.shared_enums import PortNumber, Vendor
+from gso.utils.shared_enums import FancyIPV4Address, FancyIPV6Address, PortNumber, Vendor
 
 
 class OfficeRouterBlockInactive(
@@ -22,8 +21,8 @@ class OfficeRouterBlockInactive(
 
     office_router_fqdn: str | None = None
     office_router_ts_port: PortNumber | None = None
-    office_router_lo_ipv4_address: ipaddress.IPv4Address | None = None
-    office_router_lo_ipv6_address: ipaddress.IPv6Address | None = None
+    office_router_lo_ipv4_address: FancyIPV4Address | None = None
+    office_router_lo_ipv6_address: FancyIPV6Address | None = None
     office_router_site: SiteBlockInactive | None
     vendor: Vendor | None = None
 
@@ -33,8 +32,8 @@ class OfficeRouterBlockProvisioning(OfficeRouterBlockInactive, lifecycle=[Subscr
 
     office_router_fqdn: str | None = None
     office_router_ts_port: PortNumber | None = None
-    office_router_lo_ipv4_address: ipaddress.IPv4Address | None = None
-    office_router_lo_ipv6_address: ipaddress.IPv6Address | None = None
+    office_router_lo_ipv4_address: FancyIPV4Address | None = None
+    office_router_lo_ipv6_address: FancyIPV6Address | None = None
     office_router_site: SiteBlockProvisioning | None
     vendor: Vendor | None = None
 
@@ -47,9 +46,9 @@ class OfficeRouterBlock(OfficeRouterBlockProvisioning, lifecycle=[SubscriptionLi
     #:  The port of the terminal server that this office router is connected to. Used to offer out of band access.
     office_router_ts_port: PortNumber
     #:  The IPv4 loopback address of the office router.
-    office_router_lo_ipv4_address: ipaddress.IPv4Address
+    office_router_lo_ipv4_address: FancyIPV4Address
     #:  The IPv6 loopback address of the office router.
-    office_router_lo_ipv6_address: ipaddress.IPv6Address
+    office_router_lo_ipv6_address: FancyIPV6Address
     #:  The :class:`Site` that this office router resides in. Both physically and computationally.
     office_router_site: SiteBlock
     #:  The vendor of an office router. Defaults to Juniper.

gso/products/product_blocks/router.py

 """Product block for :class:`Router` products."""
 
-import ipaddress
-
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle, strEnum
 
@@ -10,7 +8,7 @@ from gso.products.product_blocks.site import (
     SiteBlockInactive,
     SiteBlockProvisioning,
 )
-from gso.utils.shared_enums import PortNumber, Vendor
+from gso.utils.shared_enums import FancyIPV4Address, FancyIPV6Address, PortNumber, Vendor
 
 
 class RouterRole(strEnum):
@@ -31,8 +29,8 @@ class RouterBlockInactive(
     router_fqdn: str | None = None
     router_ts_port: PortNumber | None = None
     router_access_via_ts: bool | None = None
-    router_lo_ipv4_address: ipaddress.IPv4Address | None = None
-    router_lo_ipv6_address: ipaddress.IPv6Address | None = None
+    router_lo_ipv4_address: FancyIPV4Address | None = None
+    router_lo_ipv6_address: FancyIPV6Address | None = None
     router_lo_iso_address: str | None = None
     router_role: RouterRole | None = None
     router_site: SiteBlockInactive | None
@@ -45,8 +43,8 @@ class RouterBlockProvisioning(RouterBlockInactive, lifecycle=[SubscriptionLifecy
     router_fqdn: str
     router_ts_port: PortNumber
     router_access_via_ts: bool
-    router_lo_ipv4_address: ipaddress.IPv4Address
-    router_lo_ipv6_address: ipaddress.IPv6Address
+    router_lo_ipv4_address: FancyIPV4Address
+    router_lo_ipv6_address: FancyIPV6Address
     router_lo_iso_address: str
     router_role: RouterRole
     router_site: SiteBlockProvisioning
@@ -63,9 +61,9 @@ class RouterBlock(RouterBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTI
     #:  Whether this router should be accessed through the terminal server, or through its loopback address.
     router_access_via_ts: bool
     #:  The IPv4 loopback address of the router.
-    router_lo_ipv4_address: ipaddress.IPv4Address
+    router_lo_ipv4_address: FancyIPV4Address
     #:  The IPv6 loopback address of the router.
-    router_lo_ipv6_address: ipaddress.IPv6Address
+    router_lo_ipv6_address: FancyIPV6Address
     #:  The :term:`ISO` :term:`NET` of the router, used for :term:`ISIS` support.
     router_lo_iso_address: str
     #:  The role of the router, which can be any of the values defined in :class:`RouterRole`.

gso/products/product_blocks/site.py

@@ -36,7 +36,7 @@ def validate_longitude(v: float) -> float:
     return v
 
 
-LatitudeCoordinate = Annotated[
+LatitudeCoordinate: type[float] = Annotated[
     float,
     Field(
         ge=-90,

gso/products/product_blocks/super_pop_switch.py

 """Product block for :class:`Super PoP Switch` products."""
 
-import ipaddress
 
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle
@@ -10,7 +9,7 @@ from gso.products.product_blocks.site import (
     SiteBlockInactive,
     SiteBlockProvisioning,
 )
-from gso.utils.shared_enums import PortNumber, Vendor
+from gso.utils.shared_enums import FancyIPV4Address, PortNumber, Vendor
 
 
 class SuperPopSwitchBlockInactive(
@@ -22,7 +21,7 @@ class SuperPopSwitchBlockInactive(
 
     super_pop_switch_fqdn: str | None = None
     super_pop_switch_ts_port: PortNumber | None = None
-    super_pop_switch_mgmt_ipv4_address: ipaddress.IPv4Address | None = None
+    super_pop_switch_mgmt_ipv4_address: FancyIPV4Address | None = None
     super_pop_switch_site: SiteBlockInactive | None
     vendor: Vendor | None = None
 
@@ -32,7 +31,7 @@ class SuperPopSwitchBlockProvisioning(SuperPopSwitchBlockInactive, lifecycle=[Su
 
     super_pop_switch_fqdn: str | None = None
     super_pop_switch_ts_port: PortNumber | None = None
-    super_pop_switch_mgmt_ipv4_address: ipaddress.IPv4Address | None = None
+    super_pop_switch_mgmt_ipv4_address: FancyIPV4Address | None = None
     super_pop_switch_site: SiteBlockProvisioning | None
     vendor: Vendor | None = None
 
@@ -45,7 +44,7 @@ class SuperPopSwitchBlock(SuperPopSwitchBlockProvisioning, lifecycle=[Subscripti
     #:  The port of the terminal server that this Super PoP switch is connected to. Used to offer out of band access.
     super_pop_switch_ts_port: PortNumber
     #:  The IPv4 management address of the Super PoP switch.
-    super_pop_switch_mgmt_ipv4_address: ipaddress.IPv4Address
+    super_pop_switch_mgmt_ipv4_address: FancyIPV4Address
     #:  The :class:`Site` that this Super PoP switch resides in. Both physically and computationally.
     super_pop_switch_site: SiteBlock
     #:  The vendor of a Super PoP switch. Defaults to Juniper.

gso/schedules/scheduling.py

@@ -29,6 +29,7 @@ def scheduler(
 
     All time units can be specified with lists of numbers or crontab pattern strings for advanced scheduling.
     All specified time parts (minute, hour, day, etc.) must align for a task to run.
+
     """
 
     def decorator(task_func: Callable) -> Callable:

gso/services/infoblox.py

@@ -10,6 +10,7 @@ from infoblox_client.exceptions import (
 )
 
 from gso.settings import IPAMParams, load_oss_params
+from gso.utils.shared_enums import FancyIPV4Address, FancyIPV6Address
 
 logger = getLogger(__name__)
 NULL_MAC = "00:00:00:00:00:00"
@@ -233,8 +234,8 @@ def allocate_host(
 
 def create_host_by_ip(
     hostname: str,
-    ipv4_address: ipaddress.IPv4Address,
-    ipv6_address: ipaddress.IPv6Address,
+    ipv4_address: FancyIPV4Address,
+    ipv6_address: FancyIPV6Address,
     service_type: str,
     comment: str,
 ) -> None:
@@ -268,11 +269,11 @@ def create_host_by_ip(
     new_host.update()
 
 
-def find_host_by_ip(ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> objects.HostRecord | None:
+def find_host_by_ip(ip_addr: FancyIPV4Address | ipaddress.IPv6Address) -> objects.HostRecord | None:
     """Find a host record in Infoblox by its associated IP address.
 
     :param ip_addr: The IP address of a host that is searched for.
-    :type ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address
+    :type ip_addr: FancyIPV4Address | ipaddress.IPv6Address
     """
     conn, _ = _setup_connection()
     if ip_addr.version == 4:  # noqa: PLR2004, the 4 in IPv4 is well-known and not a "magic value."
@@ -314,14 +315,14 @@ def find_v6_host_by_fqdn(fqdn: str) -> objects.HostRecordV6:
     )
 
 
-def delete_host_by_ip(ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> None:
+def delete_host_by_ip(ip_addr: FancyIPV4Address | ipaddress.IPv6Address) -> None:
     """Delete a host from Infoblox.
 
     Delete a host record in Infoblox, by providing the IP address that is associated with the record. Raises a
     :class:`DeletionError` if no record can be found in Infoblox.
 
     :param ip_addr: The IP address of the host record that should get deleted.
-    :type ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address
+    :type ip_addr: FancyIPV4Address | ipaddress.IPv6Address
     """
     host = find_host_by_ip(ip_addr)
     if host:

gso/services/lso_client.py

@@ -130,8 +130,8 @@ def _show_results(state: State) -> FormGenerator:
 
         if "lso_result_extra_label" in state:
             extra_label: Label = state["lso_result_extra_label"]
-        run_status: str = ReadOnlyField(state["callback_result"]["status"])
-        run_results: LongText = ReadOnlyField(json.dumps(state["callback_result"], indent=4))
+        run_status: ReadOnlyField(state["callback_result"]["status"], default_type=str)
+        run_results: ReadOnlyField(json.dumps(state["callback_result"], indent=4), default_type=LongText)
 
     yield ConfirmRunPage
     [state.pop(key, None) for key in ["run_results", "lso_result_title", "lso_result_extra_label"]]

gso/settings.py

@@ -9,8 +9,9 @@ import json
 import logging
 import os
 from pathlib import Path
+from typing import Annotated
 
-from pydantic import NonNegativeInt
+from pydantic import Field
 from pydantic_settings import BaseSettings
 
 logger = logging.getLogger(__name__)
@@ -45,16 +46,8 @@ class InfoBloxParams(BaseSettings):
     password: str
 
 
-class V4Netmask(NonNegativeInt):
-    """A valid netmask for an IPv4 network or address."""
-
-    le = 32
-
-
-class V6Netmask(NonNegativeInt):
-    """A valid netmask for an IPv6 network or address."""
-
-    le = 128
+V4Netmask = Annotated[int, Field(ge=0, le=32)]
+V6Netmask = Annotated[int, Field(ge=0, le=128)]
 
 
 class V4NetworkParams(BaseSettings):

gso/utils/helpers.py

@@ -3,7 +3,6 @@
 import ipaddress
 import re
 from enum import StrEnum
-from ipaddress import IPv4Address
 from uuid import UUID
 
 import pycountry
@@ -13,11 +12,11 @@ from pydantic_forms.validators import Choice
 
 from gso import settings
 from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
-from gso.products.product_blocks.site import SiteTier
+from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate, SiteTier
 from gso.products.product_types.router import Router
 from gso.services.netbox_client import NetboxClient
 from gso.services.subscriptions import get_active_subscriptions_by_field_and_value
-from gso.utils.shared_enums import Vendor
+from gso.utils.shared_enums import FancyIPV4Address, Vendor
 
 
 class LAGMember(BaseModel):
@@ -106,7 +105,7 @@ def get_router_vendor(router_id: UUID) -> Vendor:
     return Router.from_subscription(router_id).router.vendor
 
 
-def iso_from_ipv4(ipv4_address: IPv4Address) -> str:
+def iso_from_ipv4(ipv4_address: FancyIPV4Address) -> str:
     """Calculate an :term:`ISO` address, based on an IPv4 address.
 
     :param IPv4Address ipv4_address: The address that's to be converted
@@ -209,47 +208,47 @@ class BaseSiteValidatorModel(BaseModel):
     site_internal_id: int
     site_tier: SiteTier
     site_ts_address: str
+    site_country_code: str
+    site_name: str
+    site_city: str
+    site_country: str
+    site_latitude: LatitudeCoordinate
+    site_longitude: LongitudeCoordinate
 
-    @classmethod
-    @field_validator("site_ts_address", check_fields=False)
+
+    @field_validator("site_ts_address")
     def validate_ts_address(cls, site_ts_address: str) -> str:
         """Validate that a terminal server address is valid."""
         validate_ipv4_or_ipv6(site_ts_address)
         return site_ts_address
 
-    @classmethod
-    @field_validator("site_country_code", check_fields=False)
+    @field_validator("site_country_code")
     def country_code_must_exist(cls, country_code: str) -> str:
         """Validate that the country code exists."""
         validate_country_code(country_code)
         return country_code
 
-    @classmethod
-    @field_validator("site_ts_address", check_fields=False)
+    @field_validator("site_ts_address")
     def site_ts_address_must_be_unique(cls, site_ts_address: str) -> str:
         """Validate that the internal and :term:`BGP` community IDs are unique."""
         return validate_site_fields_is_unique("site_ts_address", site_ts_address)
 
-    @classmethod
-    @field_validator("site_internal_id", check_fields=False)
+    @field_validator("site_internal_id")
     def site_internal_id_must_be_unique(cls, site_internal_id: int) -> int:
         """Validate that the internal and :term:`BGP` community IDs are unique."""
         return validate_site_fields_is_unique("site_internal_id", site_internal_id)
 
-    @classmethod
-    @field_validator("site_bgp_community_id", check_fields=False)
+    @field_validator("site_bgp_community_id")
     def site_bgp_community_id_must_be_unique(cls, site_bgp_community_id: int) -> int:
         """Validate that the internal and :term:`BGP` community IDs are unique."""
         return validate_site_fields_is_unique("site_bgp_community_id", site_bgp_community_id)
 
-    @classmethod
-    @field_validator("site_name", check_fields=False)
+    @field_validator("site_name")
     def site_name_must_be_unique(cls, site_name: str) -> str:
         """Validate that the internal and :term:`BGP` community IDs are unique."""
         return validate_site_fields_is_unique("site_name", site_name)
 
-    @classmethod
-    @field_validator("site_name", check_fields=False)
+    @field_validator("site_name")
     def site_name_must_be_valid(cls, site_name: str) -> str:
         """Validate the site name.
 

gso/utils/shared_enums.py

 """Shared choices for the different models."""
-
+import ipaddress
 from typing import Annotated
 
-from pydantic import Field
+from pydantic import Field, PlainSerializer
 from pydantic_forms.types import strEnum
 
 
@@ -26,8 +26,18 @@ PortNumber = Annotated[
 ]
 
 
+FancyIPV4Address = Annotated[
+    ipaddress.IPv4Address, PlainSerializer(lambda ip: str(ip), return_type=str, when_used="always")
+]
+
+FancyIPV6Address = Annotated[
+    ipaddress.IPv6Address, PlainSerializer(lambda ip: str(ip),  return_type=str, when_used="always")
+]
+
+
 class ConnectionStrategy(strEnum):
     """An enumerator for the connection Strategies."""
 
     IN_BAND = "IN BAND"
     OUT_OF_BAND = "OUT OF BAND"
+

gso/worker.py

@@ -9,7 +9,7 @@ from gso.settings import load_oss_params
 class OrchestratorCelery(Celery):
     """A :term:`GSO` instance that functions as a Celery worker."""
 
-    def on_init(self) -> None:  # noqa: PLR6301
+    def on_init(self) -> None:
         """Initialise a new Celery worker."""
         init_worker_app()
 

gso/workflows/iptrunk/create_iptrunk.py

@@ -57,7 +57,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         model_config = ConfigDict(title=product_name)
 
         tt_number: str
-        partner: str = ReadOnlyField("GEANT")
+        partner: ReadOnlyField("GEANT", default_type=str)
         geant_s_sid: str | None
         iptrunk_description: str
         iptrunk_type: IptrunkType
@@ -65,7 +65,6 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         iptrunk_number_of_members: int
 
         @field_validator("tt_number")
-        @classmethod
         def validate_tt_number(cls, tt_number: str) -> str:
             return validate_tt_number(tt_number)
 
@@ -115,7 +114,6 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         side_a_ae_members: ae_members_side_a  # type: ignore[valid-type]
 
         @field_validator("side_a_ae_members")
-        @classmethod
         def validate_side_a_ae_members(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
             validate_iptrunk_unique_interface(side_a_ae_members)
             vendor = get_router_vendor(router_a)
@@ -133,7 +131,6 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         side_b_node_id: router_enum_b  # type: ignore[valid-type]
 
         @field_validator("side_b_node_id")
-        @classmethod
         def validate_device_exists_in_netbox(cls, side_b_node_id: UUIDstr) -> str | None:
             return validate_router_in_netbox(side_b_node_id)
 
@@ -161,7 +158,6 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         side_b_ae_members: ae_members_side_b  # type: ignore[valid-type]
 
         @field_validator("side_b_ae_members")
-        @classmethod
         def validate_side_b_ae_members(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
             validate_iptrunk_unique_interface(side_b_ae_members)
             vendor = get_router_vendor(router_b)

gso/workflows/iptrunk/deploy_twamp.py

@@ -29,7 +29,6 @@ def _initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         tt_number: str
 
         @field_validator("tt_number")
-        @classmethod
         def validate_tt_number(cls, tt_number: str) -> str:
             return validate_tt_number(tt_number)
 

gso/workflows/iptrunk/migrate_iptrunk.py

@@ -128,10 +128,10 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         else 1
     )
     existing_lag_ae_members = [
-        {
-            "interface_name": iface.interface_name,
-            "interface_description": iface.interface_description,
-        }
+        LAGMember(
+            interface_name=iface.interface_name,
+            interface_description=iface.interface_description,
+    )
         for iface in subscription.iptrunk.iptrunk_sides[replace_index].iptrunk_side_ae_members
     ]
 
@@ -139,10 +139,10 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         model_config = ConfigDict(title=form_title)
 
         new_lag_interface: side_a_ae_iface  # type: ignore[valid-type]
-        existing_lag_interface: list[LAGMember] = ReadOnlyField(existing_lag_ae_members)
+        existing_lag_interface: ReadOnlyField(existing_lag_ae_members, default_type=list[LAGMember])
         new_lag_member_interfaces: ae_members  # type: ignore[valid-type]
 
-        @field_validator("new_lag_interface", mode="before")
+        @field_validator("new_lag_interface")
         def lag_interface_proper_name(cls, new_lag_interface: str) -> str:
             if get_router_vendor(new_router) == Vendor.JUNIPER:
                 juniper_lag_re = re.compile("^ae\\d{1,2}$")
@@ -151,7 +151,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
                     raise ValueError(msg)
             return new_lag_interface
 
-        @field_validator("new_lag_member_interfaces", mode="before")
+        @field_validator("new_lag_member_interfaces")
         def is_interface_names_valid_juniper(cls, new_lag_member_interfaces: list[LAGMember]) -> list[LAGMember]:
             vendor = get_router_vendor(new_router)
             return validate_interface_name_list(new_lag_member_interfaces, vendor)