diff --git a/.gitignore b/.gitignore
index fccb8082923cbdf2088055e1489862884abd02c3..5dbfe3a1a0edc59e25b82ec47454b370c5e8ed31 100644
--- a/.gitignore
+++ b/.gitignore
@@ -20,3 +20,4 @@ docs/vale/styles/*
.idea
.venv
+.env
\ No newline at end of file
diff --git a/Changelog.md b/Changelog.md
index 94ab4fd8d9c7cbd456b9a990830cb0d40bafc706..4a00b65ca1e21b6fe6e1abef87c8ee9f809cc6f8 100644
--- a/Changelog.md
+++ b/Changelog.md
@@ -1,5 +1,16 @@
# Changelog
+## [2.13] - 2024-09-18
+- Run cleanup at 1 AM only, not every minute between 1 and 2 AM.
+- Add checklist to trunk migration and include ACTIVITY_TYPE column as discussed with SM.
+- Fix SDP update logic in the promotion from P to PE.
+- Update core lib to 2.7.4
+- Refactor validators into a separate types module.
+- Add Tier3 model for Netbox.
+- Fix: Increase initial connection timeout for LibreNMS client.
+- Fix: Change type from LAGMember to dict for workflow step in trunk modification
+- Make celery concurrency level to 1
+
## [2.12] - 2024-08-22
- Add promote P to PE workflow.
- Add new cleanup task
diff --git a/gso/api/v1/network.py b/gso/api/v1/network.py
index b92a135e63ff1a4fcee89e987b3c568fccc28d7f..e4a19de9e90475f70261e4bf408498de68a95af9 100644
--- a/gso/api/v1/network.py
+++ b/gso/api/v1/network.py
@@ -10,11 +10,12 @@ from orchestrator.security import authorize
from orchestrator.services.subscriptions import build_extended_domain_model
from starlette import status
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_blocks.router import RouterRole
-from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate
from gso.services.subscriptions import get_active_iptrunk_subscriptions
from gso.utils.shared_enums import Vendor
+from gso.utils.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
+from gso.utils.types.interfaces import PhysicalPortCapacity
router = APIRouter(prefix="/networks", tags=["Network"], dependencies=[Depends(authorize)])
diff --git a/gso/cli/imports.py b/gso/cli/imports.py
index 406fa85d1dbc0e11d58187909a63dc5f6285285c..33a4636a0b5010d57a7f3c06053e468bb37e191b 100644
--- a/gso/cli/imports.py
+++ b/gso/cli/imports.py
@@ -13,17 +13,17 @@ import yaml
from orchestrator.db import db
from orchestrator.services.processes import start_process
from orchestrator.types import SubscriptionLifecycle
-from pydantic import BaseModel, EmailStr, ValidationError, field_validator, model_validator
+from pydantic import BaseModel, ValidationError, field_validator, model_validator
from sqlalchemy.exc import SQLAlchemyError
from gso.db.models import PartnerTable
from gso.products import ProductType
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_blocks.router import RouterRole
from gso.services.partners import (
+ PartnerEmail,
+ PartnerName,
PartnerNotFoundError,
- filter_partners_by_email,
- filter_partners_by_name,
get_partner_by_name,
)
from gso.services.subscriptions import (
@@ -31,8 +31,10 @@ from gso.services.subscriptions import (
get_active_subscriptions_by_field_and_value,
get_subscriptions,
)
-from gso.utils.helpers import BaseSiteValidatorModel, LAGMember
-from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.base_site import BaseSiteValidatorModel
+from gso.utils.types.interfaces import LAGMember, LAGMemberList, PhysicalPortCapacity
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType, PortNumber
app: typer.Typer = typer.Typer()
@@ -40,27 +42,8 @@ app: typer.Typer = typer.Typer()
class CreatePartner(BaseModel):
"""Required inputs for creating a partner."""
- name: str
- email: EmailStr
-
- @field_validator("name")
- def validate_name(cls, name: str) -> str:
- """Validate name."""
- if filter_partners_by_name(name=name, case_sensitive=False):
- msg = "Partner with this name already exists."
- raise ValueError(msg)
-
- return name
-
- @field_validator("email")
- def validate_email(cls, email: str) -> EmailStr:
- """Validate email."""
- email = email.lower()
- if filter_partners_by_email(email=email, case_sensitive=False):
- msg = "Partner with this email already exists."
- raise ValueError(msg)
-
- return email
+ name: PartnerName
+ email: PartnerEmail
class SiteImportModel(BaseSiteValidatorModel):
@@ -115,11 +98,11 @@ class IptrunkImportModel(BaseModel):
side_a_node_id: str
side_a_ae_iface: str
side_a_ae_geant_a_sid: str | None
- side_a_ae_members: list[LAGMember]
+ side_a_ae_members: LAGMemberList[LAGMember]
side_b_node_id: str
side_b_ae_iface: str
side_b_ae_geant_a_sid: str | None
- side_b_ae_members: list[LAGMember]
+ side_b_ae_members: LAGMemberList[LAGMember]
iptrunk_ipv4_network: ipaddress.IPv4Network
iptrunk_ipv6_network: ipaddress.IPv6Network
@@ -150,15 +133,6 @@ class IptrunkImportModel(BaseModel):
return value
- @field_validator("side_a_ae_members", "side_b_ae_members")
- def check_side_uniqueness(cls, value: list[str]) -> list[str]:
- """:term:`LAG` members must be unique."""
- if len(value) != len(set(value)):
- msg = "Items must be unique"
- raise ValueError(msg)
-
- return value
-
@model_validator(mode="after")
def check_members(self) -> Self:
"""Amount of :term:`LAG` members has to match on side A and B, and meet the minimum requirement."""
diff --git a/gso/products/product_blocks/iptrunk.py b/gso/products/product_blocks/iptrunk.py
index 88ff75cb2afad1960c4bc4f03953903c23040745..ad83609defc7289d49ff01b9501ce87469e16f5b 100644
--- a/gso/products/product_blocks/iptrunk.py
+++ b/gso/products/product_blocks/iptrunk.py
@@ -15,18 +15,7 @@ from gso.products.product_blocks.router import (
RouterBlockInactive,
RouterBlockProvisioning,
)
-
-
-class PhysicalPortCapacity(strEnum):
- """Physical port capacity enumerator.
-
- An enumerator that has the different possible capacities of ports that are available to use in subscriptions.
- """
-
- ONE_GIGABIT_PER_SECOND = "1G"
- TEN_GIGABIT_PER_SECOND = "10G"
- HUNDRED_GIGABIT_PER_SECOND = "100G"
- FOUR_HUNDRED_GIGABIT_PER_SECOND = "400G"
+from gso.utils.types.interfaces import LAGMemberList, PhysicalPortCapacity
class IptrunkType(strEnum):
@@ -36,9 +25,6 @@ class IptrunkType(strEnum):
LEASED = "Leased"
-LAGMemberList = Annotated[
- list[T], AfterValidator(validate_unique_list), Len(min_length=0), Doc("A list of :term:`LAG` member interfaces.")
-]
IptrunkSides = Annotated[
list[T],
AfterValidator(validate_unique_list),
diff --git a/gso/products/product_blocks/lan_switch_interconnect.py b/gso/products/product_blocks/lan_switch_interconnect.py
index a9b1c77366662e24499f936f9762d515b882b934..e2626de5c680bb983454ed32b523e61f1b78ad6d 100644
--- a/gso/products/product_blocks/lan_switch_interconnect.py
+++ b/gso/products/product_blocks/lan_switch_interconnect.py
@@ -3,9 +3,9 @@
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle, strEnum
-from gso.products.product_blocks.iptrunk import LAGMemberList
from gso.products.product_blocks.router import RouterBlock, RouterBlockInactive, RouterBlockProvisioning
from gso.products.product_blocks.switch import SwitchBlock, SwitchBlockInactive, SwitchBlockProvisioning
+from gso.utils.types.interfaces import LAGMemberList
class LanSwitchInterconnectAddressSpace(strEnum):
diff --git a/gso/products/product_blocks/office_router.py b/gso/products/product_blocks/office_router.py
index 65eab0256a073c699f3ea2ef84d96e3352096722..bff50dd3018afd3f2bdee771cd0ca9405ae44b47 100644
--- a/gso/products/product_blocks/office_router.py
+++ b/gso/products/product_blocks/office_router.py
@@ -8,7 +8,8 @@ from gso.products.product_blocks.site import (
SiteBlockInactive,
SiteBlockProvisioning,
)
-from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType, PortNumber
class OfficeRouterBlockInactive(
diff --git a/gso/products/product_blocks/router.py b/gso/products/product_blocks/router.py
index 17deeccb1ac8a5ee9bcfaa14fa25f27360881e7c..6d7dabf9dff3eaded338a420834f9e509d8584a1 100644
--- a/gso/products/product_blocks/router.py
+++ b/gso/products/product_blocks/router.py
@@ -8,7 +8,8 @@ from gso.products.product_blocks.site import (
SiteBlockInactive,
SiteBlockProvisioning,
)
-from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType, PortNumber
class RouterRole(strEnum):
diff --git a/gso/products/product_blocks/site.py b/gso/products/product_blocks/site.py
index be7d086adc021992ede2048b6ef4a843c1793755..eec868479f49f005aa077242a397a52f94e216b1 100644
--- a/gso/products/product_blocks/site.py
+++ b/gso/products/product_blocks/site.py
@@ -1,18 +1,11 @@
"""The product block that describes a site subscription."""
-import re
-from typing import Annotated
-
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle, strEnum
-from pydantic import AfterValidator
-from typing_extensions import Doc
-
-MAX_LONGITUDE = 180
-MIN_LONGITUDE = -180
-MAX_LATITUDE = 90
-MIN_LATITUDE = -90
+from gso.utils.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
+from gso.utils.types.ip_address import IPAddress
+from gso.utils.types.site_name import SiteName
class SiteTier(strEnum):
@@ -28,56 +21,6 @@ class SiteTier(strEnum):
TIER4 = 4
-def validate_latitude(v: str) -> str:
- """Validate a latitude coordinate."""
- msg = "Invalid latitude coordinate. Valid examples: '40.7128', '-74.0060', '90', '-90', '0'."
- regex = re.compile(r"^-?([1-8]?\d(\.\d+)?|90(\.0+)?)$")
- if not regex.match(str(v)):
- raise ValueError(msg)
-
- float_v = float(v)
- if float_v > MAX_LATITUDE or float_v < MIN_LATITUDE:
- raise ValueError(msg)
-
- return v
-
-
-def validate_longitude(v: str) -> str:
- """Validate a longitude coordinate."""
- regex = re.compile(r"^-?(180(\.0+)?|((1[0-7]\d)|([1-9]?\d))(\.\d+)?)$")
- msg = "Invalid longitude coordinate. Valid examples: '40.7128', '-74.0060', '180', '-180', '0'."
- if not regex.match(v):
- raise ValueError(msg)
-
- float_v = float(v)
- if float_v > MAX_LONGITUDE or float_v < MIN_LONGITUDE:
- raise ValueError(msg)
-
- return v
-
-
-LatitudeCoordinate = Annotated[
- str,
- AfterValidator(validate_latitude),
- Doc(
- "A latitude coordinate, modeled as a string. "
- "The coordinate must match the format conforming to the latitude range of -90 to +90 degrees. "
- "It can be a floating-point number or an integer. Valid examples: 40.7128, -74.0060, 90, -90, 0."
- ),
-]
-
-LongitudeCoordinate = Annotated[
- str,
- AfterValidator(validate_longitude),
- Doc(
- "A longitude coordinate, modeled as a string. "
- "The coordinate must match the format conforming to the longitude "
- "range of -180 to +180 degrees. It can be a floating-point number or an integer. "
- "Valid examples: 40.7128, -74.0060, 180, -180, 0."
- ),
-]
-
-
class SiteBlockInactive(
ProductBlockModel,
lifecycle=[SubscriptionLifecycle.INITIAL],
@@ -85,7 +28,7 @@ class SiteBlockInactive(
):
"""A site that's currently inactive, see :class:`SiteBlock`."""
- site_name: str | None = None
+ site_name: SiteName | None = None
site_city: str | None = None
site_country: str | None = None
site_country_code: str | None = None
@@ -94,13 +37,13 @@ class SiteBlockInactive(
site_internal_id: int | None = None
site_bgp_community_id: int | None = None
site_tier: SiteTier | None = None
- site_ts_address: str | None = None
+ site_ts_address: IPAddress | None = None
class SiteBlockProvisioning(SiteBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
"""A site that's currently being provisioned, see :class:`SiteBlock`."""
- site_name: str
+ site_name: SiteName
site_city: str
site_country: str
site_country_code: str
@@ -109,7 +52,7 @@ class SiteBlockProvisioning(SiteBlockInactive, lifecycle=[SubscriptionLifecycle.
site_internal_id: int
site_bgp_community_id: int
site_tier: SiteTier
- site_ts_address: str
+ site_ts_address: IPAddress
class SiteBlock(SiteBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
@@ -117,7 +60,7 @@ class SiteBlock(SiteBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE])
#: The name of the site, that will dictate part of the :term:`FQDN` of routers that are hosted at this site. For
#: example: ``router.X.Y.geant.net``, where X denotes the name of the site.
- site_name: str
+ site_name: SiteName
#: The city at which the site is located.
site_city: str
#: The country in which the site is located.
@@ -138,4 +81,4 @@ class SiteBlock(SiteBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE])
#: The address of the terminal server that this router is connected to. The terminal server provides out of band
#: access. This is required in case a link goes down, or when a router is initially added to the network and it
#: does not have any IP trunks connected to it.
- site_ts_address: str
+ site_ts_address: IPAddress
diff --git a/gso/products/product_blocks/super_pop_switch.py b/gso/products/product_blocks/super_pop_switch.py
index 3335b28cf90ee9d55abe59be528f404d44d905b8..872f37b59042b2534de234d26241cd2792c55c3e 100644
--- a/gso/products/product_blocks/super_pop_switch.py
+++ b/gso/products/product_blocks/super_pop_switch.py
@@ -8,7 +8,8 @@ from gso.products.product_blocks.site import (
SiteBlockInactive,
SiteBlockProvisioning,
)
-from gso.utils.shared_enums import IPv4AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.ip_address import IPv4AddressType, PortNumber
class SuperPopSwitchBlockInactive(
diff --git a/gso/products/product_blocks/switch.py b/gso/products/product_blocks/switch.py
index f0aa0414e11409b58341fc2648f3066fa91c5aee..bdf2b4657f8d8674a5c4e19c192f455ef1e31524 100644
--- a/gso/products/product_blocks/switch.py
+++ b/gso/products/product_blocks/switch.py
@@ -9,7 +9,8 @@ from gso.products.product_blocks.site import (
SiteBlockInactive,
SiteBlockProvisioning,
)
-from gso.utils.shared_enums import PortNumber, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.ip_address import PortNumber
class SwitchModel(strEnum):
diff --git a/gso/schedules/task_vacuum.py b/gso/schedules/task_vacuum.py
index de4d44842f485cc8df41f9c01eed3ead2d651a30..3ad873fde290e4b4cbfc30a3357e5d0c05f407c1 100644
--- a/gso/schedules/task_vacuum.py
+++ b/gso/schedules/task_vacuum.py
@@ -7,7 +7,7 @@ from gso.worker import celery
@celery.task
-@scheduler(CronScheduleConfig(name="Clean up tasks", hour="1"))
+@scheduler(CronScheduleConfig(name="Clean up tasks", hour="1", minute="0"))
def vacuum_tasks() -> None:
"""Run all cleanup tasks every 1 AM UTC."""
start_process("task_clean_up_tasks")
diff --git a/gso/services/infoblox.py b/gso/services/infoblox.py
index 22e56ab5a7215ed8026d1a3a80e7d72ee1ba916e..8e06be865a43b36474649a1cda3fc697a330e1f7 100644
--- a/gso/services/infoblox.py
+++ b/gso/services/infoblox.py
@@ -10,7 +10,7 @@ from infoblox_client.exceptions import (
)
from gso.settings import IPAMParams, load_oss_params
-from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType
logger = getLogger(__name__)
NULL_MAC = "00:00:00:00:00:00"
diff --git a/gso/services/librenms_client.py b/gso/services/librenms_client.py
index 2e04a866e3f85b539f22031b3f878b2149823d0f..b7c21f888ebe281ada67e64edb3e9fd671a0a655 100644
--- a/gso/services/librenms_client.py
+++ b/gso/services/librenms_client.py
@@ -10,7 +10,7 @@ from requests import HTTPError, Response
from requests.adapters import HTTPAdapter
from gso.settings import load_oss_params
-from gso.utils.helpers import SNMPVersion
+from gso.utils.types.snmp import SNMPVersion
logger = logging.getLogger(__name__)
@@ -40,7 +40,7 @@ class LibreNMSClient:
) -> Response:
url = self.base_url + endpoint
logger.debug("LibreNMS - Sending request", extra={"method": method, "endpoint": url, "form_data": data})
- result = self.session.request(method, url, json=data, timeout=(0.5, 75))
+ result = self.session.request(method, url, json=data, timeout=(10, 75))
logger.debug("LibreNMS - Received response", extra=result.__dict__)
return result
diff --git a/gso/services/partners.py b/gso/services/partners.py
index 63d55bb3f3d4044f60ce90f4f6cbd839e1efa069..c8cfeb084b4daa4136d1326ffd90f68de2399f59 100644
--- a/gso/services/partners.py
+++ b/gso/services/partners.py
@@ -1,29 +1,60 @@
"""A module that returns the partners available in :term:`GSO`."""
from datetime import datetime
-from typing import Any
+from typing import Annotated, Any
from uuid import uuid4
from orchestrator.db import db
-from pydantic import BaseModel, ConfigDict, EmailStr, Field
+from pydantic import AfterValidator, BaseModel, ConfigDict, EmailStr, Field
from sqlalchemy import func
from sqlalchemy.exc import NoResultFound
from gso.db.models import PartnerTable
+def validate_partner_name_unique(name: str) -> str:
+ """Validate that the name of a partner is unique."""
+ if filter_partners_by_name(name=name, case_sensitive=False):
+ msg = "Partner with this name already exists."
+ raise ValueError(msg)
+ return name
+
+
+def validate_partner_email_unique(email: EmailStr) -> EmailStr:
+ """Validate that the e-mail address of a partner is unique."""
+ email = email.lower()
+ if filter_partners_by_email(email=email, case_sensitive=False):
+ msg = "Partner with this email already exists."
+ raise ValueError(msg)
+ return email
+
+
+PartnerName = Annotated[str, AfterValidator(validate_partner_name_unique)]
+PartnerEmail = Annotated[EmailStr, AfterValidator(validate_partner_email_unique)]
+
+
class PartnerSchema(BaseModel):
"""Partner schema."""
partner_id: str = Field(default_factory=lambda: str(uuid4()))
- name: str
- email: EmailStr
+ name: PartnerName
+ email: PartnerEmail
created_at: datetime = Field(default_factory=lambda: datetime.now().astimezone())
updated_at: datetime = Field(default_factory=lambda: datetime.now().astimezone())
model_config = ConfigDict(from_attributes=True)
+class ModifiedPartnerSchema(PartnerSchema):
+ """Partner schema when making a modification.
+
+ The name and email can be empty in this case, if they don't need changing.
+ """
+
+ name: PartnerName | None = None # type: ignore[assignment]
+ email: PartnerEmail | None = None # type: ignore[assignment]
+
+
class PartnerNotFoundError(Exception):
"""Exception raised when a partner is not found."""
@@ -95,7 +126,7 @@ def create_partner(
def edit_partner(
- partner_data: PartnerSchema,
+ partner_data: ModifiedPartnerSchema,
) -> PartnerTable:
"""Edit an existing partner and update it in the database."""
partner = get_partner_by_id(partner_id=partner_data.partner_id)
diff --git a/gso/services/subscriptions.py b/gso/services/subscriptions.py
index 25ed84e0593b2963656167737f17b4697ef2b0df..8a8bcf98eeaf7e490e294298c3e6b4af9a070c68 100644
--- a/gso/services/subscriptions.py
+++ b/gso/services/subscriptions.py
@@ -69,20 +69,6 @@ def get_subscriptions(
return [dict(zip(includes, result, strict=True)) for result in results]
-def get_active_site_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
- """Retrieve active subscriptions specifically for sites.
-
- :param includes: The fields to be included in the returned Subscription objects.
- :type includes: list[str]
-
- :return: A list of Subscription objects for sites.
- :rtype: list[Subscription]
- """
- return get_subscriptions(
- product_types=[ProductType.SITE], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
- )
-
-
def get_router_subscriptions(
includes: list[str] | None = None, lifecycles: list[SubscriptionLifecycle] | None = None
) -> list[SubscriptionType]:
@@ -122,6 +108,20 @@ def get_provisioning_router_subscriptions(includes: list[str] | None = None) ->
)
+def get_active_switch_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
+ """Retrieve active subscriptions specifically for switches.
+
+ :param includes: The fields to be included in the returned Subscription objects.
+ :type includes: list[str]
+
+ :return: A list of Subscription objects for switches.
+ :rtype: list[Subscription]
+ """
+ return get_subscriptions(
+ product_types=[ProductType.SWITCH], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
+ )
+
+
def get_active_iptrunk_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
"""Retrieve active subscriptions specifically for IP trunks.
@@ -233,6 +233,20 @@ def get_active_insync_subscriptions() -> list[SubscriptionTable]:
)
+def get_active_site_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
+ """Retrieve active subscriptions specifically for sites.
+
+ :param includes: The fields to be included in the returned Subscription objects.
+ :type includes: list[str]
+
+ :return: A list of Subscription objects for sites.
+ :rtype: list[Subscription]
+ """
+ return get_subscriptions(
+ product_types=[ProductType.SITE], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
+ )
+
+
def get_site_by_name(site_name: str) -> Site:
"""Get a site by its name.
diff --git a/gso/settings.py b/gso/settings.py
index d2a845f3e96b526675d8d0de57972a15218a43ce..f7d40de73c73cdb75ee7c6348210fb12c543100e 100644
--- a/gso/settings.py
+++ b/gso/settings.py
@@ -16,7 +16,7 @@ from pydantic_forms.types import UUIDstr, strEnum
from pydantic_settings import BaseSettings
from typing_extensions import Doc
-from gso.utils.shared_enums import PortNumber
+from gso.utils.types.ip_address import PortNumber
logger = logging.getLogger(__name__)
diff --git a/gso/translations/en-GB.json b/gso/translations/en-GB.json
index 108a28522a74747461352dbd5133ba5ccf63f3ec..f1ede053f813aefef202d92ffd31b6d48c3908a5 100644
--- a/gso/translations/en-GB.json
+++ b/gso/translations/en-GB.json
@@ -32,16 +32,18 @@
"migrate_to_different_site": "Migrating to a different Site",
"remove_configuration": "Remove configuration from the router",
"clean_up_ipam": "Clean up related entries in IPAM",
- "restore_isis_metric": "Restore ISIS metric to original value"
+ "restore_isis_metric": "Restore ISIS metric to original value",
+ "confirm_info": "Please verify this form looks correct."
}
},
"workflow": {
"activate_iptrunk": "Activate IP Trunk",
- "activate_router": "Activate router",
- "confirm_info": "Please verify this form looks correct.",
+ "activate_router": "Activate Router",
+ "activate_switch": "Activate Switch",
"create_iptrunk": "Create IP Trunk",
"create_router": "Create Router",
"create_site": "Create Site",
+ "create_switch": "Create Switch",
"deploy_twamp": "Deploy TWAMP",
"migrate_iptrunk": "Migrate IP Trunk",
"modify_isis_metric": "Modify the ISIS metric",
@@ -52,6 +54,7 @@
"terminate_iptrunk": "Terminate IP Trunk",
"terminate_router": "Terminate Router",
"terminate_site": "Terminate Site",
+ "terminate_switch": "Terminate Switch",
"redeploy_base_config": "Redeploy base config",
"update_ibgp_mesh": "Update iBGP mesh",
"create_imported_site": "NOT FOR HUMANS -- Import existing site",
@@ -67,7 +70,8 @@
"import_super_pop_switch": "NOT FOR HUMANS -- Finalize import into a Super PoP switch",
"import_opengear": "NOT FOR HUMANS -- Finalize import into an OpenGear",
"validate_iptrunk": "Validate IP Trunk configuration",
- "validate_router": "Validate router configuration",
+ "validate_router": "Validate Router configuration",
+ "validate_switch": "Validate Switch configuration",
"task_validate_geant_products": "Validation task for GEANT products",
"task_send_email_notifications": "Send email notifications for failed tasks",
"task_create_partners": "Create partner task",
diff --git a/gso/utils/device_info.py b/gso/utils/device_info.py
index cfc81d06562775bf7709a6d7fbc68e4f884618c4..800d56779ae607a16a50a430a88a024f9cbdeb66 100644
--- a/gso/utils/device_info.py
+++ b/gso/utils/device_info.py
@@ -32,6 +32,13 @@ class TierInfo:
breakout_interfaces_per_slot=[36, 35, 34, 33],
total_10g_interfaces=60,
)
+ self.tier3 = ModuleInfo(
+ device_type="7750 SR2-se",
+ module_bays_slots=[1, 2],
+ module_type="XCMC-2SE-2",
+ breakout_interfaces_per_slot=[36, 35, 34, 33],
+ total_10g_interfaces=76,
+ )
def get_module_by_name(self, name: str) -> ModuleInfo:
"""Retrieve a module by name."""
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index 13016c76f13028db17e8397ebff7ce23455a997a..4a78b989d41ddd49e2ee99b5f703a9dd519a7a07 100644
--- a/gso/utils/helpers.py
+++ b/gso/utils/helpers.py
@@ -1,42 +1,22 @@
"""Helper methods that are used across :term:`GSO`."""
-import ipaddress
import re
-from enum import StrEnum
+from typing import TYPE_CHECKING
from uuid import UUID
-import pycountry
-from orchestrator.types import UUIDstr
-from pydantic import BaseModel, field_validator
-from pydantic_core.core_schema import ValidationInfo
from pydantic_forms.validators import Choice
from gso import settings
-from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock, PhysicalPortCapacity
from gso.products.product_blocks.router import RouterRole
-from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate, SiteTier
from gso.products.product_types.router import Router
+from gso.services import subscriptions
from gso.services.netbox_client import NetboxClient
-from gso.services.subscriptions import get_active_router_subscriptions, get_active_subscriptions_by_field_and_value
-from gso.utils.shared_enums import IPv4AddressType, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.interfaces import PhysicalPortCapacity
+from gso.utils.types.ip_address import IPv4AddressType
-
-class LAGMember(BaseModel):
- """A :term:`LAG` member interface that consists of a name and description."""
-
- interface_name: str
- interface_description: str | None = None
-
- def __hash__(self) -> int:
- """Calculate the hash based on the interface name and description, so that uniqueness can be determined."""
- return hash((self.interface_name, self.interface_description))
-
-
-class SNMPVersion(StrEnum):
- """An enumerator for the two relevant versions of :term:`SNMP`: v2c and 3."""
-
- V2C = "v2c"
- V3 = "v3"
+if TYPE_CHECKING:
+ from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
@@ -57,7 +37,7 @@ def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
def available_interfaces_choices_including_current_members(
router_id: UUID,
speed: str,
- interfaces: list[IptrunkInterfaceBlock],
+ interfaces: list["IptrunkInterfaceBlock"],
) -> Choice | None:
"""Return a list of available interfaces for a given router and speed including the current members.
@@ -119,184 +99,6 @@ def iso_from_ipv4(ipv4_address: IPv4AddressType) -> str:
return f"49.51e5.0001.{re_split}.00"
-def validate_router_in_netbox(subscription_id: UUIDstr) -> UUIDstr:
- """Verify if a device exists in Netbox.
-
- Raises a :class:`ValueError` if the device is not found.
-
- :param subscription_id: The :term:`UUID` of the router subscription.
- :type subscription_id: :class:`UUIDstr`
-
- :return: The :term:`UUID` of the router subscription.
- :rtype: :class:`UUIDstr`
- """
- router_type = Router.from_subscription(subscription_id)
- if router_type.router.vendor == Vendor.NOKIA:
- device = NetboxClient().get_device_by_name(router_type.router.router_fqdn)
- if not device:
- msg = "The selected router does not exist in Netbox."
- raise ValueError(msg)
- return subscription_id
-
-
-def validate_iptrunk_unique_interface(interfaces: list[LAGMember]) -> list[LAGMember]:
- """Verify if the interfaces are unique.
-
- Raises a :class:`ValueError` if the interfaces are not unique.
-
- :param interfaces: The list of interfaces.
- :type interfaces: list[:class:`LAGMember`]
-
- :return: The list of interfaces
- :rtype: list[:class:`LAGMember`]
- """
- interface_names = [member.interface_name for member in interfaces]
- if len(interface_names) != len(set(interface_names)):
- msg = "Interfaces must be unique."
- raise ValueError(msg)
- return interfaces
-
-
-def validate_site_fields_is_unique(field_name: str, value: str | int) -> None:
- """Validate that a site field is unique."""
- if len(get_active_subscriptions_by_field_and_value(field_name, str(value))) > 0:
- msg = f"{field_name} must be unique"
- raise ValueError(msg)
-
-
-def validate_ipv4_or_ipv6(value: str) -> str:
- """Validate that a value is a valid IPv4 or IPv6 address."""
- try:
- ipaddress.ip_address(value)
- except ValueError as e:
- msg = "Enter a valid IPv4 or IPv6 address."
- raise ValueError(msg) from e
- else:
- return value
-
-
-def validate_country_code(country_code: str) -> str:
- """Validate that a country code is valid."""
- # Check for the UK code before attempting to look it up since it's known as "GB" in the pycountry database.
- if country_code != "UK":
- try:
- pycountry.countries.lookup(country_code)
- except LookupError as e:
- msg = "Invalid or non-existent country code, it must be in ISO 3166-1 alpha-2 format."
- raise ValueError(msg) from e
- return country_code
-
-
-def validate_site_name(site_name: str) -> None:
- """Validate the site name.
-
- The site name must consist of three uppercase letters, optionally followed by a single digit.
- """
- pattern = re.compile(r"^[A-Z]{3}[0-9]?$")
- if not pattern.match(site_name):
- msg = (
- "Enter a valid site name. It must consist of three uppercase letters (A-Z), followed by an optional single "
- f"digit (0-9). Received: {site_name}"
- )
- raise ValueError(msg)
-
-
-class BaseSiteValidatorModel(BaseModel):
- """A base site validator model extended by create site and by import site."""
-
- site_bgp_community_id: int
- 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
- partner: str
-
- @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
-
- @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
-
- @field_validator("site_ts_address", "site_internal_id", "site_bgp_community_id", "site_name")
- def field_must_be_unique(cls, value: str | int, info: ValidationInfo) -> str | int:
- """Validate that a field is unique."""
- if not info.field_name:
- msg = "Field name must be provided."
- raise ValueError(msg)
-
- validate_site_fields_is_unique(info.field_name, value)
-
- return value
-
- @field_validator("site_name")
- def site_name_must_be_valid(cls, site_name: str) -> str:
- """Validate the site name.
-
- The site name must consist of three uppercase letters, followed by an optional single digit.
- """
- validate_site_name(site_name)
- return site_name
-
-
-def validate_interface_name_list(interface_name_list: list, vendor: str) -> list:
- """Validate that the provided interface name matches the expected pattern.
-
- The expected pattern for the interface name is one of 'ge', 'et', 'xe' followed by a dash '-',
- then a number between 0 and 19, a forward slash '/', another number between 0 and 99,
- another forward slash '/', and ends with a number between 0 and 99.
- For example: 'xe-1/0/0'.
-
- :param list interface_name_list: List of interface names to validate.
- :param str vendor: Router vendor to check interface names
-
- :return list: The list of interface names if all match was successful, otherwise it will throw a ValueError
- exception.
- """
- # For Nokia nothing to do
- if vendor == Vendor.NOKIA:
- return interface_name_list
- pattern = re.compile(r"^(ge|et|xe)-1?[0-9]/[0-9]{1,2}/[0-9]{1,2}$")
- for interface in interface_name_list:
- if not bool(pattern.match(interface.interface_name)):
- error_msg = (
- f"Invalid interface name. The interface name should be of format: xe-1/0/0. "
- f"Got: [{interface.interface_name}]"
- )
- raise ValueError(error_msg)
- return interface_name_list
-
-
-def validate_tt_number(tt_number: str) -> str:
- """Validate a string to match a specific pattern.
-
- This method checks if the input string starts with 'TT#' and is followed by exactly 16 digits.
-
- :param str tt_number: The TT number as string to validate
-
- :return str: The TT number string if TT number match was successful, otherwise it will raise a ValueError.
- """
- pattern = r"^TT#\d{16}$"
- if not bool(re.match(pattern, tt_number)):
- err_msg = (
- f"The given TT number: {tt_number} is not valid. "
- f" A valid TT number starts with 'TT#' followed by 16 digits."
- )
- raise ValueError(err_msg)
-
- return tt_number
-
-
def generate_fqdn(hostname: str, site_name: str, country_code: str) -> str:
"""Generate an :term:`FQDN` from a hostname, site name, and a country code."""
oss = settings.load_oss_params()
@@ -318,7 +120,9 @@ def generate_inventory_for_active_routers(
:return: A dictionary representing the inventory of active routers.
:rtype: dict[str, Any]
"""
- all_routers = [Router.from_subscription(r["subscription_id"]) for r in get_active_router_subscriptions()]
+ all_routers = [
+ Router.from_subscription(r["subscription_id"]) for r in subscriptions.get_active_router_subscriptions()
+ ]
exclude_routers = exclude_routers or []
return {
@@ -351,3 +155,33 @@ def calculate_recommended_minimum_links(iptrunk_number_of_members: int, iptrunk_
if iptrunk_speed == PhysicalPortCapacity.FOUR_HUNDRED_GIGABIT_PER_SECOND:
return iptrunk_number_of_members - 1
return iptrunk_number_of_members
+
+
+def active_site_selector() -> Choice:
+ """Generate a dropdown selector for choosing an active site in an input form."""
+ site_subscriptions = {
+ str(site["subscription_id"]): site["description"]
+ for site in subscriptions.get_active_site_subscriptions(includes=["subscription_id", "description"])
+ }
+
+ return Choice("Select a site", zip(site_subscriptions.keys(), site_subscriptions.items(), strict=True)) # type: ignore[arg-type]
+
+
+def active_router_selector() -> Choice:
+ """Generate a dropdown selector for choosing an active Router in an input form."""
+ router_subscriptions = {
+ str(router["subscription_id"]): router["description"]
+ for router in subscriptions.get_active_router_subscriptions(includes=["subscription_id", "description"])
+ }
+
+ return Choice("Select a router", zip(router_subscriptions.keys(), router_subscriptions.items(), strict=True)) # type: ignore[arg-type]
+
+
+def active_switch_selector() -> Choice:
+ """Generate a dropdown selector for choosing an active Switch in an input form."""
+ switch_subscriptions = {
+ str(switch["subscription_id"]): switch["description"]
+ for switch in subscriptions.get_active_switch_subscriptions(includes=["subscription_id", "description"])
+ }
+
+ return Choice("Select a switch", zip(switch_subscriptions.keys(), switch_subscriptions.items(), strict=True)) # type: ignore[arg-type]
diff --git a/gso/utils/shared_enums.py b/gso/utils/shared_enums.py
index 94538f78463821bf266bdd15d379ec61b0371d4b..929e5b0bc2c2553c90b36ce8f056c1f22fed74cc 100644
--- a/gso/utils/shared_enums.py
+++ b/gso/utils/shared_enums.py
@@ -1,16 +1,6 @@
"""Shared choices for the different models."""
-import ipaddress
-from typing import Annotated, Any
-
-from pydantic import Field, PlainSerializer
from pydantic_forms.types import strEnum
-from typing_extensions import Doc
-
-
-def convert_to_str(value: Any) -> str:
- """Convert the value to a string."""
- return str(value)
class Vendor(strEnum):
@@ -20,24 +10,6 @@ class Vendor(strEnum):
NOKIA = "nokia"
-PortNumber = Annotated[
- int,
- Field(
- gt=0,
- le=49151,
- ),
- Doc(
- "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."
- ),
-]
-
-
-IPv4AddressType = Annotated[ipaddress.IPv4Address, PlainSerializer(convert_to_str, return_type=str, when_used="always")]
-
-IPv6AddressType = Annotated[ipaddress.IPv6Address, PlainSerializer(convert_to_str, return_type=str, when_used="always")]
-
-
class ConnectionStrategy(strEnum):
"""An enumerator for the connection Strategies."""
diff --git a/gso/utils/types.py b/gso/utils/types.py
deleted file mode 100644
index 3e1b4091b127d9a572c12c7fad462dc4887de9f7..0000000000000000000000000000000000000000
--- a/gso/utils/types.py
+++ /dev/null
@@ -1,9 +0,0 @@
-"""Define custom types for use across the application."""
-
-from typing import Annotated
-
-from pydantic import AfterValidator
-
-from gso.utils.helpers import validate_tt_number
-
-TTNumber = Annotated[str, AfterValidator(validate_tt_number)]
diff --git a/gso/utils/types/__init__.py b/gso/utils/types/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..c3d1994f0c0c918fc35c2aabd757c67488065aa9
--- /dev/null
+++ b/gso/utils/types/__init__.py
@@ -0,0 +1 @@
+"""Define custom types for use across the application."""
diff --git a/gso/utils/types/base_site.py b/gso/utils/types/base_site.py
new file mode 100644
index 0000000000000000000000000000000000000000..c5af009ad460e56619752688f52dc2bdfa5f9e46
--- /dev/null
+++ b/gso/utils/types/base_site.py
@@ -0,0 +1,26 @@
+"""A base site type for validation purposes that can be extended elsewhere."""
+
+from pydantic import BaseModel
+
+from gso.products.product_blocks.site import SiteTier
+from gso.utils.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
+from gso.utils.types.country_code import CountryCode
+from gso.utils.types.ip_address import IPAddress
+from gso.utils.types.site_name import SiteName
+from gso.utils.types.unique_field import UniqueField
+
+
+class BaseSiteValidatorModel(BaseModel):
+ """A base site validator model extended by create site and by import site."""
+
+ site_bgp_community_id: UniqueField[int]
+ site_internal_id: UniqueField[int]
+ site_tier: SiteTier
+ site_ts_address: UniqueField[IPAddress]
+ site_country_code: CountryCode
+ site_name: UniqueField[SiteName]
+ site_city: str
+ site_country: str
+ site_latitude: LatitudeCoordinate
+ site_longitude: LongitudeCoordinate
+ partner: str
diff --git a/gso/utils/types/coordinates.py b/gso/utils/types/coordinates.py
new file mode 100644
index 0000000000000000000000000000000000000000..91f1188fbe9196a564af4ab9f34a8d2aedeafb16
--- /dev/null
+++ b/gso/utils/types/coordinates.py
@@ -0,0 +1,61 @@
+"""Custom coordinate types for latitude and longitude."""
+
+import re
+from typing import Annotated
+
+from pydantic import AfterValidator
+from typing_extensions import Doc
+
+MAX_LONGITUDE = 180
+MIN_LONGITUDE = -180
+MAX_LATITUDE = 90
+MIN_LATITUDE = -90
+
+
+def validate_latitude(v: str) -> str:
+ """Validate a latitude coordinate."""
+ msg = "Invalid latitude coordinate. Valid examples: '40.7128', '-74.0060', '90', '-90', '0'."
+ regex = re.compile(r"^-?([1-8]?\d(\.\d+)?|90(\.0+)?)$")
+ if not regex.match(str(v)):
+ raise ValueError(msg)
+
+ float_v = float(v)
+ if float_v > MAX_LATITUDE or float_v < MIN_LATITUDE:
+ raise ValueError(msg)
+
+ return v
+
+
+def validate_longitude(v: str) -> str:
+ """Validate a longitude coordinate."""
+ regex = re.compile(r"^-?(180(\.0+)?|((1[0-7]\d)|([1-9]?\d))(\.\d+)?)$")
+ msg = "Invalid longitude coordinate. Valid examples: '40.7128', '-74.0060', '180', '-180', '0'."
+ if not regex.match(v):
+ raise ValueError(msg)
+
+ float_v = float(v)
+ if float_v > MAX_LONGITUDE or float_v < MIN_LONGITUDE:
+ raise ValueError(msg)
+
+ return v
+
+
+LatitudeCoordinate = Annotated[
+ str,
+ AfterValidator(validate_latitude),
+ Doc(
+ "A latitude coordinate, modeled as a string. "
+ "The coordinate must match the format conforming to the latitude range of -90 to +90 degrees. "
+ "It can be a floating-point number or an integer. Valid examples: 40.7128, -74.0060, 90, -90, 0."
+ ),
+]
+LongitudeCoordinate = Annotated[
+ str,
+ AfterValidator(validate_longitude),
+ Doc(
+ "A longitude coordinate, modeled as a string. "
+ "The coordinate must match the format conforming to the longitude "
+ "range of -180 to +180 degrees. It can be a floating-point number or an integer. "
+ "Valid examples: 40.7128, -74.0060, 180, -180, 0."
+ ),
+]
diff --git a/gso/utils/types/country_code.py b/gso/utils/types/country_code.py
new file mode 100644
index 0000000000000000000000000000000000000000..065e6417a19320e75ef89b1002ce3e1f632f59f8
--- /dev/null
+++ b/gso/utils/types/country_code.py
@@ -0,0 +1,21 @@
+"""Country codes."""
+
+from typing import Annotated
+
+import pycountry
+from pydantic import AfterValidator
+
+
+def validate_country_code(country_code: str) -> str:
+ """Validate that a country code is valid."""
+ # Check for the UK code before attempting to look it up since it's known as "GB" in the pycountry database.
+ if country_code != "UK":
+ try:
+ pycountry.countries.lookup(country_code)
+ except LookupError as e:
+ msg = "Invalid or non-existent country code, it must be in ISO 3166-1 alpha-2 format."
+ raise ValueError(msg) from e
+ return country_code
+
+
+CountryCode = Annotated[str, AfterValidator(validate_country_code)]
diff --git a/gso/utils/types/interfaces.py b/gso/utils/types/interfaces.py
new file mode 100644
index 0000000000000000000000000000000000000000..15c91167de822fee116600b71bf0b05dedfd87ae
--- /dev/null
+++ b/gso/utils/types/interfaces.py
@@ -0,0 +1,102 @@
+"""Custom types for interfaces, both physical and logical."""
+
+import re
+from typing import Annotated
+
+from annotated_types import Len
+from orchestrator.domain.base import T
+from pydantic import AfterValidator, BaseModel
+from pydantic_forms.types import strEnum
+from pydantic_forms.validators import validate_unique_list
+from typing_extensions import Doc
+
+
+class LAGMember(BaseModel):
+ """A :term:`LAG` member interface that consists of a name and description."""
+
+ interface_name: str
+ interface_description: str | None = None
+
+ def __hash__(self) -> int:
+ """Calculate the hash based on the interface name and description, so that uniqueness can be determined."""
+ return hash(self.interface_name)
+
+
+def validate_interface_names_are_unique(interfaces: list[LAGMember]) -> list[LAGMember]:
+ """Verify if interfaces are unique.
+
+ Raises a :class:`ValueError` if the interfaces are not unique.
+
+ :param interfaces: The list of interfaces.
+ :type interfaces: list[:class:`utils.types.LAGMember`]
+
+ :return: The list of interfaces
+ :rtype: list[:class:`utils.types.LAGMember`]
+ """
+ interface_names = [member.interface_name for member in interfaces]
+ if len(interface_names) != len(set(interface_names)):
+ msg = "Interfaces must be unique."
+ raise ValueError(msg)
+ return interfaces
+
+
+def validate_juniper_phy_interface_name(interface_name: str) -> str:
+ """Validate that the provided interface name matches the expected pattern.
+
+ The expected pattern for the interface name is one of 'ge', 'et', 'xe' followed by a dash '-',
+ then a number between 0 and 19, a forward slash '/', another number between 0 and 99,
+ another forward slash '/', and ends with a number between 0 and 99.
+ For example: 'xe-1/0/0'. This only applies to Juniper-brand hardware.
+
+ :param str interface_name: Interface name to validate.
+
+ :return str: The interface name if match was successful, otherwise it will throw a ValueError exception.
+ """
+ pattern = re.compile(r"^(ge|et|xe)-1?[0-9]/[0-9]{1,2}/[0-9]{1,2}$")
+ if not bool(pattern.match(interface_name)):
+ error_msg = (
+ f"Invalid interface name. The interface name should be of format: xe-1/0/0. " f"Got: {interface_name}"
+ )
+ raise ValueError(error_msg)
+ return interface_name
+
+
+def validate_juniper_ae_interface_name(interface_name: str) -> str:
+ """Validate that the provided interface name matches the expected pattern for a :term:`LAG` interface.
+
+ Interface names must match the pattern 'ae' followed by a one- or two-digit number.
+ """
+ juniper_lag_re = re.compile("^ae\\d{1,2}$")
+ if not juniper_lag_re.match(interface_name):
+ msg = "Invalid LAG name, please try again."
+ raise ValueError(msg)
+ return interface_name
+
+
+JuniperPhyInterface = Annotated[str, AfterValidator(validate_juniper_phy_interface_name)]
+JuniperAEInterface = Annotated[str, AfterValidator(validate_juniper_ae_interface_name)]
+LAGMemberList = Annotated[
+ list[T],
+ AfterValidator(validate_unique_list),
+ AfterValidator(validate_interface_names_are_unique),
+ Len(min_length=0),
+ Doc("A list of :term:`LAG` member interfaces."),
+]
+
+
+class JuniperLAGMember(LAGMember):
+ """A Juniper-specific :term:`LAG` member interface."""
+
+ interface_name: JuniperPhyInterface
+
+
+class PhysicalPortCapacity(strEnum):
+ """Physical port capacity enumerator.
+
+ An enumerator that has the different possible capacities of ports that are available to use in subscriptions.
+ """
+
+ ONE_GIGABIT_PER_SECOND = "1G"
+ TEN_GIGABIT_PER_SECOND = "10G"
+ HUNDRED_GIGABIT_PER_SECOND = "100G"
+ FOUR_HUNDRED_GIGABIT_PER_SECOND = "400G"
diff --git a/gso/utils/types/ip_address.py b/gso/utils/types/ip_address.py
new file mode 100644
index 0000000000000000000000000000000000000000..94cb8beaf498900e4785d3e222a73685f5d35bf6
--- /dev/null
+++ b/gso/utils/types/ip_address.py
@@ -0,0 +1,38 @@
+"""IP addresses."""
+
+import ipaddress
+from typing import Annotated, Any
+
+from pydantic import AfterValidator, Field, PlainSerializer
+from typing_extensions import Doc
+
+
+def validate_ipv4_or_ipv6(value: str) -> str:
+ """Validate that a value is a valid IPv4 or IPv6 address."""
+ try:
+ ipaddress.ip_address(value)
+ except ValueError as e:
+ msg = "Enter a valid IPv4 or IPv6 address."
+ raise ValueError(msg) from e
+ else:
+ return value
+
+
+def _str(value: Any) -> str:
+ return str(value)
+
+
+IPv4AddressType = Annotated[ipaddress.IPv4Address, PlainSerializer(_str, return_type=str, when_used="always")]
+IPv6AddressType = Annotated[ipaddress.IPv6Address, PlainSerializer(_str, return_type=str, when_used="always")]
+IPAddress = Annotated[str, AfterValidator(validate_ipv4_or_ipv6)]
+PortNumber = Annotated[
+ int,
+ Field(
+ gt=0,
+ le=49151,
+ ),
+ Doc(
+ "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."
+ ),
+]
diff --git a/gso/utils/types/netbox_router.py b/gso/utils/types/netbox_router.py
new file mode 100644
index 0000000000000000000000000000000000000000..2a302b5604363deb9062a9c6649ae2140a27a856
--- /dev/null
+++ b/gso/utils/types/netbox_router.py
@@ -0,0 +1,34 @@
+"""A router that must be present in Netbox."""
+
+from typing import Annotated, TypeVar
+
+from pydantic import AfterValidator
+from pydantic_forms.types import UUIDstr
+
+from gso.products.product_types.router import Router
+from gso.services.netbox_client import NetboxClient
+from gso.utils.shared_enums import Vendor
+
+
+def validate_router_in_netbox(subscription_id: UUIDstr) -> UUIDstr:
+ """Verify if a device exists in Netbox.
+
+ Raises a :class:`ValueError` if the device is not found.
+
+ :param subscription_id: The :term:`UUID` of the router subscription.
+ :type subscription_id: :class:`UUIDstr`
+
+ :return: The :term:`UUID` of the router subscription.
+ :rtype: :class:`UUIDstr`
+ """
+ router_type = Router.from_subscription(subscription_id)
+ if router_type.router.vendor == Vendor.NOKIA:
+ device = NetboxClient().get_device_by_name(router_type.router.router_fqdn)
+ if not device:
+ msg = "The selected router does not exist in Netbox."
+ raise ValueError(msg)
+ return subscription_id
+
+
+T = TypeVar("T")
+NetboxEnabledRouter = Annotated[T, UUIDstr, AfterValidator(validate_router_in_netbox)]
diff --git a/gso/utils/types/site_name.py b/gso/utils/types/site_name.py
new file mode 100644
index 0000000000000000000000000000000000000000..903b3468abbe744280a77c27bb2cf7861e8a389e
--- /dev/null
+++ b/gso/utils/types/site_name.py
@@ -0,0 +1,24 @@
+"""Type for the name of a site."""
+
+import re
+from typing import Annotated
+
+from pydantic import AfterValidator
+
+
+def validate_site_name(site_name: str) -> str:
+ """Validate the site name.
+
+ The site name must consist of three uppercase letters, optionally followed by a single digit.
+ """
+ pattern = re.compile(r"^[A-Z]{3}[0-9]?$")
+ if not pattern.match(site_name):
+ msg = (
+ "Enter a valid site name. It must consist of three uppercase letters (A-Z), followed by an optional single "
+ f"digit (0-9). Received: {site_name}"
+ )
+ raise ValueError(msg)
+ return site_name
+
+
+SiteName = Annotated[str, AfterValidator(validate_site_name)]
diff --git a/gso/utils/types/snmp.py b/gso/utils/types/snmp.py
new file mode 100644
index 0000000000000000000000000000000000000000..03581cf970c036db37ea901d8b159d25fc480136
--- /dev/null
+++ b/gso/utils/types/snmp.py
@@ -0,0 +1,10 @@
+"""An enumerator of SNMP version numbers."""
+
+from enum import StrEnum
+
+
+class SNMPVersion(StrEnum):
+ """An enumerator for the two relevant versions of :term:`SNMP`: v2c and 3."""
+
+ V2C = "v2c"
+ V3 = "v3"
diff --git a/gso/utils/types/tt_number.py b/gso/utils/types/tt_number.py
new file mode 100644
index 0000000000000000000000000000000000000000..a9e1d05c6ad9563047764d56461afe846b99be88
--- /dev/null
+++ b/gso/utils/types/tt_number.py
@@ -0,0 +1,29 @@
+"""A Trouble Ticket number."""
+
+import re
+from typing import Annotated
+
+from pydantic import AfterValidator
+
+
+def validate_tt_number(tt_number: str) -> str:
+ """Validate a string to match a specific pattern.
+
+ This method checks if the input string starts with 'TT#' and is followed by exactly 16 digits.
+
+ :param str tt_number: The TT number as string to validate
+
+ :return str: The TT number string if TT number match was successful, otherwise it will raise a ValueError.
+ """
+ pattern = r"^TT#\d{16}$"
+ if not bool(re.match(pattern, tt_number)):
+ err_msg = (
+ f"The given TT number: {tt_number} is not valid. "
+ f" A valid TT number starts with 'TT#' followed by 16 digits."
+ )
+ raise ValueError(err_msg)
+
+ return tt_number
+
+
+TTNumber = Annotated[str, AfterValidator(validate_tt_number)]
diff --git a/gso/utils/types/unique_field.py b/gso/utils/types/unique_field.py
new file mode 100644
index 0000000000000000000000000000000000000000..32c3d9de43560998dea262b512625de60363de44
--- /dev/null
+++ b/gso/utils/types/unique_field.py
@@ -0,0 +1,21 @@
+"""An input field that must be unique in the database."""
+
+from typing import Annotated, TypeVar
+
+from pydantic import AfterValidator
+from pydantic_core.core_schema import ValidationInfo
+
+from gso.services import subscriptions
+
+T = TypeVar("T")
+
+
+def validate_field_is_unique(value: T, info: ValidationInfo) -> T:
+ """Validate that a field is unique."""
+ if len(subscriptions.get_active_subscriptions_by_field_and_value(str(info.field_name), str(value))) > 0:
+ msg = f"{info.field_name} must be unique"
+ raise ValueError(msg)
+ return value
+
+
+UniqueField = Annotated[T, AfterValidator(validate_field_is_unique)]
diff --git a/gso/workflows/iptrunk/create_imported_iptrunk.py b/gso/workflows/iptrunk/create_imported_iptrunk.py
index 9b0e6b87a8a095073875721e79d72c9e03baa66d..4d75437504ac6536cf56c8b2f5231bea0a937db8 100644
--- a/gso/workflows/iptrunk/create_imported_iptrunk.py
+++ b/gso/workflows/iptrunk/create_imported_iptrunk.py
@@ -6,7 +6,6 @@ from uuid import uuid4
from orchestrator import workflow
from orchestrator.forms import FormPage
-from orchestrator.forms.validators import Choice
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
from orchestrator.workflow import StepList, begin, done, step
@@ -15,30 +14,17 @@ from pydantic import AfterValidator, ConfigDict
from pydantic_forms.validators import validate_unique_list
from gso.products import ProductName
-from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlockInactive, IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlockInactive, IptrunkType
from gso.products.product_types.iptrunk import ImportedIptrunkInactive
from gso.products.product_types.router import Router
from gso.services import subscriptions
from gso.services.partners import get_partner_by_name
-from gso.utils.helpers import LAGMember
-
-
-def _generate_routers() -> dict[str, str]:
- """Generate a dictionary of router IDs and descriptions."""
- routers = {}
- for subscription in subscriptions.get_active_router_subscriptions(includes=["subscription_id", "description"]):
- routers[str(subscription["subscription_id"])] = subscription["description"]
-
- return routers
-
-
-LAGMemberList = Annotated[list[LAGMember], AfterValidator(validate_unique_list)]
+from gso.utils.helpers import active_router_selector
+from gso.utils.types.interfaces import LAGMember, LAGMemberList, PhysicalPortCapacity
def initial_input_form_generator() -> FormGenerator:
"""Take all information passed to this workflow by the :term:`API` endpoint that was called."""
- routers = _generate_routers()
- router_enum = Choice("Select a router", zip(routers.keys(), routers.items(), strict=True)) # type: ignore[arg-type]
class CreateIptrunkForm(FormPage):
model_config = ConfigDict(title="Import Iptrunk")
@@ -51,15 +37,15 @@ def initial_input_form_generator() -> FormGenerator:
iptrunk_minimum_links: int
iptrunk_isis_metric: int
- side_a_node_id: router_enum # type: ignore[valid-type]
+ side_a_node_id: active_router_selector() # type: ignore[valid-type]
side_a_ae_iface: str
side_a_ae_geant_a_sid: str | None = None
- side_a_ae_members: LAGMemberList
+ side_a_ae_members: Annotated[list[LAGMember], AfterValidator(validate_unique_list)]
- side_b_node_id: router_enum # type: ignore[valid-type]
+ side_b_node_id: active_router_selector() # type: ignore[valid-type]
side_b_ae_iface: str
side_b_ae_geant_a_sid: str | None = None
- side_b_ae_members: LAGMemberList
+ side_b_ae_members: Annotated[list[LAGMember], AfterValidator(validate_unique_list)]
iptrunk_ipv4_network: ipaddress.IPv4Network
iptrunk_ipv6_network: ipaddress.IPv6Network
@@ -94,11 +80,11 @@ def initialize_subscription(
side_a_node_id: str,
side_a_ae_iface: str,
side_a_ae_geant_a_sid: str | None,
- side_a_ae_members: list[dict],
+ side_a_ae_members: LAGMemberList,
side_b_node_id: str,
side_b_ae_iface: str,
side_b_ae_geant_a_sid: str | None,
- side_b_ae_members: list[dict],
+ side_b_ae_members: LAGMemberList,
) -> State:
"""Take all input from the user, and store it in the database."""
subscription.iptrunk.geant_s_sid = geant_s_sid
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index 3cd8d81cd14b6932f7ceafac247e752ec54d625a..56907a6c22e52c5f7d82642563e78d0a62b09853 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/gso/workflows/iptrunk/create_iptrunk.py
@@ -16,15 +16,14 @@ from orchestrator.workflow import StepList, begin, conditional, done, step, step
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
from orchestrator.workflows.utils import wrap_create_initial_input_form
from ping3 import ping
-from pydantic import AfterValidator, ConfigDict, field_validator
-from pydantic_forms.validators import ReadOnlyField, validate_unique_list
+from pydantic import ConfigDict
+from pydantic_forms.validators import ReadOnlyField
from pynetbox.models.dcim import Interfaces
from gso.products.product_blocks.iptrunk import (
IptrunkInterfaceBlockInactive,
IptrunkSideBlockInactive,
IptrunkType,
- PhysicalPortCapacity,
)
from gso.products.product_types.iptrunk import Iptrunk, IptrunkInactive, IptrunkProvisioning
from gso.products.product_types.router import Router
@@ -36,28 +35,25 @@ from gso.services.sharepoint import SharePointClient
from gso.services.subscriptions import get_non_terminated_iptrunk_subscriptions
from gso.settings import load_oss_params
from gso.utils.helpers import (
- LAGMember,
available_interfaces_choices,
available_lags_choices,
calculate_recommended_minimum_links,
get_router_vendor,
- validate_interface_name_list,
- validate_iptrunk_unique_interface,
- validate_router_in_netbox,
)
from gso.utils.shared_enums import Vendor
-from gso.utils.types import TTNumber
+from gso.utils.types.interfaces import JuniperLAGMember, LAGMember, LAGMemberList, PhysicalPortCapacity
+from gso.utils.types.netbox_router import NetboxEnabledRouter
+from gso.utils.types.tt_number import TTNumber
from gso.utils.workflow_steps import prompt_sharepoint_checklist_url
def initial_input_form_generator(product_name: str) -> FormGenerator:
"""Gather input from the user in three steps. General information, and information on both sides of the trunk."""
- routers = {}
- for router in subscriptions.get_active_router_subscriptions(
+ # Add both provisioning and active routers, since trunks are required for promoting a router to active.
+ active_and_provisioning_routers = subscriptions.get_active_router_subscriptions(
includes=["subscription_id", "description"]
- ) + subscriptions.get_provisioning_router_subscriptions(includes=["subscription_id", "description"]):
- # Add both provisioning and active routers, since trunks are required for promoting a router to active.
- routers[str(router["subscription_id"])] = router["description"]
+ ) + subscriptions.get_provisioning_router_subscriptions(includes=["subscription_id", "description"])
+ routers = {str(router["subscription_id"]): router["description"] for router in active_and_provisioning_routers}
class CreateIptrunkForm(FormPage):
model_config = ConfigDict(title=product_name)
@@ -86,19 +82,14 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
class SelectRouterSideA(FormPage):
model_config = ConfigDict(title="Select a router for side A of the trunk.")
- side_a_node_id: router_enum_a # type: ignore[valid-type]
-
- @field_validator("side_a_node_id")
- def validate_device_exists_in_netbox(cls, side_a_node_id: UUIDstr) -> str | None:
- return validate_router_in_netbox(side_a_node_id)
+ side_a_node_id: NetboxEnabledRouter[router_enum_a] # type: ignore[valid-type]
user_input_router_side_a = yield SelectRouterSideA
router_a = user_input_router_side_a.side_a_node_id.name
router_a_fqdn = Router.from_subscription(router_a).router.router_fqdn
juniper_ae_members = Annotated[
- list[LAGMember],
- AfterValidator(validate_unique_list),
+ LAGMemberList[JuniperLAGMember],
Len(
min_length=initial_user_input.iptrunk_number_of_members,
max_length=initial_user_input.iptrunk_number_of_members,
@@ -114,8 +105,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
)
ae_members_side_a_type = Annotated[
- list[NokiaLAGMemberA],
- AfterValidator(validate_unique_list),
+ LAGMemberList[NokiaLAGMemberA],
Len(
min_length=initial_user_input.iptrunk_number_of_members,
max_length=initial_user_input.iptrunk_number_of_members,
@@ -125,19 +115,12 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
ae_members_side_a_type = juniper_ae_members # type: ignore[assignment, misc]
class CreateIptrunkSideAForm(FormPage):
- model_config = ConfigDict(title=f"Provide subscription details for side A of the trunk.({router_a_fqdn})")
+ model_config = ConfigDict(title=f"Provide subscription details for side A of the trunk. ({router_a_fqdn})")
side_a_ae_iface: available_lags_choices(router_a) or str # type: ignore[valid-type]
side_a_ae_geant_a_sid: str | None
side_a_ae_members: ae_members_side_a_type
- @field_validator("side_a_ae_members")
- 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)
- validate_interface_name_list(side_a_ae_members, vendor)
- return side_a_ae_members
-
user_input_side_a = yield CreateIptrunkSideAForm
# Remove the selected router for side A, to prevent any loops
routers.pop(str(router_a))
@@ -146,11 +129,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
class SelectRouterSideB(FormPage):
model_config = ConfigDict(title="Select a router for side B of the trunk.")
- side_b_node_id: router_enum_b # type: ignore[valid-type]
-
- @field_validator("side_b_node_id")
- def validate_device_exists_in_netbox(cls, side_b_node_id: UUIDstr) -> str | None:
- return validate_router_in_netbox(side_b_node_id)
+ side_b_node_id: NetboxEnabledRouter[router_enum_b] # type: ignore[valid-type]
user_input_router_side_b = yield SelectRouterSideB
router_b = user_input_router_side_b.side_b_node_id.name
@@ -165,8 +144,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
)
ae_members_side_b = Annotated[
- list[NokiaLAGMemberB],
- AfterValidator(validate_unique_list),
+ LAGMemberList[NokiaLAGMemberB],
Len(
min_length=len(user_input_side_a.side_a_ae_members), max_length=len(user_input_side_a.side_a_ae_members)
),
@@ -175,19 +153,12 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
ae_members_side_b = juniper_ae_members # type: ignore[assignment, misc]
class CreateIptrunkSideBForm(FormPage):
- model_config = ConfigDict(title=f"Provide subscription details for side B of the trunk.({router_b_fqdn})")
+ model_config = ConfigDict(title=f"Provide subscription details for side B of the trunk. ({router_b_fqdn})")
side_b_ae_iface: available_lags_choices(router_b) or str # type: ignore[valid-type]
side_b_ae_geant_a_sid: str | None
side_b_ae_members: ae_members_side_b
- @field_validator("side_b_ae_members")
- 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)
- validate_interface_name_list(side_b_ae_members, vendor)
- return side_b_ae_members
-
user_input_side_b = yield CreateIptrunkSideBForm
return (
@@ -582,6 +553,7 @@ def create_new_sharepoint_checklist(subscription: IptrunkProvisioning, tt_number
"Title": f"{subscription.description} - {subscription.iptrunk.geant_s_sid}",
"TT_NUMBER": tt_number,
"GAP_PROCESS_URL": f"{load_oss_params().GENERAL.public_hostname}/workflows/{process_id}",
+ "ACTIVITY_TYPE": "Creation",
},
)
diff --git a/gso/workflows/iptrunk/deploy_twamp.py b/gso/workflows/iptrunk/deploy_twamp.py
index a45b5eca61144577c5dbf58a251b0b1ca7c76f6d..11dce53ba260fd8c8dae08611ae66e4069619d49 100644
--- a/gso/workflows/iptrunk/deploy_twamp.py
+++ b/gso/workflows/iptrunk/deploy_twamp.py
@@ -13,7 +13,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
from gso.products.product_types.iptrunk import Iptrunk
from gso.services.lso_client import execute_playbook, lso_interaction
-from gso.utils.types import TTNumber
+from gso.utils.types.tt_number import TTNumber
def _initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index be15848ae73f933fa262211e765b0901313dac66..c7021875a1d7f2e2fde5424b5493de028b209d58 100644
--- a/gso/workflows/iptrunk/migrate_iptrunk.py
+++ b/gso/workflows/iptrunk/migrate_iptrunk.py
@@ -6,7 +6,6 @@ configured to run from A to C. B is then no longer associated with this IP trunk
import copy
import json
-import re
from typing import Annotated
from uuid import uuid4
@@ -21,8 +20,8 @@ from orchestrator.utils.json import json_dumps
from orchestrator.workflow import StepList, begin, conditional, done, inputstep
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, field_validator
-from pydantic_forms.validators import ReadOnlyField, validate_unique_list
+from pydantic import ConfigDict
+from pydantic_forms.validators import ReadOnlyField
from pynetbox.models.dcim import Interfaces
from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock, IptrunkType
@@ -31,17 +30,18 @@ from gso.products.product_types.router import Router
from gso.services import infoblox
from gso.services.lso_client import execute_playbook, lso_interaction
from gso.services.netbox_client import NetboxClient
+from gso.services.sharepoint import SharePointClient
from gso.services.subscriptions import get_active_router_subscriptions
+from gso.settings import load_oss_params
from gso.utils.helpers import (
- LAGMember,
available_interfaces_choices,
available_lags_choices,
get_router_vendor,
- validate_interface_name_list,
)
from gso.utils.shared_enums import Vendor
-from gso.utils.types import TTNumber
-from gso.utils.workflow_steps import set_isis_to_max
+from gso.utils.types.interfaces import JuniperAEInterface, JuniperLAGMember, LAGMember, LAGMemberList
+from gso.utils.types.tt_number import TTNumber
+from gso.utils.workflow_steps import prompt_sharepoint_checklist_url, set_isis_to_max
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -101,7 +101,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
new_side_iptrunk_router_input = yield NewSideIPTrunkRouterForm
new_router = new_side_iptrunk_router_input.new_node
- side_a_ae_iface = available_lags_choices(new_router) or str
+ side_a_ae_iface = available_lags_choices(new_router) or JuniperAEInterface
new_side_is_nokia = get_router_vendor(new_router) == Vendor.NOKIA
if new_side_is_nokia:
@@ -113,8 +113,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
)
ae_members = Annotated[
- list[NokiaLAGMember],
- AfterValidator(validate_unique_list),
+ LAGMemberList[NokiaLAGMember],
Len(
min_length=len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members),
max_length=len(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members),
@@ -122,8 +121,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
]
else:
ae_members = Annotated[ # type: ignore[assignment, misc]
- list[LAGMember],
- AfterValidator(validate_unique_list),
+ LAGMemberList[JuniperLAGMember],
Len(
min_length=len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members),
max_length=len(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members),
@@ -148,23 +146,9 @@ 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: ReadOnlyField(existing_lag_ae_members, default_type=list[LAGMember]) # type: ignore[valid-type]
+ existing_lag_interface: ReadOnlyField(existing_lag_ae_members, default_type=LAGMemberList[LAGMember]) # type: ignore[valid-type]
new_lag_member_interfaces: ae_members
- @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}$")
- if not juniper_lag_re.match(new_lag_interface):
- msg = "Invalid LAG name, please try again."
- raise ValueError(msg)
- return new_lag_interface
-
- @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)
-
new_side_input = yield NewSideIPTrunkForm
return (
migrate_form_input.model_dump()
@@ -807,6 +791,22 @@ def netbox_allocate_new_interfaces(subscription: Iptrunk, replace_index: int) ->
return {"subscription": subscription}
+@step("Create a new SharePoint checklist item")
+def create_new_sharepoint_checklist(subscription: Iptrunk, tt_number: str, process_id: UUIDstr) -> State:
+ """Create a new checklist item in SharePoint for approving this migrated IPtrunk."""
+ new_list_item_url = SharePointClient().add_list_item(
+ list_name="ip_trunk",
+ fields={
+ "Title": f"{subscription.description} - {subscription.iptrunk.geant_s_sid}",
+ "TT_NUMBER": tt_number,
+ "GAP_PROCESS_URL": f"{load_oss_params().GENERAL.public_hostname}/workflows/{process_id}",
+ "ACTIVITY_TYPE": "Migration",
+ },
+ )
+
+ return {"checklist_url": new_list_item_url}
+
+
@workflow(
"Migrate an IP Trunk",
initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
@@ -868,5 +868,7 @@ def migrate_iptrunk() -> StepList:
>> old_side_is_nokia(netbox_remove_old_interfaces)
>> new_side_is_nokia(netbox_allocate_new_interfaces)
>> resync
+ >> create_new_sharepoint_checklist
+ >> prompt_sharepoint_checklist_url
>> done
)
diff --git a/gso/workflows/iptrunk/modify_isis_metric.py b/gso/workflows/iptrunk/modify_isis_metric.py
index 285907b45508249794bb8c5fd486b62ed0b4dac6..da14d078cea88f0c52256e02ef3dce0429cf7374 100644
--- a/gso/workflows/iptrunk/modify_isis_metric.py
+++ b/gso/workflows/iptrunk/modify_isis_metric.py
@@ -12,7 +12,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
from gso.products.product_types.iptrunk import Iptrunk
from gso.services.lso_client import execute_playbook, lso_interaction
-from gso.utils.types import TTNumber
+from gso.utils.types.tt_number import TTNumber
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
diff --git a/gso/workflows/iptrunk/modify_trunk_interface.py b/gso/workflows/iptrunk/modify_trunk_interface.py
index 394e369a88d1f64481750080410ffcb8f4335064..6d82090c56d539c3456344ec6c08b1580d33dcf4 100644
--- a/gso/workflows/iptrunk/modify_trunk_interface.py
+++ b/gso/workflows/iptrunk/modify_trunk_interface.py
@@ -1,7 +1,7 @@
"""A modification workflow that updates the :term:`LAG` interfaces that are part of an existing IP trunk."""
import json
-from typing import Annotated, TypeVar
+from typing import Annotated
from uuid import UUID, uuid4
from annotated_types import Len
@@ -12,38 +12,34 @@ 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 AfterValidator, ConfigDict, field_validator
-from pydantic_forms.validators import Label, ReadOnlyField, validate_unique_list
+from pydantic import ConfigDict
+from pydantic_forms.validators import Label, ReadOnlyField
from gso.products.product_blocks.iptrunk import (
IptrunkInterfaceBlock,
IptrunkSideBlock,
IptrunkType,
- PhysicalPortCapacity,
)
from gso.products.product_types.iptrunk import Iptrunk
from gso.services.lso_client import execute_playbook, lso_interaction
from gso.services.netbox_client import NetboxClient
from gso.utils.helpers import (
- LAGMember,
available_interfaces_choices,
available_interfaces_choices_including_current_members,
calculate_recommended_minimum_links,
get_router_vendor,
- validate_interface_name_list,
- validate_iptrunk_unique_interface,
)
-from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, Vendor
-from gso.utils.types import TTNumber
+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.workflows.iptrunk.migrate_iptrunk import check_ip_trunk_optical_levels_pre
from gso.workflows.iptrunk.validate_iptrunk import check_ip_trunk_isis
-T = TypeVar("T", bound=LAGMember)
-
def initialize_ae_members(
subscription: Iptrunk, initial_user_input: dict, side_index: int
-) -> Annotated[list[LAGMember], ""]:
+) -> type[LAGMemberList[LAGMember]]:
"""Initialize the list of AE members."""
router = subscription.iptrunk.iptrunk_sides[side_index].iptrunk_side_node
router_vendor = get_router_vendor(router.owner_subscription_id)
@@ -67,17 +63,15 @@ def initialize_ae_members(
)
)
- return Annotated[
- list[NokiaLAGMember],
- AfterValidator(validate_unique_list),
+ return Annotated[ # type: ignore[return-value]
+ LAGMemberList[NokiaLAGMember],
Len(min_length=iptrunk_number_of_members, max_length=iptrunk_number_of_members),
- ] # type: ignore[return-value]
+ ]
- return Annotated[
- list[LAGMember],
- AfterValidator(validate_unique_list),
+ return Annotated[ # type: ignore[return-value]
+ LAGMemberList[JuniperLAGMember],
Len(min_length=iptrunk_number_of_members, max_length=iptrunk_number_of_members),
- ] # type: ignore[return-value]
+ ]
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -132,15 +126,6 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
else []
)
- @field_validator("side_a_ae_members")
- def validate_iptrunk_unique_interface_side_a(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
- return validate_iptrunk_unique_interface(side_a_ae_members)
-
- @field_validator("side_a_ae_members")
- def validate_interface_name_members(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
- vendor = subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.vendor
- return validate_interface_name_list(side_a_ae_members, vendor)
-
user_input_side_a = yield ModifyIptrunkSideAForm
ae_members_side_b = initialize_ae_members(subscription, initial_user_input.model_dump(), 1)
@@ -158,15 +143,6 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
else []
)
- @field_validator("side_b_ae_members")
- def validate_iptrunk_unique_interface_side_b(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
- return validate_iptrunk_unique_interface(side_b_ae_members)
-
- @field_validator("side_b_ae_members")
- def validate_interface_name_members(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
- vendor = subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.vendor
- return validate_interface_name_list(side_b_ae_members, vendor)
-
user_input_side_b = yield ModifyIptrunkSideBForm
return (
@@ -179,7 +155,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@step("Determine whether we should be running interface checks")
def determine_change_in_capacity(
- subscription: Iptrunk, iptrunk_speed: str, side_a_ae_members: list[LAGMember], side_b_ae_members: list[LAGMember]
+ subscription: Iptrunk, iptrunk_speed: str, side_a_ae_members: list[dict], side_b_ae_members: list[dict]
) -> State:
"""Determine whether we should run pre- and post-checks on the IP trunk.
@@ -193,13 +169,13 @@ def determine_change_in_capacity(
iptrunk_speed != subscription.iptrunk.iptrunk_speed
or len(side_a_ae_members) != len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members)
or any(
- old_interface.interface_name != new_interface.interface_name
+ old_interface.interface_name != new_interface["interface_name"]
for old_interface, new_interface in zip(
subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members, side_a_ae_members, strict=False
)
)
or any(
- old_interface.interface_name != new_interface.interface_name
+ old_interface.interface_name != new_interface["interface_name"]
for old_interface, new_interface in zip(
subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members, side_b_ae_members, strict=False
)
diff --git a/gso/workflows/iptrunk/terminate_iptrunk.py b/gso/workflows/iptrunk/terminate_iptrunk.py
index bb1a6fd90b3d9a5e3b9aa9bb633db58cc2eb1cd4..ac628d393a9c6dd52880ee28ceb39d403f023fd2 100644
--- a/gso/workflows/iptrunk/terminate_iptrunk.py
+++ b/gso/workflows/iptrunk/terminate_iptrunk.py
@@ -24,7 +24,7 @@ from gso.services.lso_client import execute_playbook, lso_interaction
from gso.services.netbox_client import NetboxClient
from gso.utils.helpers import get_router_vendor
from gso.utils.shared_enums import Vendor
-from gso.utils.types import TTNumber
+from gso.utils.types.tt_number import TTNumber
from gso.utils.workflow_steps import set_isis_to_max
diff --git a/gso/workflows/lan_switch_interconnect/__init__.py b/gso/workflows/lan_switch_interconnect/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..30d685bf043395ff4a861ee3d9b50efffdf511d5
--- /dev/null
+++ b/gso/workflows/lan_switch_interconnect/__init__.py
@@ -0,0 +1 @@
+"""Workflows for the LAN Switch interconnect product."""
diff --git a/gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py b/gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py
new file mode 100644
index 0000000000000000000000000000000000000000..973cf1d88495b4343e390c258d16c3c9f78f65c9
--- /dev/null
+++ b/gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py
@@ -0,0 +1,162 @@
+"""A creation workflow for creating a new interconnect between a switch and a router."""
+
+from typing import Annotated
+from uuid import uuid4
+
+from annotated_types import Len
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
+from orchestrator.workflow import StepList, begin, done, 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 AfterValidator, ConfigDict
+from pydantic_forms.validators import Divider, ReadOnlyField
+
+from gso.products.product_blocks.lan_switch_interconnect import (
+ LanSwitchInterconnectAddressSpace,
+ LanSwitchInterconnectInterfaceBlockInactive,
+)
+from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnectInactive
+from gso.products.product_types.router import Router
+from gso.products.product_types.switch import Switch
+from gso.services.partners import get_partner_by_name
+from gso.utils.helpers import (
+ active_router_selector,
+ active_switch_selector,
+ available_interfaces_choices,
+ available_lags_choices,
+)
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.interfaces import (
+ JuniperAEInterface,
+ JuniperLAGMember,
+ JuniperPhyInterface,
+ LAGMember,
+ PhysicalPortCapacity,
+ validate_interface_names_are_unique,
+)
+from gso.utils.types.tt_number import TTNumber
+
+
+def _initial_input_form(product_name: str) -> FormGenerator:
+ class CreateLANSwitchInterconnectForm(FormPage):
+ model_config = ConfigDict(title=product_name)
+
+ tt_number: TTNumber
+ router_side: active_router_selector() # type: ignore[valid-type]
+ switch_side: active_switch_selector() # type: ignore[valid-type]
+ address_space: LanSwitchInterconnectAddressSpace
+ description: str
+ minimum_link_count: int
+ divider: Divider
+ vlan_id: ReadOnlyField(111, default_type=int) # type: ignore[valid-type]
+
+ user_input = yield CreateLANSwitchInterconnectForm
+ router = Router.from_subscription(user_input.router_side)
+
+ if router.router.vendor == Vendor.NOKIA:
+
+ class NokiaLAGMemberA(LAGMember):
+ interface_name: available_interfaces_choices( # type: ignore[valid-type]
+ router.subscription_id,
+ PhysicalPortCapacity.TEN_GIGABIT_PER_SECOND,
+ )
+
+ router_side_ae_member_list = Annotated[
+ list[NokiaLAGMemberA],
+ AfterValidator(validate_interface_names_are_unique),
+ Len(min_length=user_input.minimum_link_count),
+ ]
+ else:
+ router_side_ae_member_list = Annotated[ # type: ignore[assignment, misc]
+ list[JuniperLAGMember],
+ AfterValidator(validate_interface_names_are_unique),
+ Len(min_length=user_input.minimum_link_count),
+ ]
+
+ class InterconnectRouterSideForm(FormPage):
+ model_config = ConfigDict(title="Please enter interface names and descriptions for the router side.")
+
+ router_side_iface: available_lags_choices(user_input.router_side) or JuniperAEInterface # type: ignore[valid-type]
+ router_side_ae_members: router_side_ae_member_list
+
+ router_side_input = yield InterconnectRouterSideForm
+
+ switch_side_ae_member_list = Annotated[
+ list[JuniperLAGMember],
+ AfterValidator(validate_interface_names_are_unique),
+ Len(
+ min_length=len(router_side_input.router_side_ae_members),
+ max_length=len(router_side_input.router_side_ae_members),
+ ),
+ ]
+
+ class InterconnectSwitchSideForm(FormPage):
+ model_config = ConfigDict(title="Please enter interface names and descriptions for the switch side.")
+
+ switch_side_iface: JuniperPhyInterface
+ switch_side_ae_members: switch_side_ae_member_list
+
+ switch_side_input = yield InterconnectSwitchSideForm
+
+ return user_input.model_dump() | router_side_input.model_dump() | switch_side_input.model_dump()
+
+
+@step("Create subscription")
+def create_subscription(product: UUIDstr, partner: str) -> State:
+ """Create a new subscription object in the database."""
+ subscription = LanSwitchInterconnectInactive.from_product_id(product, get_partner_by_name(partner)["partner_id"])
+
+ return {"subscription": subscription}
+
+
+@step("Initialize subscription")
+def initialize_subscription(
+ subscription: LanSwitchInterconnectInactive,
+ description: str,
+ address_space: LanSwitchInterconnectAddressSpace,
+ minimum_link_count: int,
+ router_side: UUIDstr,
+ router_side_iface: JuniperPhyInterface,
+ router_side_ae_members: list[dict],
+ switch_side: UUIDstr,
+ switch_side_iface: JuniperPhyInterface,
+ switch_side_ae_members: list[dict],
+) -> State:
+ """Update the product model with all input from the operator."""
+ subscription.lan_switch_interconnect.lan_switch_interconnect_description = description
+ subscription.lan_switch_interconnect.address_space = address_space
+ subscription.lan_switch_interconnect.minimum_links = minimum_link_count
+ subscription.lan_switch_interconnect.router_side.node = Router.from_subscription(router_side).router
+ subscription.lan_switch_interconnect.router_side.ae_iface = router_side_iface
+ for member in router_side_ae_members:
+ subscription.lan_switch_interconnect.router_side.ae_members.append(
+ LanSwitchInterconnectInterfaceBlockInactive.new(subscription_id=uuid4(), **member)
+ )
+ subscription.lan_switch_interconnect.switch_side.node = Switch.from_subscription(switch_side).switch
+ subscription.lan_switch_interconnect.switch_side.ae_iface = switch_side_iface
+ for member in switch_side_ae_members:
+ subscription.lan_switch_interconnect.switch_side.ae_members.append(
+ LanSwitchInterconnectInterfaceBlockInactive.new(subscription_id=uuid4(), **member)
+ )
+
+ return {"subscription": subscription}
+
+
+@workflow(
+ "Create LAN switch interconnect",
+ initial_input_form=wrap_create_initial_input_form(_initial_input_form),
+ target=Target.CREATE,
+)
+def create_lan_switch_interconnect() -> StepList:
+ """Create a new LAN interconnect between a Switch and a Router."""
+ return (
+ begin
+ >> create_subscription
+ >> store_process_subscription(Target.CREATE)
+ >> initialize_subscription
+ >> set_status(SubscriptionLifecycle.ACTIVE)
+ >> resync
+ >> done
+ )
diff --git a/gso/workflows/office_router/create_imported_office_router.py b/gso/workflows/office_router/create_imported_office_router.py
index e82f69cbcb31ecc7eadfbd16c757232a3d47bb2a..8b8306343f6dcc29ae97ef7526c8d399e0b6e444 100644
--- a/gso/workflows/office_router/create_imported_office_router.py
+++ b/gso/workflows/office_router/create_imported_office_router.py
@@ -13,7 +13,8 @@ from gso.products.product_types.office_router import ImportedOfficeRouterInactiv
from gso.services import subscriptions
from gso.services.partners import get_partner_by_name
from gso.services.subscriptions import get_site_by_name
-from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType, PortNumber
@step("Create subscription")
diff --git a/gso/workflows/opengear/create_imported_opengear.py b/gso/workflows/opengear/create_imported_opengear.py
index e7699986018efc53ddd6254c2d827a47ee791cfd..ce4bd3cf039e7323ce861c11feea41f87e647947 100644
--- a/gso/workflows/opengear/create_imported_opengear.py
+++ b/gso/workflows/opengear/create_imported_opengear.py
@@ -12,7 +12,7 @@ from gso.products import ProductName
from gso.products.product_types.opengear import ImportedOpengearInactive
from gso.services.partners import get_partner_by_name
from gso.services.subscriptions import get_product_id_by_name, get_site_by_name
-from gso.utils.shared_enums import IPv4AddressType
+from gso.utils.types.ip_address import IPv4AddressType
@step("Create subscription")
diff --git a/gso/workflows/router/create_imported_router.py b/gso/workflows/router/create_imported_router.py
index c8ba442bc30bf8d26b563e83f35b7d4fb1573774..12a71c79b51e0166d976a175d9833870300af1f5 100644
--- a/gso/workflows/router/create_imported_router.py
+++ b/gso/workflows/router/create_imported_router.py
@@ -14,7 +14,8 @@ from gso.products.product_types.router import ImportedRouterInactive
from gso.services.partners import get_partner_by_name
from gso.services.subscriptions import get_product_id_by_name, get_site_by_name
from gso.utils.helpers import generate_fqdn
-from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType, PortNumber
@step("Create subscription")
diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index 59dd6210578d9556fdf79ecacf7df0b5518fdbd5..04d38b750f5fa4ca87591f7affe04674208685bf 100644
--- a/gso/workflows/router/create_router.py
+++ b/gso/workflows/router/create_router.py
@@ -24,8 +24,9 @@ from gso.services.partners import get_partner_by_name
from gso.services.sharepoint import SharePointClient
from gso.settings import load_oss_params
from gso.utils.helpers import generate_fqdn, iso_from_ipv4
-from gso.utils.shared_enums import PortNumber, Vendor
-from gso.utils.types import TTNumber
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.ip_address import PortNumber
+from gso.utils.types.tt_number import TTNumber
from gso.utils.workflow_steps import (
deploy_base_config_dry,
deploy_base_config_real,
diff --git a/gso/workflows/router/promote_p_to_pe.py b/gso/workflows/router/promote_p_to_pe.py
index 74f478c075e615900ea5e3a8c0e996aca94b364c..0f0b31da1fcf7efe8049e9d24cd1e20726bff9c0 100644
--- a/gso/workflows/router/promote_p_to_pe.py
+++ b/gso/workflows/router/promote_p_to_pe.py
@@ -23,7 +23,7 @@ from gso.services.lso_client import lso_interaction
from gso.services.subscriptions import get_all_active_sites
from gso.utils.helpers import generate_inventory_for_active_routers
from gso.utils.shared_enums import Vendor
-from gso.utils.types import TTNumber
+from gso.utils.types.tt_number import TTNumber
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -440,7 +440,7 @@ def update_subscription_model(subscription: Router) -> State:
return {"subscription": subscription}
-@step("[DRY RUN] Add all P to PE")
+@step("[DRY RUN] Add all P to this new PE")
def add_all_p_to_pe_dry(subscription: dict[str, Any], callback_route: str, tt_number: str, process_id: UUIDstr) -> None:
"""Perform a dry run of adding all P routers to the PE router."""
extra_vars = {
@@ -448,7 +448,9 @@ def add_all_p_to_pe_dry(subscription: dict[str, Any], callback_route: str, tt_nu
"subscription": subscription,
"commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Add all P-routers to this new PE",
"verb": "add_all_p_to_pe",
- "p_router_list": generate_inventory_for_active_routers(RouterRole.P)["all"]["hosts"],
+ "p_router_list": generate_inventory_for_active_routers(
+ RouterRole.P, exclude_routers=[subscription["router"]["router_fqdn"]]
+ )["all"]["hosts"],
}
lso_client.execute_playbook(
@@ -459,7 +461,7 @@ def add_all_p_to_pe_dry(subscription: dict[str, Any], callback_route: str, tt_nu
)
-@step("[FOR REAL] Add all P to PE")
+@step("[FOR REAL] Add all P to this new PE")
def add_all_p_to_pe_real(
subscription: dict[str, Any], callback_route: str, tt_number: str, process_id: UUIDstr
) -> None:
@@ -469,7 +471,9 @@ def add_all_p_to_pe_real(
"subscription": subscription,
"commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Add all P-routers to this new PE",
"verb": "add_all_p_to_pe",
- "p_router_list": generate_inventory_for_active_routers(RouterRole.P)["all"]["hosts"],
+ "p_router_list": generate_inventory_for_active_routers(
+ RouterRole.P, exclude_routers=[subscription["router"]["router_fqdn"]]
+ )["all"]["hosts"],
}
lso_client.execute_playbook(
@@ -480,7 +484,7 @@ def add_all_p_to_pe_real(
)
-@step("[DRY RUN] Add PE to all P")
+@step("[DRY RUN] Add this new PE to all P")
def add_pe_to_all_p_dry(subscription: dict[str, Any], callback_route: str, tt_number: str, process_id: UUIDstr) -> None:
"""Perform a dry run of adding promoted router to all PE routers in iGEANT/iGEANT6."""
extra_vars = {
@@ -494,12 +498,14 @@ def add_pe_to_all_p_dry(subscription: dict[str, Any], callback_route: str, tt_nu
lso_client.execute_playbook(
playbook_name="update_ibgp_mesh.yaml",
callback_route=callback_route,
- inventory=generate_inventory_for_active_routers(RouterRole.P),
+ inventory=generate_inventory_for_active_routers(
+ RouterRole.P, exclude_routers=[subscription["router"]["router_fqdn"]]
+ ),
extra_vars=extra_vars,
)
-@step("[FOR REAL] Add PE to all P")
+@step("[FOR REAL] Add this new PE to all P")
def add_pe_to_all_p_real(
subscription: dict[str, Any], callback_route: str, tt_number: str, process_id: UUIDstr
) -> None:
@@ -515,7 +521,9 @@ def add_pe_to_all_p_real(
lso_client.execute_playbook(
playbook_name="update_ibgp_mesh.yaml",
callback_route=callback_route,
- inventory=generate_inventory_for_active_routers(RouterRole.P),
+ inventory=generate_inventory_for_active_routers(
+ RouterRole.P, exclude_routers=[subscription["router"]["router_fqdn"]]
+ ),
extra_vars=extra_vars,
)
diff --git a/gso/workflows/router/redeploy_base_config.py b/gso/workflows/router/redeploy_base_config.py
index b30d02f16b6bb197a20257f9016eabb6dad0d8fa..48348abb6757ff38b40f650c88cb920bc3fda95c 100644
--- a/gso/workflows/router/redeploy_base_config.py
+++ b/gso/workflows/router/redeploy_base_config.py
@@ -10,7 +10,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
from gso.products.product_types.router import Router
from gso.services.lso_client import lso_interaction
-from gso.utils.types import TTNumber
+from gso.utils.types.tt_number import TTNumber
from gso.utils.workflow_steps import deploy_base_config_dry, deploy_base_config_real
diff --git a/gso/workflows/router/terminate_router.py b/gso/workflows/router/terminate_router.py
index 5b0c6ef546c9cbb08d93c5cd277f8e31e71f57c1..f7caa6c9936154f3afff8320f5942504f9477414 100644
--- a/gso/workflows/router/terminate_router.py
+++ b/gso/workflows/router/terminate_router.py
@@ -30,7 +30,7 @@ from gso.services.netbox_client import NetboxClient
from gso.settings import load_oss_params
from gso.utils.helpers import generate_inventory_for_active_routers
from gso.utils.shared_enums import Vendor
-from gso.utils.types import TTNumber
+from gso.utils.types.tt_number import TTNumber
logger = logging.getLogger(__name__)
diff --git a/gso/workflows/router/update_ibgp_mesh.py b/gso/workflows/router/update_ibgp_mesh.py
index 8fbb2813c94cff8443647a033c146e93f34b7ca4..a506e625ae15014439c538070fb0cf074c605309 100644
--- a/gso/workflows/router/update_ibgp_mesh.py
+++ b/gso/workflows/router/update_ibgp_mesh.py
@@ -17,8 +17,9 @@ from gso.products.product_types.router import Router
from gso.services import librenms_client, lso_client
from gso.services.lso_client import lso_interaction
from gso.services.subscriptions import get_trunks_that_terminate_on_router
-from gso.utils.helpers import SNMPVersion, generate_inventory_for_active_routers
-from gso.utils.types import TTNumber
+from gso.utils.helpers import generate_inventory_for_active_routers
+from gso.utils.types.snmp import SNMPVersion
+from gso.utils.types.tt_number import TTNumber
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
diff --git a/gso/workflows/site/create_imported_site.py b/gso/workflows/site/create_imported_site.py
index 1fd9d59d7aeddbb04faac6131a0cc68533a7b1d4..8a9b3d9eaf3160e47a810a56f5c95d1421f36130 100644
--- a/gso/workflows/site/create_imported_site.py
+++ b/gso/workflows/site/create_imported_site.py
@@ -10,11 +10,13 @@ from orchestrator.workflows.steps import resync, set_status, store_process_subsc
from pydantic import ConfigDict
from gso.products import ProductName
-from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate, SiteTier
+from gso.products.product_blocks.site import SiteTier
from gso.products.product_types.site import ImportedSiteInactive
from gso.services import subscriptions
from gso.services.partners import get_partner_by_name
-from gso.utils.helpers import BaseSiteValidatorModel
+from gso.utils.types.base_site import BaseSiteValidatorModel
+from gso.utils.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
+from gso.utils.types.ip_address import IPAddress
@step("Create subscription")
@@ -51,7 +53,7 @@ def initialize_subscription(
site_longitude: LongitudeCoordinate,
site_bgp_community_id: int,
site_internal_id: int,
- site_ts_address: str,
+ site_ts_address: IPAddress,
site_tier: SiteTier,
) -> State:
"""Initialise the subscription object with all input."""
diff --git a/gso/workflows/site/create_site.py b/gso/workflows/site/create_site.py
index d2e99d510e678a7f7e654de4cbee1a162868ed2f..eb5409d778c019e41ae82c85376ec48fcf4bdcc6 100644
--- a/gso/workflows/site/create_site.py
+++ b/gso/workflows/site/create_site.py
@@ -10,10 +10,11 @@ from pydantic import ConfigDict
from pydantic_forms.validators import ReadOnlyField
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.partners import get_partner_by_name
-from gso.utils.helpers import BaseSiteValidatorModel
+from gso.utils.types.base_site import BaseSiteValidatorModel
+from gso.utils.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
+from gso.utils.types.ip_address import IPAddress
def initial_input_form_generator(product_name: str) -> FormGenerator:
@@ -50,7 +51,7 @@ def initialize_subscription(
site_longitude: LongitudeCoordinate,
site_bgp_community_id: int,
site_internal_id: int,
- site_ts_address: str,
+ site_ts_address: IPAddress,
site_tier: site_pb.SiteTier,
) -> State:
"""Initialise the subscription object with all user input."""
diff --git a/gso/workflows/site/modify_site.py b/gso/workflows/site/modify_site.py
index 0fb2b50d3b54432138b2c15addfffddc39de4e66..9c94e55032ae9712856603463b71299cf8e37fe7 100644
--- a/gso/workflows/site/modify_site.py
+++ b/gso/workflows/site/modify_site.py
@@ -1,5 +1,7 @@
"""A modification workflow for a site."""
+from typing import Annotated
+
from orchestrator.forms import FormPage
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
@@ -11,13 +13,14 @@ from orchestrator.workflows.steps import (
unsync,
)
from orchestrator.workflows.utils import wrap_modify_initial_input_form
-from pydantic import ConfigDict, field_validator
-from pydantic_core.core_schema import ValidationInfo
+from pydantic import ConfigDict
from pydantic_forms.validators import ReadOnlyField
-from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate, SiteTier
+from gso.products.product_blocks.site import SiteTier
from gso.products.product_types.site import Site
-from gso.utils.helpers import validate_ipv4_or_ipv6, validate_site_fields_is_unique
+from gso.utils.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
+from gso.utils.types.ip_address import IPAddress
+from gso.utils.types.unique_field import UniqueField
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -33,30 +36,10 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
site_country_code: ReadOnlyField(subscription.site.site_country_code, default_type=str) # type: ignore[valid-type]
site_latitude: LatitudeCoordinate = subscription.site.site_latitude
site_longitude: LongitudeCoordinate = subscription.site.site_longitude
- site_bgp_community_id: int = subscription.site.site_bgp_community_id
- site_internal_id: int = subscription.site.site_internal_id
+ site_bgp_community_id: UniqueField[int] = subscription.site.site_bgp_community_id
+ site_internal_id: UniqueField[int] = subscription.site.site_internal_id
site_tier: ReadOnlyField(subscription.site.site_tier, default_type=SiteTier) # type: ignore[valid-type]
- site_ts_address: str | None = subscription.site.site_ts_address
-
- @field_validator("site_ts_address", "site_internal_id", "site_bgp_community_id")
- def field_must_be_unique(cls, value: str | int, info: ValidationInfo) -> str | int:
- if not info.field_name:
- msg = "Field name must be provided."
- raise ValueError(msg)
-
- if value and value == getattr(subscription.site, info.field_name):
- return value
-
- validate_site_fields_is_unique(info.field_name, value)
-
- return value
-
- @field_validator("site_ts_address")
- def validate_ts_address(cls, site_ts_address: str) -> str:
- if site_ts_address and site_ts_address != subscription.site.site_ts_address:
- validate_ipv4_or_ipv6(site_ts_address)
-
- return site_ts_address
+ site_ts_address: Annotated[IPAddress, UniqueField] | None = subscription.site.site_ts_address
user_input = yield ModifySiteForm
@@ -71,7 +54,7 @@ def modify_site_subscription(
site_longitude: LongitudeCoordinate,
site_bgp_community_id: int,
site_internal_id: int,
- site_ts_address: str,
+ site_ts_address: IPAddress,
) -> State:
"""Update the subscription model in the service database."""
subscription.site.site_city = site_city
diff --git a/gso/workflows/super_pop_switch/create_imported_super_pop_switch.py b/gso/workflows/super_pop_switch/create_imported_super_pop_switch.py
index 1cdb7d09d996baae8d1c186db35bfbf3fd27f676..da99dd68249318c5f2931d6196e08c0ae6140737 100644
--- a/gso/workflows/super_pop_switch/create_imported_super_pop_switch.py
+++ b/gso/workflows/super_pop_switch/create_imported_super_pop_switch.py
@@ -14,7 +14,8 @@ from gso.services import subscriptions
from gso.services.partners import get_partner_by_name
from gso.services.subscriptions import get_site_by_name
from gso.utils.helpers import generate_fqdn
-from gso.utils.shared_enums import IPv4AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.ip_address import IPv4AddressType, PortNumber
@step("Create subscription")
diff --git a/gso/workflows/tasks/create_partners.py b/gso/workflows/tasks/create_partners.py
index b04c5c68d4f2ba7ddc350c87d047313af111e23c..5febaf4ea38452335d96541fb1449a9f1aebacff 100644
--- a/gso/workflows/tasks/create_partners.py
+++ b/gso/workflows/tasks/create_partners.py
@@ -4,9 +4,9 @@ from orchestrator.forms import FormPage
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State
from orchestrator.workflow import StepList, begin, done, step, workflow
-from pydantic import ConfigDict, EmailStr, field_validator
+from pydantic import ConfigDict
-from gso.services.partners import PartnerSchema, create_partner, filter_partners_by_email, filter_partners_by_name
+from gso.services.partners import PartnerEmail, PartnerName, PartnerSchema, create_partner
def initial_input_form_generator() -> FormGenerator:
@@ -15,25 +15,8 @@ def initial_input_form_generator() -> FormGenerator:
class CreatePartnerForm(FormPage):
model_config = ConfigDict(title="Create a Partner")
- name: str
- email: EmailStr
-
- @field_validator("name")
- def validate_name(cls, name: str) -> str:
- if filter_partners_by_name(name=name, case_sensitive=False):
- msg = "Partner with this name already exists."
- raise ValueError(msg)
-
- return name
-
- @field_validator("email")
- def validate_email(cls, email: str) -> EmailStr:
- email = email.lower()
- if filter_partners_by_email(email=email, case_sensitive=False):
- msg = "Partner with this email already exists."
- raise ValueError(msg)
-
- return email
+ name: PartnerName
+ email: PartnerEmail
initial_user_input = yield CreatePartnerForm
@@ -42,8 +25,8 @@ def initial_input_form_generator() -> FormGenerator:
@step("Save partner information to database")
def save_partner_to_database(
- name: str,
- email: EmailStr,
+ name: PartnerName,
+ email: PartnerEmail,
) -> State:
"""Save user input as a new partner in database."""
partner = create_partner(
diff --git a/gso/workflows/tasks/modify_partners.py b/gso/workflows/tasks/modify_partners.py
index 0e82521c3ee72cbc8912b5cdde136387c1948cea..46f912cc10824a83769480935011791e35739f3d 100644
--- a/gso/workflows/tasks/modify_partners.py
+++ b/gso/workflows/tasks/modify_partners.py
@@ -9,7 +9,7 @@ from pydantic_forms.types import UUIDstr
from pydantic_forms.validators import Choice
from gso.services.partners import (
- PartnerSchema,
+ ModifiedPartnerSchema,
edit_partner,
filter_partners_by_email,
filter_partners_by_name,
@@ -38,23 +38,25 @@ def initial_input_form_generator() -> FormGenerator:
class ModifyPartnerForm(FormPage):
model_config = ConfigDict(title="Modify a Partner")
- name: str = partner["name"]
- email: EmailStr = partner["email"]
+ name: str | None = partner["name"]
+ email: EmailStr | None = partner["email"]
@field_validator("name")
- def validate_name(cls, name: str) -> str:
- if partner["name"] != name and filter_partners_by_name(name=name, case_sensitive=False):
+ def validate_name(cls, name: str) -> str | None:
+ if partner["name"] == name:
+ return None
+ if filter_partners_by_name(name=name, case_sensitive=False):
msg = "Partner with this name already exists."
raise ValueError(msg)
-
return name
@field_validator("email")
- def validate_email(cls, email: str) -> EmailStr:
- if partner["email"] != email and filter_partners_by_email(email=email, case_sensitive=False):
+ def validate_email(cls, email: str) -> EmailStr | None:
+ if partner["email"] == email:
+ return None
+ if filter_partners_by_email(email=email, case_sensitive=False):
msg = "Partner with this email already exists."
raise ValueError(msg)
-
return email
user_input = yield ModifyPartnerForm
@@ -65,12 +67,12 @@ def initial_input_form_generator() -> FormGenerator:
@step("Save partner information to database")
def save_partner_to_database(
partner_id: UUIDstr,
- name: str,
- email: EmailStr,
+ name: str | None,
+ email: EmailStr | None,
) -> State:
"""Save modified partner in database."""
partner = edit_partner(
- partner_data=PartnerSchema(
+ partner_data=ModifiedPartnerSchema(
partner_id=partner_id,
name=name,
email=email,
diff --git a/requirements.txt b/requirements.txt
index b751e9c033cdc1edb98813563e8174876037a363..632c74b3af3c83c1ea442da2e4bf11530f682e7b 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,4 @@
-orchestrator-core==2.6.1
+orchestrator-core==2.7.4
requests==2.31.0
infoblox-client~=0.6.0
pycountry==23.12.11
diff --git a/setup.py b/setup.py
index 3eb6376d0268c2b5f6291676a11ac698f24314d1..e6ae8c9b6e2e7f37e6c7179ecc4462d4a09d960e 100644
--- a/setup.py
+++ b/setup.py
@@ -4,14 +4,14 @@ from setuptools import find_packages, setup
setup(
name="geant-service-orchestrator",
- version="2.12",
+ version="2.13",
author="GÉANT Orchestration and Automation Team",
author_email="goat@geant.org",
description="GÉANT Service Orchestrator",
url="https://gitlab.software.geant.org/goat/gap/geant-service-orchestrator",
packages=find_packages(),
install_requires=[
- "orchestrator-core==2.6.1",
+ "orchestrator-core==2.7.4",
"requests==2.31.0",
"infoblox-client~=0.6.0",
"pycountry==23.12.11",
diff --git a/start-worker.sh b/start-worker.sh
index 3c18dd4422ae6d60e16c78f45aa7189659dd5ca9..cccd84bff48ebc3e7392e15847e8f898c5120ffb 100755
--- a/start-worker.sh
+++ b/start-worker.sh
@@ -4,4 +4,4 @@ set -o errexit
set -o nounset
cd /app
-python -m celery -A gso.worker worker --loglevel=info
+python -m celery -A gso.worker worker --loglevel=info --concurrency=1
diff --git a/test/cli/test_imports.py b/test/cli/test_imports.py
index 8933627b82b06a35ca17e5a2a4417dc9cd9859c6..d794d0f0a63dd55f97f5ac98ec49774de574d31d 100644
--- a/test/cli/test_imports.py
+++ b/test/cli/test_imports.py
@@ -13,11 +13,12 @@ from gso.cli.imports import (
import_super_pop_switches,
)
from gso.products import Router, Site
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_blocks.router import RouterRole
from gso.products.product_blocks.site import SiteTier
from gso.utils.helpers import iso_from_ipv4
from gso.utils.shared_enums import Vendor
+from gso.utils.types.interfaces import PhysicalPortCapacity
##############
@@ -327,11 +328,11 @@ def test_import_iptrunk_non_unique_members_side_a_and_b(mock_start_process, mock
assert "Validation error: 2 validation errors for IptrunkImportModel" in captured_output
assert (
"""side_a_ae_members
- Value error, Items must be unique [type=value_error, input_value=[{'interface_name':"""
+ List must be unique [type=unique_list, input_value=[{'interface_name':"""
) in captured_output
assert (
"""side_b_ae_members
- Value error, Items must be unique [type=value_error, input_value=[{'interface_name':"""
+ List must be unique [type=unique_list, input_value=[{'interface_name':"""
) in captured_output
assert mock_start_process.call_count == 0
diff --git a/test/conftest.py b/test/conftest.py
index 0c4e36e4f52d60d8270493b9dd234ffb080ba507..2eb7e6a28089c621b620fa56d9a55b293c75b099 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -34,7 +34,7 @@ from urllib3_mock import Responses
from gso.main import init_gso_app
from gso.services.partners import PartnerSchema, create_partner
-from gso.utils.helpers import LAGMember
+from gso.utils.types.interfaces import LAGMember, LAGMemberList
from test.fixtures import ( # noqa: F401
iptrunk_side_subscription_factory,
iptrunk_subscription_factory,
@@ -107,7 +107,7 @@ class FakerProvider(BaseProvider):
def network_interface(self) -> str:
return self.generator.numerify("ge-@#/@#/@#")
- def link_members_juniper(self) -> list[LAGMember]:
+ def link_members_juniper(self) -> LAGMemberList[LAGMember]:
iface_amount = self.generator.random_int(min=2, max=5)
interface_names = [f"{prefix}{i}" for prefix in ["xe-1/0/", "ge-3/0/", "xe-2/1/"] for i in range(iface_amount)]
return [
@@ -115,7 +115,7 @@ class FakerProvider(BaseProvider):
for interface_name in interface_names
]
- def link_members_nokia(self) -> list[LAGMember]:
+ def link_members_nokia(self) -> LAGMemberList[LAGMember]:
iface_amount = self.generator.random_int(min=2, max=5)
return [
LAGMember(interface_name=f"Interface{i}", interface_description=self.generator.sentence())
diff --git a/test/fixtures.py b/test/fixtures.py
index 84481eda9c9d816e438414636cd335ecb2ef1b63..0bb920c0cfb98e7387d5c9766df77c36a503bf88 100644
--- a/test/fixtures.py
+++ b/test/fixtures.py
@@ -26,7 +26,6 @@ from gso.products.product_blocks.iptrunk import (
IptrunkInterfaceBlock,
IptrunkSideBlock,
IptrunkType,
- PhysicalPortCapacity,
)
from gso.products.product_blocks.router import RouterRole
from gso.products.product_blocks.site import SiteTier
@@ -38,6 +37,7 @@ from gso.products.product_types.site import ImportedSiteInactive, Site, SiteInac
from gso.products.product_types.super_pop_switch import ImportedSuperPopSwitchInactive, SuperPopSwitchInactive
from gso.services import subscriptions
from gso.utils.shared_enums import Vendor
+from gso.utils.types.interfaces import PhysicalPortCapacity
from test.workflows import WorkflowInstanceForTests
@@ -65,7 +65,7 @@ def site_subscription_factory(faker, geant_partner):
partner = geant_partner
description = description or "Site Subscription"
- site_name = site_name or faker.domain_word()
+ site_name = site_name or faker.site_name()
site_city = site_city or faker.city()
site_country = site_country or faker.country()
site_country_code = site_country_code or faker.country_code()
diff --git a/test/schemas/test_types.py b/test/schemas/test_types.py
index a968084f06e2674828b6a59df58484e1fd965851..d34f35d78fe148bcaf7edc4753d27620a4a73253 100644
--- a/test/schemas/test_types.py
+++ b/test/schemas/test_types.py
@@ -1,7 +1,7 @@
import pytest
from pydantic import BaseModel, ValidationError
-from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate
+from gso.utils.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
class LatitudeModel(BaseModel):
diff --git a/test/services/test_librenms_client.py b/test/services/test_librenms_client.py
index e28eaef74742d59d8b443ba0fad1d323f5b26254..c64b5933ba2f4b845fe34c288cd2be34c8655aca 100644
--- a/test/services/test_librenms_client.py
+++ b/test/services/test_librenms_client.py
@@ -5,7 +5,7 @@ import pytest
from requests import HTTPError
from gso.services.librenms_client import LibreNMSClient
-from gso.utils.helpers import SNMPVersion
+from gso.utils.types.snmp import SNMPVersion
@pytest.fixture()
diff --git a/test/utils/test_helpers.py b/test/utils/test_helpers.py
index dc7854eaa3287b4b98132a8243bf2068636a684a..28779cd706a2eb8f5e12caf1f5bd003ebcd8474f 100644
--- a/test/utils/test_helpers.py
+++ b/test/utils/test_helpers.py
@@ -9,9 +9,9 @@ from gso.products.product_blocks.router import RouterRole
from gso.utils.helpers import (
available_interfaces_choices_including_current_members,
generate_inventory_for_active_routers,
- validate_tt_number,
)
from gso.utils.shared_enums import Vendor
+from gso.utils.types.tt_number import validate_tt_number
@pytest.fixture()
diff --git a/test/workflows/iptrunk/test_create_imported_iptrunk.py b/test/workflows/iptrunk/test_create_imported_iptrunk.py
index c08ddbe643a122e0ed710cadffdddea70d78add3..d172dc8d7ec3d38c5e35ca6d61b7ec1cdcc0db12 100644
--- a/test/workflows/iptrunk/test_create_imported_iptrunk.py
+++ b/test/workflows/iptrunk/test_create_imported_iptrunk.py
@@ -2,8 +2,9 @@ import pytest
from orchestrator.types import SubscriptionLifecycle
from gso.products import ProductName
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_types.iptrunk import ImportedIptrunk
+from gso.utils.types.interfaces import PhysicalPortCapacity
from test.workflows import (
assert_complete,
extract_state,
diff --git a/test/workflows/iptrunk/test_create_iptrunk.py b/test/workflows/iptrunk/test_create_iptrunk.py
index 117444b7a33b5725df7558c2a3f7cd67b1cd00f1..0b7e00e638db87ddb27674d524bcbdde990d6067 100644
--- a/test/workflows/iptrunk/test_create_iptrunk.py
+++ b/test/workflows/iptrunk/test_create_iptrunk.py
@@ -5,9 +5,10 @@ import pytest
from infoblox_client.objects import HostRecord
from gso.products import Iptrunk, ProductName
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
from gso.services.subscriptions import get_product_id_by_name
from gso.utils.shared_enums import Vendor
+from gso.utils.types.interfaces import PhysicalPortCapacity
from test import USER_CONFIRM_EMPTY_FORM
from test.services.conftest import MockedNetboxClient, MockedSharePointClient
from test.workflows import (
diff --git a/test/workflows/iptrunk/test_migrate_iptrunk.py b/test/workflows/iptrunk/test_migrate_iptrunk.py
index 5a8f49b3ad3e54bf106fafbb0d4392be8c696937..083d85d9c09461099343c1b267a3724bf886fd63 100644
--- a/test/workflows/iptrunk/test_migrate_iptrunk.py
+++ b/test/workflows/iptrunk/test_migrate_iptrunk.py
@@ -3,11 +3,12 @@ from unittest.mock import patch
import pytest
-from gso.products import Iptrunk
+from gso.products.product_types.iptrunk import Iptrunk
from gso.products.product_types.router import Router
from gso.utils.shared_enums import Vendor
from test import USER_CONFIRM_EMPTY_FORM
from test.conftest import UseJuniperSide
+from test.services.conftest import MockedSharePointClient
from test.workflows import (
assert_complete,
assert_lso_interaction_success,
@@ -121,7 +122,9 @@ def interface_lists_are_equal(list1, list2):
@patch("gso.services.netbox_client.NetboxClient.allocate_interface")
@patch("gso.services.netbox_client.NetboxClient.free_interface")
@patch("gso.services.netbox_client.NetboxClient.delete_interface")
+@patch("gso.workflows.iptrunk.migrate_iptrunk.SharePointClient")
def test_migrate_iptrunk_success(
+ mock_sharepoint_client,
mocked_delete_interface,
mocked_free_interface,
mocked_allocate_interface,
@@ -146,6 +149,7 @@ def test_migrate_iptrunk_success(
mocked_create_interface.return_value = mocked_netbox.create_interface()
mocked_get_available_lags.return_value = mocked_netbox.get_available_lags()
mocked_delete_interface.return_value = mocked_netbox.delete_interface()
+ mock_sharepoint_client.return_value = MockedSharePointClient
result, process_stat, step_log = run_workflow("migrate_iptrunk", migrate_form_input)
@@ -164,6 +168,10 @@ def test_migrate_iptrunk_success(
for _ in range(1):
result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+ # Continue workflow after it has displayed a checklist URL.
+ assert_suspended(result)
+ result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
+
assert_complete(result)
state = extract_state(result)
diff --git a/test/workflows/iptrunk/test_modify_trunk_interface.py b/test/workflows/iptrunk/test_modify_trunk_interface.py
index 1383ea2edc1380613900ae62b0024b4622ac6d57..85c395e8bff7491b700cc7cda33219a25c69a9ab 100644
--- a/test/workflows/iptrunk/test_modify_trunk_interface.py
+++ b/test/workflows/iptrunk/test_modify_trunk_interface.py
@@ -3,8 +3,9 @@ from unittest.mock import patch
import pytest
from gso.products import Iptrunk
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
from gso.utils.shared_enums import Vendor
+from gso.utils.types.interfaces import LAGMemberList, PhysicalPortCapacity
from test.conftest import UseJuniperSide
from test.workflows import (
assert_complete,
@@ -172,7 +173,7 @@ def test_iptrunk_modify_trunk_interface_success(
assert subscription.iptrunk.iptrunk_minimum_links == input_form_iptrunk_data[1]["iptrunk_number_of_members"] - 1
assert subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_geant_a_sid == new_side_a_sid
- def _find_interface_by_name(interfaces: list[dict[str, str]], name: str):
+ def _find_interface_by_name(interfaces: LAGMemberList, name: str):
for interface in interfaces:
if interface.interface_name == name:
return interface