Skip to content
Snippets Groups Projects
Commit b74618ad authored by Mohammad Torkashvand's avatar Mohammad Torkashvand
Browse files

expand vrf subscription objects

parent 0b2e777d
Branches
Tags
1 merge request!357fix redeploy base config if there is a vprn
Pipeline #91905 passed
......@@ -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.
......
......@@ -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
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment