diff --git a/gso/products/__init__.py b/gso/products/__init__.py
index e81b36b8bd9d1777edac64aea4b4da77cba49b4c..ea6af917b9cf8e42c170412075de256db79cca18 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 a752f07e958748584b4642f62776b8e88ab1fa54..dcfff5f55c4356b812549d50be54a4dedc1751f1 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 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
diff --git a/gso/workflows/opengear/create_imported_opengear.py b/gso/workflows/opengear/create_imported_opengear.py
new file mode 100644
index 0000000000000000000000000000000000000000..814de99bf9521525228b836bebc64e347107394c
--- /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 0000000000000000000000000000000000000000..e9d9c4e47a04fc525c259217a5c16704c8e77751
--- /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