Skip to content
Snippets Groups Projects

upgrade to orchestrato-core v2

Merged Mohammad Torkashvand requested to merge upgrade-to-v2 into develop
All threads resolved!
30 files
+ 302
243
Compare changes
  • Side-by-side
  • Inline
Files
30
+ 12
16
""":term:`GSO` :term:`API` endpoints that import different types of existing services."""
import ipaddress
from typing import Any
from typing import Any, Self
from uuid import UUID
from fastapi import Depends, HTTPException, status
from fastapi.routing import APIRouter
from orchestrator.services import processes
from pydantic import BaseModel, root_validator, validator
from pydantic import BaseModel, field_validator, model_validator
from gso.auth.security import opa_security_default
from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
@@ -87,7 +87,7 @@ class IptrunkImportModel(BaseModel):
for router in subscriptions.get_active_router_subscriptions(includes=["subscription_id"])
}
@validator("partner")
@field_validator("partner")
def check_if_partner_exists(cls, value: str) -> str:
"""Validate that the partner exists."""
try:
@@ -98,7 +98,7 @@ class IptrunkImportModel(BaseModel):
return value
@validator("side_a_node_id", "side_b_node_id")
@field_validator("side_a_node_id", "side_b_node_id")
def check_if_router_side_is_available(cls, value: str) -> str:
"""Both sides of the trunk must exist in :term:`GSO`."""
if value not in cls._get_active_routers():
@@ -107,7 +107,7 @@ class IptrunkImportModel(BaseModel):
return value
@validator("side_a_ae_members", "side_b_ae_members")
@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)):
@@ -116,25 +116,21 @@ class IptrunkImportModel(BaseModel):
return value
@root_validator
def check_members(cls, values: dict[str, Any]) -> dict[str, Any]:
@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."""
min_links = values["iptrunk_minimum_links"]
side_a_members = values.get("side_a_ae_members", [])
side_b_members = values.get("side_b_ae_members", [])
len_a = len(self.side_a_ae_members)
len_b = len(self.side_b_ae_members)
len_a = len(side_a_members)
len_b = len(side_b_members)
if len_a < min_links:
msg = f"Side A members should be at least {min_links} (iptrunk_minimum_links)"
if len_a < self.iptrunk_minimum_links:
msg = f"Side A members should be at least {self.iptrunk_minimum_links} (iptrunk_minimum_links)"
raise ValueError(msg)
if len_a != len_b:
msg = "Mismatch between Side A and B members"
raise ValueError(msg)
return values
return self
class SuperPopSwitchImportModel(BaseModel):
Loading