From 600d3c3a89df958f84293003fcdac69d7699a6fa Mon Sep 17 00:00:00 2001
From: Neda Moeini <neda.moeini@geant.org>
Date: Thu, 27 Jun 2024 13:53:15 +0200
Subject: [PATCH] Added import opengear workflows and Fixed the migrations
issue.
---
gso/cli/imports.py | 30 ++++++++-
gso/main.py | 0
...11-21_e8378fbcfbf3_add_initial_products.py | 6 +-
...vd7mfcfbs1q_update_modify_isis_workflow.py | 2 +-
...c7ac05063b_add_importedopengear_product.py | 47 +++++++++++++
...38f9e9e18_add_import_opengear_workflows.py | 45 +++++++++++++
gso/products/__init__.py | 2 +-
gso/products/product_types/opengear.py | 2 +-
gso/workflows/__init__.py | 4 ++
gso/workflows/opengear/__init__.py | 1 +
.../opengear/create_imported_opengear.py | 67 +++++++------------
gso/workflows/opengear/import_opengear.py | 24 +++----
test/cli/conftest.py | 1 +
test/cli/test_imports.py | 25 +++++++
test/conftest.py | 1 +
test/fixtures.py | 60 +++++++++++++++++
test/workflows/opengear/__init__.py | 0
.../opengear/test_create_imported_opengear.py | 33 +++++++++
.../opengear/test_import_opengear.py | 18 +++++
19 files changed, 308 insertions(+), 60 deletions(-)
mode change 100644 => 100755 gso/main.py
create mode 100644 gso/migrations/versions/2024-06-27_ccc7ac05063b_add_importedopengear_product.py
create mode 100644 gso/migrations/versions/2024-06-27_f6a38f9e9e18_add_import_opengear_workflows.py
create mode 100644 test/workflows/opengear/__init__.py
create mode 100644 test/workflows/opengear/test_create_imported_opengear.py
create mode 100644 test/workflows/opengear/test_import_opengear.py
diff --git a/gso/cli/imports.py b/gso/cli/imports.py
index c4dc1adf..15118b80 100644
--- a/gso/cli/imports.py
+++ b/gso/cli/imports.py
@@ -145,8 +145,25 @@ class IptrunkImportModel(BaseModel):
return self
+class OpenGearImportModel(BaseModel):
+ """Required fields for importing an existing :class:`gso.products.product_types.opengear`."""
+
+ partner: str
+ opengear_site: str
+ opengear_hostname: str
+ opengear_wan_address: str | None
+ opengear_wan_netmask: str | None
+ opengear_wan_gateway: str | None
+
+
T = TypeVar(
- "T", SiteImportModel, RouterImportModel, IptrunkImportModel, SuperPopSwitchImportModel, OfficeRouterImportModel
+ "T",
+ SiteImportModel,
+ RouterImportModel,
+ IptrunkImportModel,
+ SuperPopSwitchImportModel,
+ OfficeRouterImportModel,
+ OpenGearImportModel,
)
common_filepath_option = typer.Option(
@@ -263,6 +280,17 @@ def import_office_routers(filepath: str = common_filepath_option) -> None:
)
+def import_opengear(filepath: str = common_filepath_option) -> None:
+ """Import Opengear into GSO."""
+ _generic_import_product(
+ Path(filepath),
+ ProductType.IMPORTED_OPENGEAR,
+ "opengear",
+ "opengear_hostname",
+ OpenGearImportModel,
+ )
+
+
@app.command()
def import_iptrunks(filepath: str = common_filepath_option) -> None:
"""Import IP trunks into GSO."""
diff --git a/gso/main.py b/gso/main.py
old mode 100644
new mode 100755
diff --git a/gso/migrations/versions/2023-11-21_e8378fbcfbf3_add_initial_products.py b/gso/migrations/versions/2023-11-21_e8378fbcfbf3_add_initial_products.py
index d0c1a388..b91dfc67 100644
--- a/gso/migrations/versions/2023-11-21_e8378fbcfbf3_add_initial_products.py
+++ b/gso/migrations/versions/2023-11-21_e8378fbcfbf3_add_initial_products.py
@@ -10,9 +10,9 @@ from alembic import op
# revision identifiers, used by Alembic.
revision = 'e8378fbcfbf3'
-down_revision = 'da5c9f4cce1c'
-branch_labels = None
-depends_on = None
+down_revision = None
+branch_labels = ("data",)
+depends_on = "da5c9f4cce1c"
def upgrade() -> None:
diff --git a/gso/migrations/versions/2024-06-19_fvd7mfcfbs1q_update_modify_isis_workflow.py b/gso/migrations/versions/2024-06-19_fvd7mfcfbs1q_update_modify_isis_workflow.py
index f3a1ae00..96a18957 100644
--- a/gso/migrations/versions/2024-06-19_fvd7mfcfbs1q_update_modify_isis_workflow.py
+++ b/gso/migrations/versions/2024-06-19_fvd7mfcfbs1q_update_modify_isis_workflow.py
@@ -1,4 +1,4 @@
-"""Add upstream migrations as a dependency.
+""" Modify ISIS metric workflow description
Revision ID: fvd7mfcfbs1q
Revises:
diff --git a/gso/migrations/versions/2024-06-27_ccc7ac05063b_add_importedopengear_product.py b/gso/migrations/versions/2024-06-27_ccc7ac05063b_add_importedopengear_product.py
new file mode 100644
index 00000000..f6bee99f
--- /dev/null
+++ b/gso/migrations/versions/2024-06-27_ccc7ac05063b_add_importedopengear_product.py
@@ -0,0 +1,47 @@
+"""Add ImportedOpenGear product..
+
+Revision ID: ccc7ac05063b
+Revises: fvd7mfcfbs1q
+Create Date: 2024-06-27 11:07:11.122519
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = 'ccc7ac05063b'
+down_revision = 'fvd7mfcfbs1q'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+ conn = op.get_bind()
+ conn.execute(sa.text("""
+INSERT INTO products (name, description, product_type, tag, status) VALUES ('Imported Opengear', 'Imported Opengear Product', 'ImportedOpengear', 'IMPORTED_OPENGEAR', 'active') RETURNING products.product_id
+ """))
+ conn.execute(sa.text("""
+INSERT INTO product_product_blocks (product_id, product_block_id) VALUES ((SELECT products.product_id FROM products WHERE products.name IN ('Imported Opengear')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('OpengearBlock')))
+ """))
+
+
+def downgrade() -> None:
+ conn = op.get_bind()
+ conn.execute(sa.text("""
+DELETE FROM product_product_blocks WHERE product_product_blocks.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Imported Opengear')) AND product_product_blocks.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('OpengearBlock'))
+ """))
+ conn.execute(sa.text("""
+DELETE FROM processes WHERE processes.pid IN (SELECT processes_subscriptions.pid FROM processes_subscriptions WHERE processes_subscriptions.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Imported Opengear'))))
+ """))
+ conn.execute(sa.text("""
+DELETE FROM processes_subscriptions WHERE processes_subscriptions.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Imported Opengear')))
+ """))
+ conn.execute(sa.text("""
+DELETE FROM subscription_instances WHERE subscription_instances.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Imported Opengear')))
+ """))
+ conn.execute(sa.text("""
+DELETE FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Imported Opengear'))
+ """))
+ conn.execute(sa.text("""
+DELETE FROM products WHERE products.name IN ('Imported Opengear')
+ """))
diff --git a/gso/migrations/versions/2024-06-27_f6a38f9e9e18_add_import_opengear_workflows.py b/gso/migrations/versions/2024-06-27_f6a38f9e9e18_add_import_opengear_workflows.py
new file mode 100644
index 00000000..ff8a2ab3
--- /dev/null
+++ b/gso/migrations/versions/2024-06-27_f6a38f9e9e18_add_import_opengear_workflows.py
@@ -0,0 +1,45 @@
+"""Add import Opengear workflows..
+
+Revision ID: f6a38f9e9e18
+Revises: ccc7ac05063b
+Create Date: 2024-06-27 11:48:05.331149
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = 'f6a38f9e9e18'
+down_revision = 'ccc7ac05063b'
+branch_labels = None
+depends_on = None
+
+
+from orchestrator.migrations.helpers import create_workflow, delete_workflow
+
+new_workflows = [
+ {
+ "name": "create_imported_opengear",
+ "target": "CREATE",
+ "description": "Import Opengear",
+ "product_type": "ImportedOpengear"
+ },
+ {
+ "name": "import_opengear",
+ "target": "MODIFY",
+ "description": "Import Opengear",
+ "product_type": "ImportedOpengear"
+ }
+]
+
+
+def upgrade() -> None:
+ conn = op.get_bind()
+ for workflow in new_workflows:
+ create_workflow(conn, workflow)
+
+
+def downgrade() -> None:
+ conn = op.get_bind()
+ for workflow in new_workflows:
+ delete_workflow(conn, workflow["name"])
diff --git a/gso/products/__init__.py b/gso/products/__init__.py
index ea6af917..9278fbe7 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, ImportedOpengear
+from gso.products.product_types.opengear import ImportedOpengear, Opengear
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
diff --git a/gso/products/product_types/opengear.py b/gso/products/product_types/opengear.py
index dcfff5f5..816ff2cc 100644
--- a/gso/products/product_types/opengear.py
+++ b/gso/products/product_types/opengear.py
@@ -35,4 +35,4 @@ class ImportedOpengear(
):
"""An imported Opengear that is currently active."""
- opengear: OpengearBlock
\ No newline at end of file
+ opengear: OpengearBlock
diff --git a/gso/workflows/__init__.py b/gso/workflows/__init__.py
index e10c8b7f..cf2eca0f 100644
--- a/gso/workflows/__init__.py
+++ b/gso/workflows/__init__.py
@@ -59,3 +59,7 @@ LazyWorkflowInstance(
# Office router workflows
LazyWorkflowInstance("gso.workflows.office_router.import_office_router", "import_office_router")
LazyWorkflowInstance("gso.workflows.office_router.create_imported_office_router", "create_imported_office_router")
+
+# Opengear workflows
+LazyWorkflowInstance("gso.workflows.opengear.create_imported_opengear", "create_imported_opengear")
+LazyWorkflowInstance("gso.workflows.opengear.import_opengear", "import_opengear")
diff --git a/gso/workflows/opengear/__init__.py b/gso/workflows/opengear/__init__.py
index e69de29b..111b8882 100644
--- a/gso/workflows/opengear/__init__.py
+++ b/gso/workflows/opengear/__init__.py
@@ -0,0 +1 @@
+"""Workflows related to Opengear subscriptions."""
diff --git a/gso/workflows/opengear/create_imported_opengear.py b/gso/workflows/opengear/create_imported_opengear.py
index 814de99b..e88e329a 100644
--- a/gso/workflows/opengear/create_imported_opengear.py
+++ b/gso/workflows/opengear/create_imported_opengear.py
@@ -9,12 +9,10 @@ from orchestrator.workflows.steps import resync, set_status, store_process_subsc
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.products.product_types.opengear import ImportedOpengearInactive
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
+from gso.utils.shared_enums import IPv4AddressType
@step("Create subscription")
@@ -22,7 +20,7 @@ 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)
+ subscription = ImportedOpengearInactive.from_product_id(product_id, partner_id)
return {
"subscription": subscription,
@@ -33,60 +31,47 @@ def create_subscription(partner: str) -> State:
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")
+ class ImportOpengear(FormPage):
+ model_config = ConfigDict(title="Import Opengear")
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
+ opengear_site: str
+ opengear_hostname: str
+ opengear_wan_address: IPv4AddressType
+ opengear_wan_netmask: IPv4AddressType
+ opengear_wan_gateway: IPv4AddressType
- user_input = yield ImportRouter
+ user_input = yield ImportOpengear
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,
+ subscription: ImportedOpengearInactive,
+ opengear_site: str,
+ opengear_hostname: str,
+ opengear_wan_address: IPv4AddressType | None,
+ opengear_wan_netmask: IPv4AddressType | None,
+ opengear_wan_gateway: IPv4AddressType | 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
+ """Initialise the Imported Opengear subscription using input data."""
+ subscription.opengear.opengear_site = get_site_by_name(opengear_site).site
+ subscription.opengear.opengear_hostname = opengear_hostname
+ subscription.opengear.opengear_wan_address = opengear_wan_address
+ subscription.opengear.opengear_wan_netmask = opengear_wan_netmask
+ subscription.opengear.opengear_wan_gateway = opengear_wan_gateway
return {"subscription": subscription}
@workflow(
- "Import router",
+ "Import Opengear",
initial_input_form=initial_input_form_generator,
target=Target.CREATE,
)
-def create_imported_router() -> StepList:
- """Import a router without provisioning it."""
+def create_imported_opengear() -> StepList:
+ """Import an Opengear without provisioning it."""
return (
init
>> create_subscription
diff --git a/gso/workflows/opengear/import_opengear.py b/gso/workflows/opengear/import_opengear.py
index e9d9c4e4..d9bb7530 100644
--- a/gso/workflows/opengear/import_opengear.py
+++ b/gso/workflows/opengear/import_opengear.py
@@ -1,4 +1,4 @@
-"""A modification workflow for setting a new :term:`ISIS` metric for an IP trunk."""
+"""A modification workflow for migrating an ImportedOpengear to an Opengear subscription."""
from orchestrator.targets import Target
from orchestrator.types import State, UUIDstr
@@ -7,21 +7,21 @@ from orchestrator.workflows.steps import resync, store_process_subscription, uns
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.products.product_types.opengear import ImportedOpengear, Opengear
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]
+@step("Create new Opengear subscription")
+def import_opengear_subscription(subscription_id: UUIDstr) -> State:
+ """Take an ImportedOpengear subscription, and turn it into an Opengear subscription."""
+ old_opengear = ImportedOpengear.from_subscription(subscription_id)
+ product_id = get_product_id_by_name(ProductName.OPENGEAR)
+ new_subscription = Opengear.from_other_product(old_opengear, product_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
+@workflow("Import Opengear", target=Target.MODIFY, initial_input_form=wrap_modify_initial_input_form(None))
+def import_opengear() -> StepList:
+ """Modify an ImportedOpengear subscription into an Opengear subscription to complete the import."""
+ return init >> store_process_subscription(Target.MODIFY) >> unsync >> import_opengear_subscription >> resync >> done
diff --git a/test/cli/conftest.py b/test/cli/conftest.py
index e002fa13..92d726fa 100644
--- a/test/cli/conftest.py
+++ b/test/cli/conftest.py
@@ -3,6 +3,7 @@ from test.fixtures import ( # noqa: F401
iptrunk_subscription_factory,
nokia_router_subscription_factory,
office_router_subscription_factory,
+ opengear_subscription_factory,
site_subscription_factory,
super_pop_switch_subscription_factory,
)
diff --git a/test/cli/test_imports.py b/test/cli/test_imports.py
index 3cd45db7..f25c64a9 100644
--- a/test/cli/test_imports.py
+++ b/test/cli/test_imports.py
@@ -7,6 +7,7 @@ import pytest
from gso.cli.imports import (
import_iptrunks,
import_office_routers,
+ import_opengear,
import_routers,
import_sites,
import_super_pop_switches,
@@ -175,6 +176,24 @@ def office_router_data(temp_file, faker, site_subscription_factory):
return _office_router_data
+@pytest.fixture()
+def opengear_data(temp_file, faker, site_subscription_factory):
+ def _opengear_data(**kwargs):
+ opengear_data = {
+ "opengear_site": Site.from_subscription(site_subscription_factory()).site.site_name,
+ "opengear_hostname": faker.domain_name(levels=4),
+ "opengear_wan_address": str(faker.ipv4()),
+ "opengear_wan_netmask": str(faker.ipv4()),
+ "opengear_wan_gateway": str(faker.ipv4()),
+ }
+ opengear_data.update(**kwargs)
+
+ temp_file.write_text(json.dumps([opengear_data]))
+ return {"path": str(temp_file), "data": opengear_data}
+
+ return _opengear_data
+
+
###########
# TESTS #
###########
@@ -339,3 +358,9 @@ def test_import_office_router_success(mock_start_process, office_router_data):
def test_import_super_pop_switch_success(mock_start_process, super_pop_switch_data):
import_super_pop_switches(super_pop_switch_data()["path"])
assert mock_start_process.call_count == 1
+
+
+@patch("gso.cli.imports.start_process")
+def test_import_opengear_success(mock_start_process, opengear_data):
+ import_opengear(opengear_data()["path"])
+ assert mock_start_process.call_count == 1
diff --git a/test/conftest.py b/test/conftest.py
index 8fae41e7..d0391c02 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -42,6 +42,7 @@ from test.fixtures import ( # noqa: F401
juniper_router_subscription_factory,
nokia_router_subscription_factory,
office_router_subscription_factory,
+ opengear_subscription_factory,
site_subscription_factory,
super_pop_switch_subscription_factory,
test_workflow,
diff --git a/test/fixtures.py b/test/fixtures.py
index 316a43c3..df3100c9 100644
--- a/test/fixtures.py
+++ b/test/fixtures.py
@@ -25,6 +25,7 @@ from gso.products.product_blocks.router import RouterRole
from gso.products.product_blocks.site import SiteTier
from gso.products.product_types.iptrunk import ImportedIptrunkInactive, IptrunkInactive
from gso.products.product_types.office_router import ImportedOfficeRouterInactive, OfficeRouterInactive
+from gso.products.product_types.opengear import ImportedOpengearInactive, OpengearInactive
from gso.products.product_types.router import ImportedRouterInactive, Router, RouterInactive
from gso.products.product_types.site import ImportedSiteInactive, Site, SiteInactive
from gso.products.product_types.super_pop_switch import ImportedSuperPopSwitchInactive, SuperPopSwitchInactive
@@ -463,6 +464,65 @@ def super_pop_switch_subscription_factory(site_subscription_factory, faker, gean
return subscription_create
+@pytest.fixture()
+def opengear_subscription_factory(site_subscription_factory, faker, geant_partner):
+ def subscription_create(
+ description=None,
+ start_date="2023-05-24T00:00:00+00:00",
+ opengear_site=None,
+ opengear_hostname=None,
+ opengear_wan_address=None,
+ opengear_wan_netmask=None,
+ opengear_wan_gateway=None,
+ status: SubscriptionLifecycle | None = None,
+ partner: dict | None = None,
+ *,
+ is_imported: bool | None = True,
+ ) -> UUIDstr:
+ if partner is None:
+ partner = geant_partner
+
+ description = description or faker.text(max_nb_chars=30)
+ opengear_site = opengear_site or site_subscription_factory()
+ opengear_hostname = opengear_hostname or faker.domain_name(levels=4)
+ opengear_wan_address = opengear_wan_address or faker.ipv4()
+ opengear_wan_netmask = opengear_wan_netmask or faker.ipv4()
+ opengear_wan_gateway = opengear_wan_gateway or faker.ipv4()
+
+ if is_imported:
+ product_id = subscriptions.get_product_id_by_name(ProductName.OPENGEAR)
+ opengear_subscription = OpengearInactive.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
+ else:
+ product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_OPENGEAR)
+ opengear_subscription = ImportedOpengearInactive.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
+
+ opengear_subscription.opengear.opengear_site = Site.from_subscription(opengear_site).site
+ opengear_subscription.opengear.opengear_hostname = opengear_hostname
+ opengear_subscription.opengear.opengear_wan_address = opengear_wan_address
+ opengear_subscription.opengear.opengear_wan_netmask = opengear_wan_netmask
+ opengear_subscription.opengear.opengear_wan_gateway = opengear_wan_gateway
+
+ opengear_subscription = SubscriptionModel.from_other_lifecycle(
+ opengear_subscription, SubscriptionLifecycle.ACTIVE
+ )
+ opengear_subscription.description = description
+ opengear_subscription.start_date = start_date
+
+ if status:
+ opengear_subscription.status = status
+
+ opengear_subscription.save()
+ db.session.commit()
+
+ return str(opengear_subscription.subscription_id)
+
+ return subscription_create
+
+
@pytest.fixture()
def test_workflow(generic_subscription_1: UUIDstr, generic_product_type_1) -> Generator:
_, generic_product_one = generic_product_type_1
diff --git a/test/workflows/opengear/__init__.py b/test/workflows/opengear/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/test/workflows/opengear/test_create_imported_opengear.py b/test/workflows/opengear/test_create_imported_opengear.py
new file mode 100644
index 00000000..0f6d9bf1
--- /dev/null
+++ b/test/workflows/opengear/test_create_imported_opengear.py
@@ -0,0 +1,33 @@
+import pytest
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products import ImportedOpengear, ProductName
+from gso.products.product_types.site import Site
+from test.workflows import (
+ assert_complete,
+ extract_state,
+ run_workflow,
+)
+
+
+@pytest.fixture()
+def workflow_input_data(faker, site_subscription_factory):
+ return {
+ "partner": "GEANT",
+ "opengear_site": Site.from_subscription(site_subscription_factory()).site.site_name,
+ "opengear_hostname": faker.hostname(),
+ "opengear_wan_address": faker.ipv4(),
+ "opengear_wan_netmask": faker.ipv4(),
+ "opengear_wan_gateway": faker.ipv4(),
+ }
+
+
+@pytest.mark.workflow()
+def test_create_imported_opengear_success(workflow_input_data):
+ result, _, _ = run_workflow("create_imported_opengear", [workflow_input_data])
+ state = extract_state(result)
+ imported_opengear = ImportedOpengear.from_subscription(state["subscription_id"])
+
+ assert_complete(result)
+ assert imported_opengear.product.name == ProductName.IMPORTED_OPENGEAR
+ assert imported_opengear.status == SubscriptionLifecycle.ACTIVE
diff --git a/test/workflows/opengear/test_import_opengear.py b/test/workflows/opengear/test_import_opengear.py
new file mode 100644
index 00000000..37c7d171
--- /dev/null
+++ b/test/workflows/opengear/test_import_opengear.py
@@ -0,0 +1,18 @@
+import pytest
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products import ProductName
+from gso.products.product_types.opengear import Opengear
+from test.workflows import assert_complete, run_workflow
+
+
+@pytest.mark.workflow()
+def test_import_office_router_success(opengear_subscription_factory):
+ imported_opengear = opengear_subscription_factory(is_imported=False)
+ result, _, _ = run_workflow("import_opengear", [{"subscription_id": imported_opengear}])
+ subscription = Opengear.from_subscription(imported_opengear)
+
+ assert_complete(result)
+ assert subscription.product.name == ProductName.OPENGEAR
+ assert subscription.status == SubscriptionLifecycle.ACTIVE
+ assert subscription.insync
--
GitLab