diff --git a/test/fixtures.py b/test/fixtures.py index 3ce5bc71a57de51c4d9a61e508dd9d89bccab982..316a43c338f20601fa4235f0cac11ceacc2317f4 100644 --- a/test/fixtures.py +++ b/test/fixtures.py @@ -50,6 +50,7 @@ def site_subscription_factory(faker, geant_partner): site_ts_address=None, status: SubscriptionLifecycle | None = None, partner: dict | None = None, + *, is_imported: bool | None = True, ) -> UUIDstr: if partner is None: @@ -115,6 +116,7 @@ def nokia_router_subscription_factory(site_subscription_factory, faker, geant_pa router_site=None, status: SubscriptionLifecycle | None = None, partner: dict | None = None, + *, is_imported: bool | None = True, ) -> UUIDstr: if partner is None: @@ -180,6 +182,7 @@ def juniper_router_subscription_factory(site_subscription_factory, faker, geant_ router_site=None, status: SubscriptionLifecycle | None = None, partner: dict | None = None, + *, is_imported: bool | None = True, ) -> UUIDstr: if partner is None: @@ -283,6 +286,7 @@ def iptrunk_subscription_factory(iptrunk_side_subscription_factory, faker, geant iptrunk_sides=None, status: SubscriptionLifecycle | None = None, partner: dict | None = None, + *, is_imported: bool | None = True, ) -> UUIDstr: if partner is None: @@ -350,6 +354,7 @@ def office_router_subscription_factory(site_subscription_factory, faker, geant_p office_router_site=None, status: SubscriptionLifecycle | None = None, partner: dict | None = None, + *, is_imported: bool | None = True, ) -> UUIDstr: if partner is None: @@ -408,6 +413,7 @@ def super_pop_switch_subscription_factory(site_subscription_factory, faker, gean super_pop_switch_site=None, status: SubscriptionLifecycle | None = None, partner: dict | None = None, + *, is_imported: bool | None = True, ) -> UUIDstr: if partner is None: diff --git a/test/workflows/iptrunk/test_create_imported_iptrunk.py b/test/workflows/iptrunk/test_create_imported_iptrunk.py index 4623fe8902af4246ab053dae5536230ed1910725..450ae8b6b06f8ffa508561a653b75875cdaf2975 100644 --- a/test/workflows/iptrunk/test_create_imported_iptrunk.py +++ b/test/workflows/iptrunk/test_create_imported_iptrunk.py @@ -1,7 +1,7 @@ import pytest from orchestrator.types import SubscriptionLifecycle -from products import ProductName +from gso.products import ProductName from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity from gso.products.product_types.iptrunk import ImportedIptrunk from test.workflows import ( diff --git a/test/workflows/office_router/__init__.py b/test/workflows/office_router/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/test/workflows/office_router/test_create_imported_office_router.py b/test/workflows/office_router/test_create_imported_office_router.py new file mode 100644 index 0000000000000000000000000000000000000000..6e711ba881be8269e1d5985759117cbd7e74438a --- /dev/null +++ b/test/workflows/office_router/test_create_imported_office_router.py @@ -0,0 +1,31 @@ +import pytest +from orchestrator.types import SubscriptionLifecycle + +from gso.products import ProductName +from gso.products.product_types.office_router import ImportedOfficeRouter +from gso.products.product_types.site import Site +from test.workflows import ( + assert_complete, + extract_state, + run_workflow, +) + + +@pytest.mark.workflow() +def test_create_imported_office_router_success(faker, site_subscription_factory): + input_data = { + "partner": "GEANT", + "office_router_site": Site.from_subscription(site_subscription_factory()).site.site_name, + "office_router_fqdn": faker.domain_name(levels=4), + "office_router_ts_port": faker.port_number(is_user=True), + "office_router_lo_ipv4_address": faker.ipv4(), + "office_router_lo_ipv6_address": faker.ipv6(), + } + + result, _, _ = run_workflow("create_imported_office_router", [input_data]) + state = extract_state(result) + imported_office_router = ImportedOfficeRouter.from_subscription(state["subscription_id"]) + + assert_complete(result) + assert imported_office_router.product.name == ProductName.IMPORTED_OFFICE_ROUTER + assert imported_office_router.status == SubscriptionLifecycle.ACTIVE diff --git a/test/workflows/office_router/test_import_office_router.py b/test/workflows/office_router/test_import_office_router.py new file mode 100644 index 0000000000000000000000000000000000000000..c9894e98eeb5abb69843f849e04c6209daff8983 --- /dev/null +++ b/test/workflows/office_router/test_import_office_router.py @@ -0,0 +1,18 @@ +import pytest +from orchestrator.types import SubscriptionLifecycle + +from gso.products import ProductName +from gso.products.product_types.office_router import OfficeRouter +from test.workflows import assert_complete, run_workflow + + +@pytest.mark.workflow() +def test_import_office_router_success(office_router_subscription_factory): + imported_office_router = office_router_subscription_factory(is_imported=False) + result, _, _ = run_workflow("import_office_router", [{"subscription_id": imported_office_router}]) + subscription = OfficeRouter.from_subscription(imported_office_router) + + assert_complete(result) + assert subscription.product.name == ProductName.OFFICE_ROUTER + assert subscription.status == SubscriptionLifecycle.ACTIVE + assert subscription.insync diff --git a/test/workflows/site/test_create_imported_site.py b/test/workflows/site/test_create_imported_site.py index 1fcdc6f5f79baefdadd35328d54b70698991f49d..62c1a5946f599fd754abe70cc3e66d99a63c84ac 100644 --- a/test/workflows/site/test_create_imported_site.py +++ b/test/workflows/site/test_create_imported_site.py @@ -1,7 +1,7 @@ import pytest from orchestrator.types import SubscriptionLifecycle -from products.product_blocks.site import SiteTier +from gso.products.product_blocks.site import SiteTier from gso.products.product_types.site import ImportedSite from test.workflows import assert_complete, extract_state, run_workflow diff --git a/test/workflows/super_pop_switch/__init__.py b/test/workflows/super_pop_switch/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/test/workflows/super_pop_switch/test_create_imported_super_pop_switch.py b/test/workflows/super_pop_switch/test_create_imported_super_pop_switch.py new file mode 100644 index 0000000000000000000000000000000000000000..c1e283a6d7488950304dd5043905ac8192210e91 --- /dev/null +++ b/test/workflows/super_pop_switch/test_create_imported_super_pop_switch.py @@ -0,0 +1,30 @@ +import pytest +from orchestrator.types import SubscriptionLifecycle + +from gso.products import ProductName +from gso.products.product_types.site import Site +from gso.products.product_types.super_pop_switch import ImportedSuperPopSwitch +from test.workflows import ( + assert_complete, + extract_state, + run_workflow, +) + + +@pytest.mark.workflow() +def test_create_imported_office_router_success(faker, site_subscription_factory): + input_data = { + "partner": "GEANT", + "hostname": faker.domain_word(), + "super_pop_switch_site": Site.from_subscription(site_subscription_factory()).site.site_name, + "super_pop_switch_ts_port": faker.port_number(is_user=True), + "super_pop_switch_mgmt_ipv4_address": faker.ipv4(), + } + + result, _, _ = run_workflow("create_imported_super_pop_switch", [input_data]) + state = extract_state(result) + imported_super_pop_switch = ImportedSuperPopSwitch.from_subscription(state["subscription_id"]) + + assert_complete(result) + assert imported_super_pop_switch.product.name == ProductName.IMPORTED_SUPER_POP_SWITCH + assert imported_super_pop_switch.status == SubscriptionLifecycle.ACTIVE diff --git a/test/workflows/super_pop_switch/test_import_super_pop_switch.py b/test/workflows/super_pop_switch/test_import_super_pop_switch.py new file mode 100644 index 0000000000000000000000000000000000000000..4d20774c7d91470f6210da9e19d15632794bbf7c --- /dev/null +++ b/test/workflows/super_pop_switch/test_import_super_pop_switch.py @@ -0,0 +1,18 @@ +import pytest +from orchestrator.types import SubscriptionLifecycle + +from gso.products import ProductName +from gso.products.product_types.super_pop_switch import SuperPopSwitch +from test.workflows import assert_complete, run_workflow + + +@pytest.mark.workflow() +def test_import_super_pop_switch_success(super_pop_switch_subscription_factory): + imported_super_pop_switch = super_pop_switch_subscription_factory(is_imported=False) + result, _, _ = run_workflow("import_super_pop_switch", [{"subscription_id": imported_super_pop_switch}]) + subscription = SuperPopSwitch.from_subscription(imported_super_pop_switch) + + assert_complete(result) + assert subscription.product.name == ProductName.SUPER_POP_SWITCH + assert subscription.status == SubscriptionLifecycle.ACTIVE + assert subscription.insync