From 3adb60225f47b5f417caa5de32dd88e79a2c108e Mon Sep 17 00:00:00 2001
From: Neda Moeini <neda.moeini@geant.org>
Date: Mon, 17 Jun 2024 15:43:02 +0200
Subject: [PATCH] Added imported opengear product and workflow- Partially.
---
gso/products/__init__.py | 5 +-
gso/products/product_types/opengear.py | 14 +++
gso/workflows/opengear/__init__.py | 0
.../opengear/create_imported_opengear.py | 98 +++++++++++++++++++
gso/workflows/opengear/import_opengear.py | 27 +++++
5 files changed, 143 insertions(+), 1 deletion(-)
create mode 100644 gso/workflows/opengear/__init__.py
create mode 100644 gso/workflows/opengear/create_imported_opengear.py
create mode 100644 gso/workflows/opengear/import_opengear.py
diff --git a/gso/products/__init__.py b/gso/products/__init__.py
index e81b36b8..ea6af917 100644
--- a/gso/products/__init__.py
+++ b/gso/products/__init__.py
@@ -11,7 +11,7 @@ from pydantic_forms.types import strEnum
from gso.products.product_types.iptrunk import ImportedIptrunk, Iptrunk
from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
from gso.products.product_types.office_router import ImportedOfficeRouter, OfficeRouter
-from gso.products.product_types.opengear import Opengear
+from gso.products.product_types.opengear import Opengear, ImportedOpengear
from gso.products.product_types.pop_vlan import PopVlan
from gso.products.product_types.router import ImportedRouter, Router
from gso.products.product_types.site import ImportedSite, Site
@@ -36,6 +36,7 @@ class ProductName(strEnum):
IMPORTED_SUPER_POP_SWITCH = "Imported super PoP switch"
IMPORTED_OFFICE_ROUTER = "Imported office router"
OPENGEAR = "Opengear"
+ IMPORTED_OPENGEAR = "Imported Opengear"
class ProductType(strEnum):
@@ -55,6 +56,7 @@ class ProductType(strEnum):
IMPORTED_SUPER_POP_SWITCH = ImportedSuperPopSwitch.__name__
IMPORTED_OFFICE_ROUTER = ImportedOfficeRouter.__name__
OPENGEAR = Opengear.__name__
+ IMPORTED_OPENGEAR = Opengear.__name__
SUBSCRIPTION_MODEL_REGISTRY.update(
@@ -73,5 +75,6 @@ SUBSCRIPTION_MODEL_REGISTRY.update(
ProductName.IMPORTED_SUPER_POP_SWITCH.value: ImportedSuperPopSwitch,
ProductName.IMPORTED_OFFICE_ROUTER.value: ImportedOfficeRouter,
ProductName.OPENGEAR.value: Opengear,
+ ProductName.IMPORTED_OPENGEAR.value: ImportedOpengear,
},
)
diff --git a/gso/products/product_types/opengear.py b/gso/products/product_types/opengear.py
index a752f07e..dcfff5f5 100644
--- a/gso/products/product_types/opengear.py
+++ b/gso/products/product_types/opengear.py
@@ -22,3 +22,17 @@ class Opengear(OpengearProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
"""An Opengear that is currently active."""
opengear: OpengearBlock
+
+
+class ImportedOpengearInactive(SubscriptionModel, is_base=True):
+ """An imported, inactive Opengear."""
+
+ opengear: OpengearBlockInactive
+
+
+class ImportedOpengear(
+ ImportedOpengearInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE]
+):
+ """An imported Opengear that is currently active."""
+
+ opengear: OpengearBlock
\ No newline at end of file
diff --git a/gso/workflows/opengear/__init__.py b/gso/workflows/opengear/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/gso/workflows/opengear/create_imported_opengear.py b/gso/workflows/opengear/create_imported_opengear.py
new file mode 100644
index 00000000..814de99b
--- /dev/null
+++ b/gso/workflows/opengear/create_imported_opengear.py
@@ -0,0 +1,98 @@
+"""A creation workflow that adds an existing opengear to the service database."""
+
+from orchestrator import workflow
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
+from orchestrator.workflow import StepList, done, init, step
+from orchestrator.workflows.steps import resync, set_status, store_process_subscription
+from pydantic import ConfigDict
+
+from gso.products import ProductName
+from gso.products.product_blocks.router import RouterRole
+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
+
+
+@step("Create subscription")
+def create_subscription(partner: str) -> State:
+ """Create a new subscription object."""
+ partner_id = get_partner_by_name(partner)["partner_id"]
+ product_id = get_product_id_by_name(ProductName.IMPORTED_OPENGEAR)
+ subscription = ImportedRouterInactive.from_product_id(product_id, partner_id)
+
+ return {
+ "subscription": subscription,
+ "subscription_id": subscription.subscription_id,
+ }
+
+
+def initial_input_form_generator() -> FormGenerator:
+ """Generate a form that is filled in using information passed through the :term:`API` endpoint."""
+
+ class ImportRouter(FormPage):
+ model_config = ConfigDict(title="Import Router")
+
+ partner: str
+ router_site: str
+ hostname: str
+ ts_port: int
+ router_vendor: Vendor
+ router_role: RouterRole
+ router_lo_ipv4_address: IPv4AddressType
+ router_lo_ipv6_address: IPv6AddressType
+ router_lo_iso_address: str
+
+ user_input = yield ImportRouter
+
+ return user_input.dict()
+
+
+@step("Initialize subscription")
+def initialize_subscription(
+ subscription: ImportedRouterInactive,
+ hostname: str,
+ ts_port: PortNumber,
+ router_site: str,
+ router_role: RouterRole,
+ router_vendor: Vendor,
+ router_lo_ipv4_address: IPv4AddressType | None = None,
+ router_lo_ipv6_address: IPv6AddressType | None = None,
+ router_lo_iso_address: str | None = None,
+) -> State:
+ """Initialise the router subscription using input data."""
+ subscription.router.router_ts_port = ts_port
+ router_site_obj = get_site_by_name(router_site).site
+ subscription.router.router_site = router_site_obj
+ fqdn = generate_fqdn(hostname, router_site_obj.site_name, router_site_obj.site_country_code)
+ subscription.router.router_fqdn = fqdn
+ subscription.router.router_role = router_role
+ subscription.router.router_access_via_ts = False
+ subscription.description = f"Router {fqdn}"
+ subscription.router.router_lo_ipv4_address = router_lo_ipv4_address
+ subscription.router.router_lo_ipv6_address = router_lo_ipv6_address
+ subscription.router.router_lo_iso_address = router_lo_iso_address
+ subscription.router.vendor = router_vendor
+
+ return {"subscription": subscription}
+
+
+@workflow(
+ "Import router",
+ initial_input_form=initial_input_form_generator,
+ target=Target.CREATE,
+)
+def create_imported_router() -> StepList:
+ """Import a router without provisioning it."""
+ return (
+ init
+ >> create_subscription
+ >> store_process_subscription(Target.CREATE)
+ >> initialize_subscription
+ >> set_status(SubscriptionLifecycle.ACTIVE)
+ >> resync
+ >> done
+ )
diff --git a/gso/workflows/opengear/import_opengear.py b/gso/workflows/opengear/import_opengear.py
new file mode 100644
index 00000000..e9d9c4e4
--- /dev/null
+++ b/gso/workflows/opengear/import_opengear.py
@@ -0,0 +1,27 @@
+"""A modification workflow for setting a new :term:`ISIS` metric for an IP trunk."""
+
+from orchestrator.targets import Target
+from orchestrator.types import State, UUIDstr
+from orchestrator.workflow import StepList, done, init, step, workflow
+from orchestrator.workflows.steps import resync, store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+
+from gso.products import ProductName
+from gso.products.product_types.router import ImportedRouter, Router
+from gso.services.subscriptions import get_product_id_by_name
+
+
+@step("Create new router subscription")
+def import_router_subscription(subscription_id: UUIDstr) -> State:
+ """Take an ImportedRouter subscription, and turn it into a Router subscription."""
+ old_router = ImportedRouter.from_subscription(subscription_id)
+ new_subscription_id = get_product_id_by_name(ProductName.ROUTER)
+ new_subscription = Router.from_other_product(old_router, new_subscription_id) # type: ignore[arg-type]
+
+ return {"subscription": new_subscription}
+
+
+@workflow("Import Router", target=Target.MODIFY, initial_input_form=wrap_modify_initial_input_form(None))
+def import_router() -> StepList:
+ """Modify an ImportedRouter subscription into a Router subscription to complete the import."""
+ return init >> store_process_subscription(Target.MODIFY) >> unsync >> import_router_subscription >> resync >> done
--
GitLab