gso/workflows/router/modify_connection_strategy.py

@@ -34,8 +34,8 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
 def update_subscription_model(subscription: Router, connection_strategy: str) -> State:
     """Update the database model to reflect the new connection strategy.
 
-    If the connection strategy is set to IN-BAND, then access_via_ts should be set to False.
-    Conversely, if the connection strategy is set to OUT-OF-BAND, access_via_ts should be set to True.
+    If the connection strategy is set to in-band, then access_via_ts should be set to False.
+    Conversely, if the connection strategy is set to out-of-band, access_via_ts should be set to True.
     """
     subscription.router.router_access_via_ts = connection_strategy == ConnectionStrategy.OUT_OF_BAND
 

gso/workflows/router/promote_p_to_pe.py

@@ -236,7 +236,7 @@ def deploy_routing_instances_real(subscription: dict[str, Any], tt_number: str,
 
 @step("Remove ISIS overload")
 def remove_isis_overload(subscription: dict[str, Any], tt_number: str, process_id: UUIDstr) -> LSOState:
-    """Remove ISIS overload."""
+    """Remove :term:`ISIS` overload."""
     extra_vars = {
         "dry_run": False,
         "subscription": subscription,

gso/workflows/router/update_ibgp_mesh.py

@@ -18,7 +18,7 @@ from gso.services import librenms_client
 from gso.services.lso_client import LSOState, lso_interaction
 from gso.services.subscriptions import get_trunks_that_terminate_on_router
 from gso.utils.helpers import generate_inventory_for_active_routers
-from gso.utils.types.snmp import SNMPVersion
+from gso.utils.shared_enums import SNMPVersion
 from gso.utils.types.tt_number import TTNumber
 from gso.utils.workflow_steps import (
     add_all_p_to_pe_dry,

gso/workflows/tasks/delete_partners.py

@@ -7,23 +7,18 @@ from orchestrator.targets import Target
 from orchestrator.types import FormGenerator, State, UUIDstr
 from orchestrator.workflow import StepList, begin, done, step, workflow
 from pydantic import ConfigDict, EmailStr, field_validator
-from pydantic_forms.validators import Choice
 
-from gso.services.partners import delete_partner, get_all_partners, get_partner_by_name
+from gso.services.partners import delete_partner, get_partner_by_name
 from gso.services.subscriptions import get_subscriptions
+from gso.utils.helpers import partner_choice
 
 
 def initial_input_form_generator() -> FormGenerator:
     """Gather input from the user needed for deleting a partner."""
-    partners = {}
-    for partner in get_all_partners():
-        partners[partner["partner_id"]] = partner["name"]
-
-    partner_choice = Choice("Select a partner", zip(partners.values(), partners.items(), strict=True))  # type: ignore[arg-type]
 
     class SelectPartnerForm(FormPage):
         model_config = ConfigDict(title="Delete a Partner")
-        partners: partner_choice  # type: ignore[valid-type]
+        partners: partner_choice()  # type: ignore[valid-type]
 
         @field_validator("partners")
         def validate_partners(cls, value: Enum) -> Enum:

gso/workflows/tasks/modify_partners.py

@@ -5,30 +5,24 @@ from orchestrator.targets import Target
 from orchestrator.types import FormGenerator, State, UUIDstr
 from orchestrator.workflow import StepList, begin, done, step, workflow
 from pydantic import ConfigDict, EmailStr, field_validator
-from pydantic_forms.validators import Choice
 
 from gso.services.partners import (
     ModifiedPartnerSchema,
     edit_partner,
     filter_partners_by_email,
     filter_partners_by_name,
-    get_all_partners,
     get_partner_by_name,
 )
+from gso.utils.helpers import partner_choice
 
 
 def initial_input_form_generator() -> FormGenerator:
     """Gather input from the user needed for modifying a partner."""
-    partners = {}
-    for partner in get_all_partners():
-        partners[partner["partner_id"]] = partner["name"]
-
-    partner_choice = Choice("Select a partner", zip(partners.values(), partners.items(), strict=True))  # type: ignore[arg-type]
 
     class SelectPartnerForm(FormPage):
         model_config = ConfigDict(title="Choose a Partner")
 
-        partners: partner_choice  # type: ignore[valid-type]
+        partners: partner_choice()  # type: ignore[valid-type]
 
     initial_user_input = yield SelectPartnerForm
 

test/cli/test_imports.py

@@ -5,6 +5,7 @@ from unittest.mock import patch
 import pytest
 
 from gso.cli.imports import (
+    import_edge_port,
     import_iptrunks,
     import_office_routers,
     import_opengear,
@@ -13,6 +14,7 @@ from gso.cli.imports import (
     import_super_pop_switches,
 )
 from gso.products import Router, Site
+from gso.products.product_blocks.edge_port import EdgePortType, EncapsulationType
 from gso.products.product_blocks.iptrunk import IptrunkType
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_blocks.site import SiteTier
@@ -195,6 +197,41 @@ def opengear_data(temp_file, faker, site_subscription_factory):
     return _opengear_data
 
 
+@pytest.fixture()
+def edge_port_data(temp_file, faker, nokia_router_subscription_factory, partner_factory):
+    def _edge_port_data(**kwargs):
+        edge_port_data = {
+            "node": Router.from_subscription(nokia_router_subscription_factory()).router.router_fqdn,
+            "service_type": EdgePortType.CUSTOMER,
+            "speed": PhysicalPortCapacity.TEN_GIGABIT_PER_SECOND,
+            "encapsulation": EncapsulationType.DOT1Q,
+            "name": "lag34",
+            "minimum_links": 2,
+            "geant_ga_id": faker.geant_gid(),
+            "mac_address": faker.mac_address(),
+            "partner": partner_factory()["name"],
+            "enable_lacp": True,
+            "ignore_if_down": False,
+            "ae_members": [
+                {
+                    "interface_name": faker.network_interface(),
+                    "interface_description": faker.sentence(),
+                },
+                {
+                    "interface_name": faker.network_interface(),
+                    "interface_description": faker.sentence(),
+                },
+            ],
+            "description": faker.sentence(),
+        }
+        edge_port_data.update(**kwargs)
+
+        temp_file.write_text(json.dumps([edge_port_data]))
+        return {"path": str(temp_file), "data": edge_port_data}
+
+    return _edge_port_data
+
+
 ###########
 #  TESTS  #
 ###########
@@ -377,3 +414,35 @@ def test_import_super_pop_switch_success(mock_start_process, mock_sleep, super_p
 def test_import_opengear_success(mock_start_process, opengear_data):
     import_opengear(opengear_data()["path"])
     assert mock_start_process.call_count == 1
+
+
+@patch("gso.cli.imports.time.sleep")
+@patch("gso.cli.imports.start_process")
+def test_import_edge_port_successful(mock_start_process, mock_sleep, edge_port_data):
+    import_edge_port(edge_port_data()["path"])
+    assert mock_start_process.call_count == 1
+
+
+@patch("gso.cli.imports.time.sleep")
+@patch("gso.cli.imports.start_process")
+def test_import_edge_port_with_invalid_router(
+    mock_start_process, mock_sleep, edge_port_data, capfd, nokia_router_subscription_factory
+):
+    p_router = nokia_router_subscription_factory(router_role=RouterRole.P)
+    broken_data = edge_port_data(node=Router.from_subscription(p_router).router.router_fqdn)
+    import_edge_port(broken_data["path"])
+
+    captured_output, _ = capfd.readouterr()
+    assert f"Router {p_router} not found" in captured_output
+    assert mock_start_process.call_count == 0
+
+
+@patch("gso.cli.imports.time.sleep")
+@patch("gso.cli.imports.start_process")
+def test_import_edge_port_with_invalid_partner(mock_start_process, mock_sleep, edge_port_data, capfd):
+    broken_data = edge_port_data(partner="INVALID")
+    import_edge_port(broken_data["path"])
+
+    captured_output, _ = capfd.readouterr()
+    assert "Partner INVALID not found" in captured_output
+    assert mock_start_process.call_count == 0

test/conftest.py

@@ -36,15 +36,19 @@ from gso.main import init_gso_app
 from gso.services.partners import PartnerSchema, create_partner
 from gso.utils.types.interfaces import LAGMember, LAGMemberList
 from test.fixtures import (  # noqa: F401
+    bgp_session_subscription_factory,
+    edge_port_subscription_factory,
+    geant_ip_subscription_factory,
     iptrunk_side_subscription_factory,
     iptrunk_subscription_factory,
     juniper_router_subscription_factory,
     nokia_router_subscription_factory,
+    nren_access_port_factory,
     office_router_subscription_factory,
     opengear_subscription_factory,
+    service_binding_port_factory,
     site_subscription_factory,
     super_pop_switch_subscription_factory,
-    test_workflow,
 )
 
 logging.getLogger("faker.factory").setLevel(logging.WARNING)
@@ -104,6 +108,12 @@ class FakerProvider(BaseProvider):
 
         return site_name
 
+    def ipv4_netmask(self) -> int:
+        return self.generator.random_int(min=1, max=32)
+
+    def ipv6_netmask(self) -> int:
+        return self.generator.random_int(min=1, max=128)
+
     def network_interface(self) -> str:
         return self.generator.numerify("ge-@#/@#/@#")
 
@@ -122,6 +132,9 @@ class FakerProvider(BaseProvider):
             for i in range(iface_amount)
         ]
 
+    def vlan_id(self) -> int:
+        return self.generator.random_int(min=1, max=4095)
+
 
 @pytest.fixture(scope="session")
 def faker() -> Faker:
@@ -272,15 +285,15 @@ def test_client(fastapi_app):
 
 
 @pytest.fixture(scope="session")
-def partner_factory():
+def partner_factory(faker):
     def _create_partner(
-        name: str,
-        email: str,
+        name: str | None = None,
+        email: str | None = None,
     ) -> dict:
         return create_partner(
             PartnerSchema(
-                name=name,
-                email=email,
+                name=name or faker.company(),
+                email=email or faker.email(),
             )
         )
 

test/fixtures/__init__.py

+from test.fixtures.edge_port_fixtures import edge_port_subscription_factory
+from test.fixtures.geant_ip_fixtures import (
+    bgp_session_subscription_factory,
+    geant_ip_subscription_factory,
+    nren_access_port_factory,
+    service_binding_port_factory,
+)
+from test.fixtures.iptrunk_fixtures import iptrunk_side_subscription_factory, iptrunk_subscription_factory
+from test.fixtures.office_router_fixtures import office_router_subscription_factory
+from test.fixtures.opengear_fixtures import opengear_subscription_factory
+from test.fixtures.router_fixtures import juniper_router_subscription_factory, nokia_router_subscription_factory
+from test.fixtures.site_fixtures import site_subscription_factory
+from test.fixtures.super_pop_switch_fixtures import super_pop_switch_subscription_factory
+
+__all__ = [
+    "bgp_session_subscription_factory",
+    "edge_port_subscription_factory",
+    "geant_ip_subscription_factory",
+    "iptrunk_side_subscription_factory",
+    "iptrunk_subscription_factory",
+    "juniper_router_subscription_factory",
+    "nokia_router_subscription_factory",
+    "nren_access_port_factory",
+    "office_router_subscription_factory",
+    "opengear_subscription_factory",
+    "service_binding_port_factory",
+    "site_subscription_factory",
+    "super_pop_switch_subscription_factory",
+]

test/fixtures/common_fixtures.py

+from typing import Any
+
+from orchestrator.db import (
+    ProductTable,
+    SubscriptionInstanceTable,
+    SubscriptionInstanceValueTable,
+    SubscriptionTable,
+    db,
+)
+from orchestrator.utils.datetime import nowtz
+from pydantic_forms.types import SubscriptionMapping
+
+
+def create_subscription_for_mapping(
+    product: ProductTable, mapping: SubscriptionMapping, values: dict[str, Any], **kwargs: Any
+) -> SubscriptionTable:
+    """Create a subscription in the test coredb for the given subscription_mapping and values.
+
+    This function handles optional resource types starting with a ? in the mapping not supplied in the values array.
+
+    Args:
+        product: the ProductTable to create a sub for
+        mapping: the subscription_mapping belonging to that product
+        values: a dictionary of keys from the sub_map and their corresponding test values
+        kwargs: The rest of the arguments
+
+    Returns: The conforming subscription.
+    """
+
+    def build_instance(name, value_mapping):
+        block = product.find_block_by_name(name)
+
+        def build_value(rt, value):
+            resource_type = block.find_resource_type_by_name(rt)
+            return SubscriptionInstanceValueTable(resource_type_id=resource_type.resource_type_id, value=value)
+
+        return SubscriptionInstanceTable(
+            product_block_id=block.product_block_id,
+            values=[
+                build_value(resource_type, values[value_key]) for (resource_type, value_key) in value_mapping.items()
+            ],
+        )
+
+    # recreate the mapping: leave out the ?keys if no value supplied for them
+    mapping = {
+        name: [
+            {
+                **{k: value_map[k] for k in value_map if not value_map[k].startswith("?")},
+                **{
+                    k: value_map[k][1:]
+                    for k in value_map
+                    if value_map[k].startswith("?") and value_map[k][1:] in values
+                },
+            }
+            for value_map in mapping[name]
+        ]
+        for name in mapping
+    }
+
+    instances = [
+        build_instance(name, value_mapping)
+        for (name, value_mappings) in mapping.items()
+        for value_mapping in value_mappings
+    ]
+
+    return create_subscription(instances=instances, product=product, **kwargs)
+
+
+def create_subscription(**kwargs):
+    attrs = {
+        "description": "A subscription.",
+        "customer_id": kwargs.get("customer_id", "85938c4c-0a11-e511-80d0-005056956c1a"),
+        "start_date": nowtz(),
+        "status": "active",
+        "insync": True,
+        **kwargs,
+    }
+    o = SubscriptionTable(**attrs)
+    db.session.add(o)
+    db.session.commit()
+    return o

test/services/conftest.py

@@ -12,10 +12,22 @@ class MockedNetboxClient:
     def get_device_by_name(self):
         return self.BaseMockObject(id=1, name="test")
 
+    @staticmethod
+    def get_interface_by_name_and_device(interface_name: str, device_name: str):
+        return {
+            "name": f"{interface_name}",
+            "module": {"display": f"Module{interface_name}"},
+            "description": f"Description{interface_name}-{device_name}",
+        }
+
     @staticmethod
     def get_available_lags() -> list[str]:
         return [f"lag-{lag}" for lag in range(1, 5)]
 
+    @staticmethod
+    def get_available_services_lags() -> list[str]:
+        return [f"lag-{lag}" for lag in range(21, 50)]
+
     @staticmethod
     def get_available_interfaces():
         interfaces = []

test/services/test_librenms_client.py

@@ -5,7 +5,7 @@ import pytest
 from requests import HTTPError
 
 from gso.services.librenms_client import LibreNMSClient
-from gso.utils.types.snmp import SNMPVersion
+from gso.utils.shared_enums import SNMPVersion
 
 
 @pytest.fixture()