From b74618ad2f0c4767dcdfbc608948e40ded06160f Mon Sep 17 00:00:00 2001
From: Mohammad Torkashvand <mohammad.torkashvand@geant.org>
Date: Thu, 13 Feb 2025 15:22:50 +0100
Subject: [PATCH] expand vrf subscription objects
---
gso/services/subscriptions.py | 18 ++++++++++++------
test/fixtures/vrf_fixtures.py | 7 ++++++-
2 files changed, 18 insertions(+), 7 deletions(-)
diff --git a/gso/services/subscriptions.py b/gso/services/subscriptions.py
index dde43c547..102b2ddb2 100644
--- a/gso/services/subscriptions.py
+++ b/gso/services/subscriptions.py
@@ -185,7 +185,7 @@ def get_trunks_that_terminate_on_router(
)
-def get_active_l3_services_linked_to_edge_port(edge_port_id: UUIDstr) -> list[SubscriptionTable]:
+def get_active_l3_services_linked_to_edge_port(edge_port_id: UUIDstr) -> list[SubscriptionModel]:
"""Retrieve all active l3 core services that are on top of the given edge port.
Args:
@@ -194,7 +194,7 @@ def get_active_l3_services_linked_to_edge_port(edge_port_id: UUIDstr) -> list[Su
Returns:
A list of active services that are on top of the edge port.
"""
- return (
+ results = (
query_in_use_by_subscriptions(UUID(edge_port_id))
.join(ProductTable)
.filter(
@@ -206,10 +206,12 @@ def get_active_l3_services_linked_to_edge_port(edge_port_id: UUIDstr) -> list[Su
.all()
)
+ return [SubscriptionModel.from_subscription(result.subscription_id) for result in results]
+
-def get_active_l2_circuit_services_linked_to_edge_port(edge_port_id: UUIDstr) -> list[SubscriptionTable]:
+def get_active_l2_circuit_services_linked_to_edge_port(edge_port_id: UUIDstr) -> list[SubscriptionModel]:
"""Retrieve all active l2 circuit services that are on top of the given edge port."""
- return (
+ results = (
query_in_use_by_subscriptions(UUID(edge_port_id))
.join(ProductTable)
.filter(
@@ -221,8 +223,10 @@ def get_active_l2_circuit_services_linked_to_edge_port(edge_port_id: UUIDstr) ->
.all()
)
+ return [SubscriptionModel.from_subscription(result.subscription_id) for result in results]
-def get_active_vrfs_linked_to_router(router_id: UUIDstr) -> list[SubscriptionTable]:
+
+def get_active_vrfs_linked_to_router(router_id: UUIDstr) -> list[SubscriptionModel]:
"""Retrieve all active VRFs that are linked to the router.
Args:
@@ -231,7 +235,7 @@ def get_active_vrfs_linked_to_router(router_id: UUIDstr) -> list[SubscriptionTab
Returns:
A list of active VRFs that are linked to the router.
"""
- return (
+ results = (
query_in_use_by_subscriptions(UUID(router_id))
.join(ProductTable)
.filter(
@@ -243,6 +247,8 @@ def get_active_vrfs_linked_to_router(router_id: UUIDstr) -> list[SubscriptionTab
.all()
)
+ return [SubscriptionModel.from_subscription(result.subscription_id) for result in results]
+
def get_product_id_by_name(product_name: ProductName) -> UUID:
"""Retrieve the UUID of a product by its name.
diff --git a/test/fixtures/vrf_fixtures.py b/test/fixtures/vrf_fixtures.py
index 09b02ed26..32f53447c 100644
--- a/test/fixtures/vrf_fixtures.py
+++ b/test/fixtures/vrf_fixtures.py
@@ -3,7 +3,7 @@ from orchestrator.db import db
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle
-from gso.products import ProductName
+from gso.products import ProductName, Router
from gso.products.product_types.vrf import VRFInactive
from gso.services.subscriptions import get_product_id_by_name
@@ -19,9 +19,13 @@ def vrf_subscription_factory(faker, geant_partner):
route_target: str | None = None,
vrf_as_number: int | None = None,
status: SubscriptionLifecycle | None = None,
+ vrf_router_list: list[Router] | None = None,
) -> SubscriptionModel:
if partner is None:
partner = geant_partner
+
+ vrf_router_list = vrf_router_list or []
+
product_id = get_product_id_by_name(ProductName.VRF)
vrf_subscription = VRFInactive.from_product_id(product_id, customer_id=partner["partner_id"], insync=True)
vrf_subscription.vrf.vrf_name = vrf_name or faker.pystr()
@@ -32,6 +36,7 @@ def vrf_subscription_factory(faker, geant_partner):
vrf_subscription = SubscriptionModel.from_other_lifecycle(vrf_subscription, SubscriptionLifecycle.ACTIVE)
vrf_subscription.description = description or f"VRF {vrf_subscription.vrf.vrf_name}"
vrf_subscription.start_date = start_date
+ vrf_subscription.vrf.vrf_router_list = [router.router for router in vrf_router_list]
if status:
vrf_subscription.status = status
--
GitLab