diff --git a/gso/api/v1/subscriptions.py b/gso/api/v1/subscriptions.py
index e2221779b66786429bb4db61f2c94bfee53e8371..bf4e96bfb6c0b2932d183d1c2707f2a18bf5a608 100644
--- a/gso/api/v1/subscriptions.py
+++ b/gso/api/v1/subscriptions.py
@@ -1,4 +1,5 @@
""":term:`API` endpoint for fetching different types of subscriptions."""
+
from typing import Any
from fastapi import Depends, Response, status
@@ -19,17 +20,6 @@ router = APIRouter(
)
-# class MySubscriptionDomainModelSchema(SubscriptionDomainModelSchema):
-# model_config = ConfigDict(
-# extra="allow",
-# json_encoders={
-# # datetime: lambda dt: dt.timestamp(),
-# ipaddress.IPv4Address: lambda v: 1/0,
-# ipaddress.IPv6Address: lambda v: str(v),
-# }
-# )
-
-
@router.get(
"/routers",
status_code=status.HTTP_200_OK,
diff --git a/gso/auth/oidc_policy_helper.py b/gso/auth/oidc_policy_helper.py
index 241641dcfe3a93ffd41bb2962255c6557b95d235..cb85b2c5e8d8279d80cf1f33aa1bce307c6ea491 100644
--- a/gso/auth/oidc_policy_helper.py
+++ b/gso/auth/oidc_policy_helper.py
@@ -245,7 +245,7 @@ class OIDCUser(HTTPBearer):
return
response = await async_request.get(self.openid_url + "/.well-known/openid-configuration")
- self.openid_config = OIDCConfig.parse_obj(response.json())
+ self.openid_config = OIDCConfig.model_validate(response.json())
async def userinfo(self, async_request: AsyncClient, token: str) -> OIDCUserModel:
"""Get the userinfo from the openid server.
diff --git a/gso/middlewares.py b/gso/middlewares.py
index 58106502b70a794cde29cfb714ce61101d056dbb..5ffca88ef1caf559eafd10a9a3d6700767929661 100644
--- a/gso/middlewares.py
+++ b/gso/middlewares.py
@@ -93,9 +93,9 @@ class ModifyProcessEndpointResponse(BaseHTTPMiddleware):
if callback_result and isinstance(callback_result, str):
callback_result = json.loads(callback_result)
if callback_result.get("output") and len(callback_result["output"]) > max_output_length:
- callback_result[
- "output"
- ] = f'{request.base_url}api/v1/processes/steps/{step["step_id"]}/callback-results{token}'
+ callback_result["output"] = (
+ f'{request.base_url}api/v1/processes/steps/{step["step_id"]}/callback-results{token}'
+ )
step["state"]["callback_result"] = callback_result
except (AttributeError, KeyError, TypeError):
pass
diff --git a/gso/migrations/versions/2024-04-02_1ec810b289c0_add_orchestrator_2_1_2_migrations.py b/gso/migrations/versions/2024-04-02_1ec810b289c0_add_orchestrator_2_1_2_migrations.py
index aa9593a8ba0279329a6361900a1965b2eddc365c..c9842e731cab00e90653b8ca5bb32185a3d0b025 100644
--- a/gso/migrations/versions/2024-04-02_1ec810b289c0_add_orchestrator_2_1_2_migrations.py
+++ b/gso/migrations/versions/2024-04-02_1ec810b289c0_add_orchestrator_2_1_2_migrations.py
@@ -5,6 +5,8 @@ Revises:
Create Date: 2024-04-02 10:21:08.539591
"""
+from alembic import op
+from orchestrator.migrations.helpers import create_workflow, delete_workflow
# revision identifiers, used by Alembic.
revision = '1ec810b289c0'
@@ -13,11 +15,47 @@ branch_labels = None
# TODO: check it carefuly
depends_on = '048219045729' # in this revision, SURF has added a new columns to the workflow table like delted_at, so we need to add a dependency on the revision that added the columns to the workflow table.
+new_workflows = [
+ {
+ "name": "import_site",
+ "target": "SYSTEM",
+ "description": "Import a site without provisioning it.",
+ "product_type": "Site"
+ },
+ {
+ "name": "import_router",
+ "target": "SYSTEM",
+ "description": "Import a router without provisioning it.",
+ "product_type": "Router"
+ },
+ {
+ "name": "import_iptrunk",
+ "target": "SYSTEM",
+ "description": "Import an IP trunk without provisioning it.",
+ "product_type": "Iptrunk"
+ },
+ {
+ "name": "import_super_pop_switch",
+ "target": "SYSTEM",
+ "description": "Import a Super PoP switch without provisioning it.",
+ "product_type": "SuperPopSwitch"
+ },
+ {
+ "name": "import_office_router",
+ "target": "SYSTEM",
+ "description": "Import an office router without provisioning it.",
+ "product_type": "OfficeRouter"
+ },
+]
+
def upgrade() -> None:
- pass
+ conn = op.get_bind()
+ for workflow in new_workflows:
+ create_workflow(conn, workflow)
def downgrade() -> None:
- pass
-
+ conn = op.get_bind()
+ for workflow in new_workflows:
+ delete_workflow(conn, workflow["name"])
diff --git a/gso/monkeypatches.py b/gso/monkeypatches.py
index 7c929d9f0b7e3b58a968373b6d51daab36feb36e..dd9a3f48242c227dcc4c33ecb40742817ea9f734 100644
--- a/gso/monkeypatches.py
+++ b/gso/monkeypatches.py
@@ -3,6 +3,7 @@
This adjustment is typically done to extend or modify the functionality of the original
oauth2_lib package to meet specific requirements of the gso application.
"""
+
from datetime import datetime
from ipaddress import IPv4Address, IPv6Address
diff --git a/gso/products/product_blocks/office_router.py b/gso/products/product_blocks/office_router.py
index c94bc9de79e67c2801122b4d76d7b12befbd6dbe..719f3323c7181ef31acb104a5deef8b4d122cea3 100644
--- a/gso/products/product_blocks/office_router.py
+++ b/gso/products/product_blocks/office_router.py
@@ -1,6 +1,5 @@
"""Product block for :class:`office router` products."""
-
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle
diff --git a/gso/products/product_blocks/site.py b/gso/products/product_blocks/site.py
index de39478cc8e66d4643edd8509bf4fff44ce068a6..18a049b9a01bcd84f942bb4fc5a65b1d9981eaec 100644
--- a/gso/products/product_blocks/site.py
+++ b/gso/products/product_blocks/site.py
@@ -44,7 +44,7 @@ LatitudeCoordinate: type[float] = Annotated[
example="40.7128",
description="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.",
),
- AfterValidator(validate_latitude)
+ AfterValidator(validate_latitude),
]
LongitudeCoordinate = Annotated[
float,
@@ -54,7 +54,7 @@ LongitudeCoordinate = Annotated[
example="74.0060",
description="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.",
),
- AfterValidator(validate_longitude)
+ AfterValidator(validate_longitude),
]
diff --git a/gso/products/product_blocks/super_pop_switch.py b/gso/products/product_blocks/super_pop_switch.py
index 8dc124ce59c4756931051faad4a2aa9f3318a2a0..b337b403dd6e63d29438196758e12c25cfe7893a 100644
--- a/gso/products/product_blocks/super_pop_switch.py
+++ b/gso/products/product_blocks/super_pop_switch.py
@@ -1,6 +1,5 @@
"""Product block for :class:`Super PoP Switch` products."""
-
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index c0913a181420216f8280bde7674dd4db437d9042..72b7937e7a9479b5a15beb143d4f4d4becebebfa 100644
--- a/gso/utils/helpers.py
+++ b/gso/utils/helpers.py
@@ -215,7 +215,6 @@ class BaseSiteValidatorModel(BaseModel):
site_latitude: LatitudeCoordinate
site_longitude: LongitudeCoordinate
-
@field_validator("site_ts_address")
def validate_ts_address(cls, site_ts_address: str) -> str:
"""Validate that a terminal server address is valid."""
diff --git a/gso/utils/shared_enums.py b/gso/utils/shared_enums.py
index 7fe5c1522a83a128b54081eb893086475f111487..33046f44f8fcddf48b7b1177c58c69ccd97aadc4 100644
--- a/gso/utils/shared_enums.py
+++ b/gso/utils/shared_enums.py
@@ -1,4 +1,5 @@
"""Shared choices for the different models."""
+
import ipaddress
from typing import Annotated
@@ -31,7 +32,7 @@ FancyIPV4Address = Annotated[
]
FancyIPV6Address = Annotated[
- ipaddress.IPv6Address, PlainSerializer(lambda ip: str(ip), return_type=str, when_used="always")
+ ipaddress.IPv6Address, PlainSerializer(lambda ip: str(ip), return_type=str, when_used="always")
]
@@ -40,4 +41,3 @@ class ConnectionStrategy(strEnum):
IN_BAND = "IN BAND"
OUT_OF_BAND = "OUT OF BAND"
-
diff --git a/gso/workflows/__init__.py b/gso/workflows/__init__.py
index 05848a322c6b893231a8c754954c43ecf38da943..c946483bc5592e5a5ab6072bb1fd5e838e453478 100644
--- a/gso/workflows/__init__.py
+++ b/gso/workflows/__init__.py
@@ -3,17 +3,15 @@
from orchestrator.services.subscriptions import WF_USABLE_MAP
from orchestrator.workflows import LazyWorkflowInstance
-WF_USABLE_MAP.update(
- {
- "cancel_subscription": ["initial"],
- "redeploy_base_config": ["provisioning", "active"],
- "update_ibgp_mesh": ["provisioning", "active"],
- "activate_router": ["provisioning"],
- "deploy_twamp": ["provisioning", "active"],
- "modify_trunk_interface": ["provisioning", "active"],
- "activate_iptrunk": ["provisioning"],
- }
-)
+WF_USABLE_MAP.update({
+ "cancel_subscription": ["initial"],
+ "redeploy_base_config": ["provisioning", "active"],
+ "update_ibgp_mesh": ["provisioning", "active"],
+ "activate_router": ["provisioning"],
+ "deploy_twamp": ["provisioning", "active"],
+ "modify_trunk_interface": ["provisioning", "active"],
+ "activate_iptrunk": ["provisioning"],
+})
LazyWorkflowInstance("gso.workflows.iptrunk.activate_iptrunk", "activate_iptrunk")
LazyWorkflowInstance("gso.workflows.iptrunk.create_iptrunk", "create_iptrunk")
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index 17152d26c2e03476b35391d7509189d6e343388a..bb0c8367bcb519b8b4059fe7124c6e8604be723c 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/gso/workflows/iptrunk/create_iptrunk.py
@@ -84,7 +84,9 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
user_input_router_side_a = yield SelectRouterSideA
router_a = user_input_router_side_a.side_a_node_id.name
- JuniperAeMembers = Annotated[list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=initial_user_input.iptrunk_minimum_links)]
+ JuniperAeMembers = Annotated[
+ list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=initial_user_input.iptrunk_minimum_links)
+ ]
if get_router_vendor(router_a) == Vendor.NOKIA:
@@ -94,7 +96,11 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
initial_user_input.iptrunk_speed,
)
- ae_members_side_a = Annotated[list[NokiaLAGMemberA], AfterValidator(validate_unique_list), Len(min_length=initial_user_input.iptrunk_minimum_links)]
+ ae_members_side_a = Annotated[
+ list[NokiaLAGMemberA],
+ AfterValidator(validate_unique_list),
+ Len(min_length=initial_user_input.iptrunk_minimum_links),
+ ]
else:
ae_members_side_a = JuniperAeMembers # type: ignore[assignment]
@@ -137,7 +143,13 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
initial_user_input.iptrunk_speed,
)
- ae_members_side_b = Annotated[list[NokiaLAGMemberB], AfterValidator(validate_unique_list), Len(min_length=len(user_input_side_a.side_a_ae_members), max_length= len(user_input_side_a.side_a_ae_members))]
+ ae_members_side_b = Annotated[
+ list[NokiaLAGMemberB],
+ AfterValidator(validate_unique_list),
+ Len(
+ min_length=len(user_input_side_a.side_a_ae_members), max_length=len(user_input_side_a.side_a_ae_members)
+ ),
+ ]
else:
ae_members_side_b = JuniperAeMembers # type: ignore[assignment]
diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index cfb5cf2b0c2d755f1e7082b53077ca4dcef486ff..df7b0bc6f53a6237d0c7da70587cb5c5653fe21d 100644
--- a/gso/workflows/iptrunk/migrate_iptrunk.py
+++ b/gso/workflows/iptrunk/migrate_iptrunk.py
@@ -118,10 +118,23 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
subscription.iptrunk.iptrunk_speed,
)
- ae_members = Annotated[list[NokiaLAGMember], AfterValidator(validate_unique_list), 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))]
+ ae_members = Annotated[
+ list[NokiaLAGMember],
+ AfterValidator(validate_unique_list),
+ 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),
+ ),
+ ]
else:
-
- ae_members = Annotated[list[LAGMember], AfterValidator(validate_unique_list), 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))]
+ ae_members = Annotated[
+ list[LAGMember],
+ AfterValidator(validate_unique_list),
+ 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),
+ ),
+ ]
replace_index = (
0
@@ -133,7 +146,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
LAGMember(
interface_name=iface.interface_name,
interface_description=iface.interface_description,
- )
+ )
for iface in subscription.iptrunk.iptrunk_sides[replace_index].iptrunk_side_ae_members
]
diff --git a/gso/workflows/iptrunk/modify_trunk_interface.py b/gso/workflows/iptrunk/modify_trunk_interface.py
index d61b19fbf70d6a3f18a4946ff6fc3bd15424e1f9..5a7874ee599629492d6fb1901cdeeeb6ee32e66c 100644
--- a/gso/workflows/iptrunk/modify_trunk_interface.py
+++ b/gso/workflows/iptrunk/modify_trunk_interface.py
@@ -60,10 +60,13 @@ def initialize_ae_members(subscription: Iptrunk, initial_user_input: dict, side_
)
)
- ae_members = Annotated[list[NokiaLAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_minimum_link)]
+ ae_members = Annotated[
+ list[NokiaLAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_minimum_link)
+ ]
else:
-
- ae_members = Annotated[list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_minimum_link)]
+ ae_members = Annotated[
+ list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_minimum_link)
+ ]
return ae_members # type: ignore[return-value]
@@ -83,8 +86,12 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
iptrunk_speed: PhysicalPortCapacity = subscription.iptrunk.iptrunk_speed
iptrunk_minimum_links: int = subscription.iptrunk.iptrunk_minimum_links
iptrunk_isis_metric: ReadOnlyField(subscription.iptrunk.iptrunk_isis_metric, default_type=int)
- iptrunk_ipv4_network: ReadOnlyField(str(subscription.iptrunk.iptrunk_ipv4_network), default_type=FancyIPV4Address)
- iptrunk_ipv6_network: ReadOnlyField(str(subscription.iptrunk.iptrunk_ipv6_network), default_type=FancyIPV6Address)
+ iptrunk_ipv4_network: ReadOnlyField(
+ str(subscription.iptrunk.iptrunk_ipv4_network), default_type=FancyIPV4Address
+ )
+ iptrunk_ipv6_network: ReadOnlyField(
+ str(subscription.iptrunk.iptrunk_ipv6_network), default_type=FancyIPV6Address
+ )
@field_validator("tt_number")
def validate_tt_number(cls, tt_number: str) -> str:
@@ -96,7 +103,9 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
class ModifyIptrunkSideAForm(FormPage):
model_config = ConfigDict(title="Provide subscription details for side A of the trunk.")
- side_a_node: ReadOnlyField(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn, default_type=str)
+ side_a_node: ReadOnlyField(
+ subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn, default_type=str
+ )
side_a_ae_iface: ReadOnlyField(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_iface, default_type=str)
side_a_ae_geant_a_sid: str = subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_geant_a_sid
side_a_ae_members: ae_members_side_a = ( # type: ignore[valid-type]
@@ -120,7 +129,9 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
class ModifyIptrunkSideBForm(FormPage):
model_config = ConfigDict(title="Provide subscription details for side B of the trunk.")
- side_b_node: ReadOnlyField(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn, default_type=str)
+ side_b_node: ReadOnlyField(
+ subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn, default_type=str
+ )
side_b_ae_iface: ReadOnlyField(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_iface, default_type=str)
side_b_ae_geant_a_sid: str = subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_geant_a_sid
side_b_ae_members: ae_members_side_b = ( # type: ignore[valid-type]
diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index fe5260cdec75f74b2bb6afbd9786ebc9492968a7..5117ce8b8c050699ae3a3b92e424525002472dba 100644
--- a/gso/workflows/router/create_router.py
+++ b/gso/workflows/router/create_router.py
@@ -52,7 +52,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
@model_validator(mode="after")
def hostname_must_be_available(self) -> Self:
router_site = self.router_site
- if not router_site: # TODO Test on UI
+ if not router_site: # TODO Test on UI
msg = "Please select a site before setting the hostname."
raise ValueError(msg)
diff --git a/gso/workflows/router/update_ibgp_mesh.py b/gso/workflows/router/update_ibgp_mesh.py
index 9149c026ccb1c0e9afd4178f0691af904af27a2d..cf720272bba6c8395f194da5529222c8bacd5c32 100644
--- a/gso/workflows/router/update_ibgp_mesh.py
+++ b/gso/workflows/router/update_ibgp_mesh.py
@@ -34,7 +34,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
tt_number: str
@model_validator(mode="before")
- def router_has_a_trunk(self ) -> Self:
+ def router_has_a_trunk(self) -> Self:
terminating_trunks = get_trunks_that_terminate_on_router(
subscription_id, SubscriptionLifecycle.PROVISIONING
) + get_trunks_that_terminate_on_router(subscription_id, SubscriptionLifecycle.ACTIVE)
diff --git a/gso/workflows/site/create_site.py b/gso/workflows/site/create_site.py
index 42e4fc0cf5639b02a5455f0e8f4574d70ab1e2af..48dbc9f42ed522a9702604ad3a2759e4fff1b935 100644
--- a/gso/workflows/site/create_site.py
+++ b/gso/workflows/site/create_site.py
@@ -23,7 +23,6 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
model_config = ConfigDict(title=product_name)
partner: ReadOnlyField("GEANT", default_type=str)
-
user_input = yield CreateSiteForm
return user_input.dict()
diff --git a/gso/workflows/tasks/import_iptrunk.py b/gso/workflows/tasks/import_iptrunk.py
index 330215bca0cd5664b0668584f42a0fa671f86a07..bea224e83588eb4050d51a0818cf51355b58a6f8 100644
--- a/gso/workflows/tasks/import_iptrunk.py
+++ b/gso/workflows/tasks/import_iptrunk.py
@@ -146,7 +146,7 @@ def update_ipam_stub_for_subscription(
@workflow(
"Import iptrunk",
initial_input_form=initial_input_form_generator,
- target=Target.CREATE,
+ target=Target.SYSTEM,
)
def import_iptrunk() -> StepList:
"""Import an IP trunk without provisioning it."""
diff --git a/gso/workflows/tasks/import_office_router.py b/gso/workflows/tasks/import_office_router.py
index 2abb7ed57c4f6e972cb66c487d37844c79945160..36012c15b59fe33f8549c0fc41b6cba5b495bfef 100644
--- a/gso/workflows/tasks/import_office_router.py
+++ b/gso/workflows/tasks/import_office_router.py
@@ -1,6 +1,5 @@
"""A creation workflow that adds existing office routers to the coreDB."""
-
from orchestrator import workflow
from orchestrator.forms import FormPage
from orchestrator.targets import Target
diff --git a/gso/workflows/tasks/import_router.py b/gso/workflows/tasks/import_router.py
index 1a8e67ee06fc19c58d45391b207f404cee9ccfa8..cf265d67dd2af146e84dd8d6f3552fc09df9f63e 100644
--- a/gso/workflows/tasks/import_router.py
+++ b/gso/workflows/tasks/import_router.py
@@ -1,6 +1,5 @@
"""A creation workflow that adds an existing router to the service database."""
-
from orchestrator import workflow
from orchestrator.forms import FormPage
from orchestrator.targets import Target
@@ -89,7 +88,7 @@ def initialize_subscription(
@workflow(
"Import router",
initial_input_form=initial_input_form_generator,
- target=Target.CREATE,
+ target=Target.SYSTEM,
)
def import_router() -> StepList:
"""Import a router without provisioning it."""
diff --git a/gso/workflows/tasks/import_site.py b/gso/workflows/tasks/import_site.py
index b2fa07a9a80efef5bdc7e8c2775cab4350760d26..f4d256279bb4e3d99332fe372e3c71d900aab23a 100644
--- a/gso/workflows/tasks/import_site.py
+++ b/gso/workflows/tasks/import_site.py
@@ -57,11 +57,11 @@ def generate_initial_input_form() -> FormGenerator:
@workflow(
"Import Site",
- target=Target.CREATE,
+ target=Target.SYSTEM,
initial_input_form=generate_initial_input_form,
)
def import_site() -> StepList:
- """Workflow to import a site without provisioning it."""
+ """Import a site without provisioning it."""
return (
init
>> create_subscription
diff --git a/gso/workflows/tasks/import_super_pop_switch.py b/gso/workflows/tasks/import_super_pop_switch.py
index 58e9ac1af903755aef7bd42b8edbc3a5e9843fb6..3f8750fc0f0c82f9b4e3d3d3971177ef9869c761 100644
--- a/gso/workflows/tasks/import_super_pop_switch.py
+++ b/gso/workflows/tasks/import_super_pop_switch.py
@@ -1,6 +1,5 @@
"""A creation workflow that adds existing Super PoP switches to the coreDB."""
-
from orchestrator import workflow
from orchestrator.forms import FormPage
from orchestrator.targets import Target
diff --git a/test/api/test_imports.py b/test/api/test_imports.py
index 45cd62a869d7ff726d164e4a9a2368090fcd7e55..f17276a19885475668805a8d3dab4999134b1afc 100644
--- a/test/api/test_imports.py
+++ b/test/api/test_imports.py
@@ -1,422 +1,389 @@
-# from unittest.mock import patch
-# from uuid import uuid4
-#
-# import pytest
-# from orchestrator.db import SubscriptionTable
-# from orchestrator.services import subscriptions
-#
-# from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
-# from gso.products.product_blocks.router import RouterRole
-# from gso.products.product_blocks.site import SiteTier
-# from gso.utils.helpers import iso_from_ipv4
-# from gso.utils.shared_enums import Vendor
-#
-# SITE_IMPORT_ENDPOINT = "/api/v1/imports/sites"
-# ROUTER_IMPORT_ENDPOINT = "/api/v1/imports/routers"
-# IPTRUNK_IMPORT_API_URL = "/api/v1/imports/iptrunks"
-# SUPER_POP_SWITCH_IMPORT_API_URL = "/api/v1/imports/super-pop-switches"
-# OFFICE_ROUTER_IMPORT_API_URL = "/api/v1/imports/office-routers"
-#
-#
-# @pytest.fixture()
-# def iptrunk_data(nokia_router_subscription_factory, faker):
-# router_side_a = nokia_router_subscription_factory()
-# router_side_b = nokia_router_subscription_factory()
-# return {
-# "partner": "GEANT",
-# "geant_s_sid": faker.geant_sid(),
-# "iptrunk_type": IptrunkType.DARK_FIBER,
-# "iptrunk_description": faker.sentence(),
-# "iptrunk_speed": PhysicalPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
-# "iptrunk_minimum_links": 5,
-# "iptrunk_isis_metric": 500,
-# "side_a_node_id": router_side_a,
-# "side_a_ae_iface": faker.network_interface(),
-# "side_a_ae_geant_a_sid": faker.geant_sid(),
-# "side_a_ae_members": [
-# {
-# "interface_name": faker.network_interface(),
-# "interface_description": faker.sentence(),
-# }
-# for _ in range(5)
-# ],
-# "side_b_node_id": router_side_b,
-# "side_b_ae_iface": faker.network_interface(),
-# "side_b_ae_geant_a_sid": faker.geant_sid(),
-# "side_b_ae_members": [
-# {
-# "interface_name": faker.network_interface(),
-# "interface_description": faker.sentence(),
-# }
-# for _ in range(5)
-# ],
-# "iptrunk_ipv4_network": str(faker.ipv4(network=True)),
-# "iptrunk_ipv6_network": str(faker.ipv6(network=True)),
-# }
-#
-#
-# @pytest.fixture()
-# def mock_routers(iptrunk_data):
-# with patch("gso.services.subscriptions.get_active_router_subscriptions") as mock_get_active_router_subscriptions:
-#
-# def _active_router_subscriptions(*args, **kwargs):
-# if kwargs["includes"] == ["subscription_id", "description"]:
-# return [
-# {
-# "subscription_id": iptrunk_data["side_a_node_id"],
-# "description": "iptrunk_sideA_node_id description",
-# },
-# {
-# "subscription_id": iptrunk_data["side_b_node_id"],
-# "description": "iptrunk_sideB_node_id description",
-# },
-# {
-# "subscription_id": str(uuid4()),
-# "description": "random description",
-# },
-# ]
-# return [
-# {"subscription_id": iptrunk_data["side_a_node_id"]},
-# {"subscription_id": iptrunk_data["side_b_node_id"]},
-# {"subscription_id": str(uuid4())},
-# ]
-#
-# mock_get_active_router_subscriptions.side_effect = _active_router_subscriptions
-# yield mock_get_active_router_subscriptions
-#
-#
-# @patch("gso.api.v1.imports._start_process")
-# def test_import_iptrunk_successful_with_mocked_process(mock_start_process, test_client, mock_routers, iptrunk_data):
-# mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
-# response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
-#
-# assert response.status_code == 201
-# assert response.json()["pid"] == "123e4567-e89b-12d3-a456-426655440000"
-#
-#
-# @pytest.fixture()
-# def site_data(faker):
-# return {
-# "site_name": faker.site_name(),
-# "site_city": faker.city(),
-# "site_country": faker.country(),
-# "site_country_code": faker.country_code(),
-# "site_latitude": float(faker.latitude()),
-# "site_longitude": float(faker.longitude()),
-# "site_bgp_community_id": faker.pyint(),
-# "site_internal_id": faker.pyint(),
-# "site_tier": SiteTier.TIER1,
-# "site_ts_address": faker.ipv4(),
-# "partner": "GEANT",
-# }
-#
-#
-# @pytest.fixture()
-# def router_data(faker, site_data):
-# mock_ipv4 = faker.ipv4()
-# return {
-# "hostname": "127.0.0.1",
-# "router_role": RouterRole.PE,
-# "router_vendor": Vendor.JUNIPER,
-# "router_site": site_data["site_name"],
-# "ts_port": 1234,
-# "partner": "GEANT",
-# "router_lo_ipv4_address": mock_ipv4,
-# "router_lo_ipv6_address": faker.ipv6(),
-# "router_lo_iso_address": iso_from_ipv4(mock_ipv4),
-# }
-#
-#
-# @pytest.fixture()
-# def super_pop_switch_data(faker, site_data):
-# mock_ipv4 = faker.ipv4()
-# return {
-# "hostname": "127.0.0.1",
-# "super_pop_switch_site": site_data["site_name"],
-# "super_pop_switch_ts_port": 1234,
-# "partner": "GEANT",
-# "super_pop_switch_mgmt_ipv4_address": mock_ipv4,
-# }
-#
-#
-# @pytest.fixture()
-# def office_router_data(faker, site_data):
-# return {
-# "office_router_fqdn": "127.0.0.1",
-# "office_router_site": site_data["site_name"],
-# "office_router_ts_port": 1234,
-# "partner": "GEANT",
-# "office_router_lo_ipv4_address": faker.ipv4(),
-# "office_router_lo_ipv6_address": faker.ipv6(),
-# }
-#
-#
-# def test_import_site_endpoint(test_client, site_data):
-# assert SubscriptionTable.query.all() == []
-# # Post data to the endpoint
-# response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
-# assert response.status_code == 201
-# assert "detail" in response.json()
-# assert "pid" in response.json()
-# subscription = subscriptions.retrieve_subscription_by_subscription_instance_value(
-# resource_type="site_name",
-# value=site_data["site_name"],
-# )
-# assert subscription is not None
-#
-#
-# def test_import_site_endpoint_with_existing_site(test_client, site_data):
-# response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
-# assert SubscriptionTable.query.count() == 1
-# assert response.status_code == 201
-#
-# response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
-# assert response.status_code == 422
-# assert SubscriptionTable.query.count() == 1
-#
-#
-# def test_import_site_endpoint_with_invalid_data(test_client, site_data):
-# # invalid data, missing site_latitude and invalid site_longitude
-# site_data.pop("site_latitude")
-# site_data["site_longitude"] = "invalid"
-# assert SubscriptionTable.query.count() == 0
-# response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
-# assert response.status_code == 422
-# assert SubscriptionTable.query.count() == 0
-# response = response.json()
-# assert response["detail"][0]["loc"] == ["body", "site_latitude"]
-# assert response["detail"][0]["msg"] == "field required"
-# assert response["detail"][1]["loc"] == ["body", "site_longitude"]
-# assert response["detail"][1]["msg"] == "value is not a valid float"
-#
-#
-# def test_import_router_endpoint(test_client, site_data, router_data):
-# # Create a site first
-# response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
-# assert response.status_code == 201
-# assert SubscriptionTable.query.count() == 1
-#
-# response = test_client.post(ROUTER_IMPORT_ENDPOINT, json=router_data)
-# assert response.status_code == 201
-# assert SubscriptionTable.query.count() == 2
-#
-#
-# def test_import_router_endpoint_with_invalid_data(test_client, site_data, router_data):
-# response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
-# assert response.status_code == 201
-# assert SubscriptionTable.query.count() == 1
-#
-# # invalid data, missing hostname and invalid router_lo_ipv6_address
-# router_data.pop("hostname")
-# router_data["router_lo_ipv6_address"] = "invalid"
-# response = test_client.post(ROUTER_IMPORT_ENDPOINT, json=router_data)
-# assert response.status_code == 422
-# assert SubscriptionTable.query.count() == 1
-# response = response.json()
-# assert response["detail"][0]["loc"] == ["body", "hostname"]
-# assert response["detail"][0]["msg"] == "field required"
-# assert response["detail"][1]["loc"] == ["body", "router_lo_ipv6_address"]
-# assert response["detail"][1]["msg"] == "value is not a valid IPv6 address"
-#
-#
-# def test_import_iptrunk_successful_with_real_process(test_client, mock_routers, iptrunk_data):
-# response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
-# assert response.status_code == 201
-#
-# response = response.json()
-# assert "detail" in response
-# assert "pid" in response
-#
-# subscription = subscriptions.retrieve_subscription_by_subscription_instance_value(
-# resource_type="geant_s_sid",
-# value=iptrunk_data["geant_s_sid"],
-# )
-# assert subscription is not None
-#
-#
-# @patch("gso.api.v1.imports._start_process")
-# def test_import_iptrunk_invalid_partner(mock_start_process, test_client, mock_routers, iptrunk_data):
-# iptrunk_data["partner"] = "not_existing_partner"
-# mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
-# response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
-#
-# assert response.status_code == 422
-# assert response.json() == {
-# "detail": [
-# {
-# "loc": ["body", "partner"],
-# "msg": "partner not_existing_partner not found",
-# "type": "value_error",
-# },
-# ],
-# }
-#
-#
-# @patch("gso.api.v1.imports._start_process")
-# def test_import_iptrunk_invalid_router_id_side_a_and_b(mock_start_process, test_client, iptrunk_data):
-# iptrunk_data["side_a_node_id"] = "NOT FOUND"
-# iptrunk_data["side_b_node_id"] = "NOT FOUND"
-#
-# mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
-# response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
-#
-# assert response.status_code == 422
-# assert response.json() == {
-# "detail": [
-# {
-# "loc": ["body", "side_a_node_id"],
-# "msg": f"Router {iptrunk_data['side_a_node_id']} not found",
-# "type": "value_error",
-# },
-# {
-# "loc": ["body", "side_b_node_id"],
-# "msg": f"Router {iptrunk_data['side_b_node_id']} not found",
-# "type": "value_error",
-# },
-# ],
-# }
-#
-#
-# @patch("gso.api.v1.imports._start_process")
-# def test_import_iptrunk_non_unique_members_side_a(mock_start_process, test_client, mock_routers, iptrunk_data, faker):
-# mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
-#
-# repeat_interface_a = {
-# "interface_name": faker.network_interface(),
-# "interface_description": faker.sentence(),
-# }
-# repeat_interface_b = {
-# "interface_name": faker.network_interface(),
-# "interface_description": faker.sentence(),
-# }
-# iptrunk_data["side_a_ae_members"] = [repeat_interface_a for _ in range(5)]
-# iptrunk_data["side_b_ae_members"] = [repeat_interface_b for _ in range(5)]
-#
-# response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
-#
-# assert response.status_code == 422
-# assert response.json() == {
-# "detail": [
-# {
-# "loc": ["body", "side_a_ae_members"],
-# "msg": "Items must be unique",
-# "type": "value_error",
-# },
-# {
-# "loc": ["body", "side_b_ae_members"],
-# "msg": "Items must be unique",
-# "type": "value_error",
-# },
-# {
-# "loc": ["body", "__root__"],
-# "msg": "Side A members should be at least 5 (iptrunk_minimum_links)",
-# "type": "value_error",
-# },
-# ],
-# }
-#
-#
-# @patch("gso.api.v1.imports._start_process")
-# def test_import_iptrunk_fails_on_side_a_member_count_mismatch(
-# mock_start_process,
-# test_client,
-# mock_routers,
-# iptrunk_data,
-# ):
-# mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
-#
-# iptrunk_data["side_a_ae_members"].remove(iptrunk_data["side_a_ae_members"][0])
-#
-# response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
-#
-# assert response.status_code == 422
-# assert response.json() == {
-# "detail": [
-# {
-# "loc": ["body", "__root__"],
-# "msg": "Side A members should be at least 5 (iptrunk_minimum_links)",
-# "type": "value_error",
-# },
-# ],
-# }
-#
-#
-# @patch("gso.api.v1.imports._start_process")
-# def test_import_iptrunk_fails_on_side_a_and_b_members_mismatch(
-# mock_start_process,
-# test_client,
-# iptrunk_data,
-# mock_routers,
-# ):
-# mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
-#
-# iptrunk_data["side_b_ae_members"].remove(iptrunk_data["side_b_ae_members"][0])
-#
-# response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
-#
-# assert response.status_code == 422
-# assert response.json() == {
-# "detail": [
-# {
-# "loc": ["body", "__root__"],
-# "msg": "Mismatch between Side A and B members",
-# "type": "value_error",
-# },
-# ],
-# }
-#
-#
-# def test_import_super_pop_switch_endpoint(test_client, site_data, super_pop_switch_data):
-# response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
-# assert response.status_code == 201
-# assert SubscriptionTable.query.count() == 1
-#
-# response = test_client.post(SUPER_POP_SWITCH_IMPORT_API_URL, json=super_pop_switch_data)
-# assert response.status_code == 201
-# assert SubscriptionTable.query.count() == 2
-#
-#
-# def test_import_super_pop_switch_endpoint_with_invalid_data(test_client, site_data, super_pop_switch_data):
-# response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
-# assert response.status_code == 201
-# assert SubscriptionTable.query.count() == 1
-#
-# # invalid data, missing hostname and invalid mgmt_ipv4_address
-# super_pop_switch_data.pop("hostname")
-# super_pop_switch_data["super_pop_switch_mgmt_ipv4_address"] = "invalid"
-# response = test_client.post(SUPER_POP_SWITCH_IMPORT_API_URL, json=super_pop_switch_data)
-# assert response.status_code == 422
-# assert SubscriptionTable.query.count() == 1
-# response = response.json()
-# assert response["detail"][0]["loc"] == ["body", "hostname"]
-# assert response["detail"][0]["msg"] == "field required"
-# assert response["detail"][1]["loc"] == ["body", "super_pop_switch_mgmt_ipv4_address"]
-# assert response["detail"][1]["msg"] == "value is not a valid IPv4 address"
-#
-#
-# def test_import_office_router_endpoint(test_client, site_data, office_router_data):
-# response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
-# assert response.status_code == 201
-# assert SubscriptionTable.query.count() == 1
-#
-# response = test_client.post(OFFICE_ROUTER_IMPORT_API_URL, json=office_router_data)
-# assert response.status_code == 201
-# assert SubscriptionTable.query.count() == 2
-#
-#
-# def test_import_office_router_endpoint_with_invalid_data(test_client, site_data, office_router_data):
-# response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
-# assert response.status_code == 201
-# assert SubscriptionTable.query.count() == 1
-#
-# # invalid data, missing FQDN and invalid lo_ipv6_address
-# office_router_data.pop("office_router_fqdn")
-# office_router_data["office_router_lo_ipv6_address"] = "invalid"
-# response = test_client.post(OFFICE_ROUTER_IMPORT_API_URL, json=office_router_data)
-# assert response.status_code == 422
-# assert SubscriptionTable.query.count() == 1
-# response = response.json()
-# assert response["detail"][0]["loc"] == ["body", "office_router_fqdn"]
-# assert response["detail"][0]["msg"] == "field required"
-# assert response["detail"][1]["loc"] == ["body", "office_router_lo_ipv6_address"]
-# assert response["detail"][1]["msg"] == "value is not a valid IPv6 address"
+from unittest.mock import patch
+from uuid import uuid4
+
+import pytest
+from orchestrator.db import SubscriptionTable
+from orchestrator.services import subscriptions
+
+from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.router import RouterRole
+from gso.products.product_blocks.site import SiteTier
+from gso.utils.helpers import iso_from_ipv4
+from gso.utils.shared_enums import Vendor
+
+SITE_IMPORT_ENDPOINT = "/api/v1/imports/sites"
+ROUTER_IMPORT_ENDPOINT = "/api/v1/imports/routers"
+IPTRUNK_IMPORT_API_URL = "/api/v1/imports/iptrunks"
+SUPER_POP_SWITCH_IMPORT_API_URL = "/api/v1/imports/super-pop-switches"
+OFFICE_ROUTER_IMPORT_API_URL = "/api/v1/imports/office-routers"
+
+
+@pytest.fixture()
+def iptrunk_data(nokia_router_subscription_factory, faker):
+ router_side_a = nokia_router_subscription_factory()
+ router_side_b = nokia_router_subscription_factory()
+ return {
+ "partner": "GEANT",
+ "geant_s_sid": faker.geant_sid(),
+ "iptrunk_type": IptrunkType.DARK_FIBER,
+ "iptrunk_description": faker.sentence(),
+ "iptrunk_speed": PhysicalPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
+ "iptrunk_minimum_links": 5,
+ "iptrunk_isis_metric": 500,
+ "side_a_node_id": router_side_a,
+ "side_a_ae_iface": faker.network_interface(),
+ "side_a_ae_geant_a_sid": faker.geant_sid(),
+ "side_a_ae_members": [
+ {
+ "interface_name": faker.network_interface(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(5)
+ ],
+ "side_b_node_id": router_side_b,
+ "side_b_ae_iface": faker.network_interface(),
+ "side_b_ae_geant_a_sid": faker.geant_sid(),
+ "side_b_ae_members": [
+ {
+ "interface_name": faker.network_interface(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(5)
+ ],
+ "iptrunk_ipv4_network": str(faker.ipv4(network=True)),
+ "iptrunk_ipv6_network": str(faker.ipv6(network=True)),
+ }
+
+
+@pytest.fixture()
+def mock_routers(iptrunk_data):
+ with patch("gso.services.subscriptions.get_active_router_subscriptions") as mock_get_active_router_subscriptions:
+ def _active_router_subscriptions(*args, **kwargs):
+ if kwargs["includes"] == ["subscription_id", "description"]:
+ return [
+ {
+ "subscription_id": iptrunk_data["side_a_node_id"],
+ "description": "iptrunk_sideA_node_id description",
+ },
+ {
+ "subscription_id": iptrunk_data["side_b_node_id"],
+ "description": "iptrunk_sideB_node_id description",
+ },
+ {
+ "subscription_id": str(uuid4()),
+ "description": "random description",
+ },
+ ]
+ return [
+ {"subscription_id": iptrunk_data["side_a_node_id"]},
+ {"subscription_id": iptrunk_data["side_b_node_id"]},
+ {"subscription_id": str(uuid4())},
+ ]
+
+ mock_get_active_router_subscriptions.side_effect = _active_router_subscriptions
+ yield mock_get_active_router_subscriptions
+
+
+@patch("gso.api.v1.imports._start_process")
+def test_import_iptrunk_successful_with_mocked_process(mock_start_process, test_client, mock_routers, iptrunk_data):
+ mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
+ response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
+
+ assert response.status_code == 201
+ assert response.json()["pid"] == "123e4567-e89b-12d3-a456-426655440000"
+
+
+@pytest.fixture()
+def site_data(faker):
+ return {
+ "site_name": faker.site_name(),
+ "site_city": faker.city(),
+ "site_country": faker.country(),
+ "site_country_code": faker.country_code(),
+ "site_latitude": float(faker.latitude()),
+ "site_longitude": float(faker.longitude()),
+ "site_bgp_community_id": faker.pyint(),
+ "site_internal_id": faker.pyint(),
+ "site_tier": SiteTier.TIER1,
+ "site_ts_address": faker.ipv4(),
+ "partner": "GEANT",
+ }
+
+
+@pytest.fixture()
+def router_data(faker, site_data):
+ mock_ipv4 = faker.ipv4()
+ return {
+ "hostname": "127.0.0.1",
+ "router_role": RouterRole.PE,
+ "router_vendor": Vendor.JUNIPER,
+ "router_site": site_data["site_name"],
+ "ts_port": 1234,
+ "partner": "GEANT",
+ "router_lo_ipv4_address": mock_ipv4,
+ "router_lo_ipv6_address": faker.ipv6(),
+ "router_lo_iso_address": iso_from_ipv4(mock_ipv4),
+ }
+
+
+@pytest.fixture()
+def super_pop_switch_data(faker, site_data):
+ mock_ipv4 = faker.ipv4()
+ return {
+ "hostname": "127.0.0.1",
+ "super_pop_switch_site": site_data["site_name"],
+ "super_pop_switch_ts_port": 1234,
+ "partner": "GEANT",
+ "super_pop_switch_mgmt_ipv4_address": mock_ipv4,
+ }
+
+
+@pytest.fixture()
+def office_router_data(faker, site_data):
+ return {
+ "office_router_fqdn": "127.0.0.1",
+ "office_router_site": site_data["site_name"],
+ "office_router_ts_port": 1234,
+ "partner": "GEANT",
+ "office_router_lo_ipv4_address": faker.ipv4(),
+ "office_router_lo_ipv6_address": faker.ipv6(),
+ }
+
+
+def test_import_site_endpoint(test_client, site_data):
+ assert SubscriptionTable.query.all() == []
+ # Post data to the endpoint
+ response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
+ assert response.status_code == 201
+ assert "detail" in response.json()
+ assert "pid" in response.json()
+ subscription = subscriptions.retrieve_subscription_by_subscription_instance_value(
+ resource_type="site_name",
+ value=site_data["site_name"],
+ )
+ assert subscription is not None
+
+
+def test_import_site_endpoint_with_existing_site(test_client, site_data):
+ response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
+ assert SubscriptionTable.query.count() == 1
+ assert response.status_code == 201
+
+ response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
+ assert response.status_code == 422
+ assert SubscriptionTable.query.count() == 1
+
+
+def test_import_site_endpoint_with_invalid_data(test_client, site_data):
+ # invalid data, missing site_latitude and invalid site_longitude
+ site_data.pop("site_latitude")
+ site_data["site_longitude"] = "invalid"
+ assert SubscriptionTable.query.count() == 0
+ response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
+ assert response.status_code == 422
+ assert SubscriptionTable.query.count() == 0
+ response = response.json()
+ assert response["detail"][0]["loc"] == ["body", "site_latitude"]
+ assert response["detail"][0]["msg"] == "Field required"
+ assert response["detail"][1]["loc"] == ["body", "site_longitude"]
+ assert response["detail"][1]["msg"] == "Input should be a valid number, unable to parse string as a number"
+
+
+def test_import_router_endpoint(test_client, site_data, router_data):
+ # Create a site first
+ response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
+ assert response.status_code == 201
+ assert SubscriptionTable.query.count() == 1
+
+ response = test_client.post(ROUTER_IMPORT_ENDPOINT, json=router_data)
+ assert response.status_code == 201
+ assert SubscriptionTable.query.count() == 2
+
+
+def test_import_router_endpoint_with_invalid_data(test_client, site_data, router_data):
+ response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
+ assert response.status_code == 201
+ assert SubscriptionTable.query.count() == 1
+
+ # invalid data, missing hostname and invalid router_lo_ipv6_address
+ router_data.pop("hostname")
+ router_data["router_lo_ipv6_address"] = "invalid"
+ response = test_client.post(ROUTER_IMPORT_ENDPOINT, json=router_data)
+ assert response.status_code == 422
+ assert SubscriptionTable.query.count() == 1
+ response = response.json()
+ assert response["detail"][0]["loc"] == ["body", "hostname"]
+ assert response["detail"][0]["msg"] == "Field required"
+ assert response["detail"][1]["loc"] == ["body", "router_lo_ipv6_address"]
+ assert response["detail"][1]["msg"] == "Input is not a valid IPv6 address"
+
+
+def test_import_iptrunk_successful_with_real_process(test_client, mock_routers, iptrunk_data):
+ response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
+ assert response.status_code == 201
+
+ response = response.json()
+ assert "detail" in response
+ assert "pid" in response
+
+ subscription = subscriptions.retrieve_subscription_by_subscription_instance_value(
+ resource_type="geant_s_sid",
+ value=iptrunk_data["geant_s_sid"],
+ )
+ assert subscription is not None
+
+
+@patch("gso.api.v1.imports._start_process")
+def test_import_iptrunk_invalid_partner(mock_start_process, test_client, mock_routers, iptrunk_data):
+ iptrunk_data["partner"] = "not_existing_partner"
+ mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
+ response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
+
+ assert response.status_code == 422
+ assert response.json() == {'detail': [{'ctx': {'error': {}},
+ 'input': 'not_existing_partner',
+ 'loc': ['body', 'partner'],
+ 'msg': 'Value error, partner not_existing_partner not found',
+ 'type': 'value_error',
+ 'url': 'https://errors.pydantic.dev/2.5/v/value_error'}]}
+
+
+@patch("gso.api.v1.imports._start_process")
+def test_import_iptrunk_invalid_router_id_side_a_and_b(mock_start_process, test_client, iptrunk_data):
+ iptrunk_data["side_a_node_id"] = "NOT FOUND"
+ iptrunk_data["side_b_node_id"] = "NOT FOUND"
+
+ mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
+ response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
+
+ assert response.status_code == 422
+ assert response.json() == {'detail': [{'ctx': {'error': {}},
+ 'input': 'NOT FOUND',
+ 'loc': ['body', 'side_a_node_id'],
+ 'msg': 'Value error, Router NOT FOUND not found',
+ 'type': 'value_error',
+ 'url': 'https://errors.pydantic.dev/2.5/v/value_error'},
+ {'ctx': {'error': {}},
+ 'input': 'NOT FOUND',
+ 'loc': ['body', 'side_b_node_id'],
+ 'msg': 'Value error, Router NOT FOUND not found',
+ 'type': 'value_error',
+ 'url': 'https://errors.pydantic.dev/2.5/v/value_error'}]}
+
+
+@patch("gso.api.v1.imports._start_process")
+def test_import_iptrunk_non_unique_members_side_a(mock_start_process, test_client, mock_routers, iptrunk_data, faker):
+ mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
+
+ repeat_interface_a = {
+ "interface_name": faker.network_interface(),
+ "interface_description": faker.sentence(),
+ }
+ repeat_interface_b = {
+ "interface_name": faker.network_interface(),
+ "interface_description": faker.sentence(),
+ }
+ iptrunk_data["side_a_ae_members"] = [repeat_interface_a for _ in range(5)]
+ iptrunk_data["side_b_ae_members"] = [repeat_interface_b for _ in range(5)]
+
+ response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
+
+ assert response.status_code == 422
+
+ for detail in response.json()['detail']:
+ assert detail['msg'] == 'Value error, Items must be unique'
+
+
+@patch("gso.api.v1.imports._start_process")
+def test_import_iptrunk_fails_on_side_a_member_count_mismatch(
+ mock_start_process,
+ test_client,
+ mock_routers,
+ iptrunk_data,
+):
+ mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
+
+ iptrunk_data["side_a_ae_members"].remove(iptrunk_data["side_a_ae_members"][0])
+
+ response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
+
+ assert response.status_code == 422
+ for detail in response.json()['detail']:
+ assert detail['msg'] == 'Value error, Side A members should be at least 5 (iptrunk_minimum_links)'
+
+ input_data = detail['input']
+ assert input_data['iptrunk_minimum_links'] == 5, "The reported minimum links should be 5"
+
+
+@patch("gso.api.v1.imports._start_process")
+def test_import_iptrunk_fails_on_side_a_and_b_members_mismatch(
+ mock_start_process,
+ test_client,
+ iptrunk_data,
+ mock_routers,
+):
+ mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
+
+ iptrunk_data["side_b_ae_members"].remove(iptrunk_data["side_b_ae_members"][0])
+
+ response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
+
+ assert response.status_code == 422
+ for detail in response.json()['detail']:
+ assert detail['msg'] == 'Value error, Mismatch between Side A and B members'
+
+
+def test_import_super_pop_switch_endpoint(test_client, site_data, super_pop_switch_data):
+ response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
+ assert response.status_code == 201
+ assert SubscriptionTable.query.count() == 1
+
+ response = test_client.post(SUPER_POP_SWITCH_IMPORT_API_URL, json=super_pop_switch_data)
+ assert response.status_code == 201
+ assert SubscriptionTable.query.count() == 2
+
+
+def test_import_super_pop_switch_endpoint_with_invalid_data(test_client, site_data, super_pop_switch_data):
+ response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
+ assert response.status_code == 201
+ assert SubscriptionTable.query.count() == 1
+
+ # invalid data, missing hostname and invalid mgmt_ipv4_address
+ super_pop_switch_data.pop("hostname")
+ super_pop_switch_data["super_pop_switch_mgmt_ipv4_address"] = "invalid"
+ response = test_client.post(SUPER_POP_SWITCH_IMPORT_API_URL, json=super_pop_switch_data)
+ assert response.status_code == 422
+ assert SubscriptionTable.query.count() == 1
+ response = response.json()
+ assert response["detail"][0]["loc"] == ["body", "hostname"]
+ assert response["detail"][0]["msg"] == "Field required"
+ assert response["detail"][1]["loc"] == ["body", "super_pop_switch_mgmt_ipv4_address"]
+ assert response["detail"][1]["msg"] == "Input is not a valid IPv4 address"
+
+
+def test_import_office_router_endpoint(test_client, site_data, office_router_data):
+ response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
+ assert response.status_code == 201
+ assert SubscriptionTable.query.count() == 1
+
+ response = test_client.post(OFFICE_ROUTER_IMPORT_API_URL, json=office_router_data)
+ assert response.status_code == 201
+ assert SubscriptionTable.query.count() == 2
+
+
+def test_import_office_router_endpoint_with_invalid_data(test_client, site_data, office_router_data):
+ response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
+ assert response.status_code == 201
+ assert SubscriptionTable.query.count() == 1
+
+ # invalid data, missing FQDN and invalid lo_ipv6_address
+ office_router_data.pop("office_router_fqdn")
+ office_router_data["office_router_lo_ipv6_address"] = "invalid"
+ response = test_client.post(OFFICE_ROUTER_IMPORT_API_URL, json=office_router_data)
+ assert response.status_code == 422
+ assert SubscriptionTable.query.count() == 1
+ response = response.json()
+ assert response["detail"][0]["loc"] == ["body", "office_router_fqdn"]
+ assert response["detail"][0]["msg"] == "Field required"
+ assert response["detail"][1]["loc"] == ["body", "office_router_lo_ipv6_address"]
+ assert response["detail"][1]["msg"] == "Input is not a valid IPv6 address"
diff --git a/test/auth/test_oidc_policy_helper.py b/test/auth/test_oidc_policy_helper.py
index b1259d5dacf9aae4f8f53d58647bb1d842319451..767e34423a5c4969d3a08fc4d5ee01f005fc0b40 100644
--- a/test/auth/test_oidc_policy_helper.py
+++ b/test/auth/test_oidc_policy_helper.py
@@ -57,7 +57,7 @@ def oidc_user(mock_openid_config):
resource_server_id="resource_server",
resource_server_secret="secret", # noqa: S106
)
- user.openid_config = OIDCConfig.parse_obj(mock_openid_config)
+ user.openid_config = OIDCConfig.model_validate(mock_openid_config)
return user