Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found
Select Git revision
  • 1048-service-config-backfilling
  • NAT-1154-import-edge-port-update
  • develop
  • feature/10GGBS-NAT-980
  • feature/NAT-1260-refactor-precheck-command
  • feature/NAT-732-ias-to-re-interconnect
  • feature/add-modify-note
  • feature/add-moodi-wf-to-router
  • feature/nat-1211-edgeport-lacp-xmit
  • fix/NAT-1009/fix-redeploy-base-config-if-there-is-a-vprn
  • fix/l3-imports
  • fix/nat-1120-sdp-validation
  • master
  • 0.1
  • 0.2
  • 0.3
  • 0.4
  • 0.5
  • 0.6
  • 0.7
  • 0.8
  • 0.9
  • 1.0
  • 1.1
  • 1.4
  • 1.5
  • 2.0
  • 2.1
  • 2.10
  • 2.11
  • 2.12
  • 2.13
  • 2.14
  • 2.15
  • 2.16
  • 2.17
  • 2.18
  • 2.19
  • 2.2
  • 2.20
  • 2.21
  • 2.22
  • 2.23
  • 2.24
  • 2.25
  • 2.26
  • 2.27
  • 2.28
  • 2.29
  • 2.3
  • 2.31
  • 2.32
  • 2.33
  • 2.34
  • 2.35
  • 2.36
  • 2.37
  • 2.38
  • 2.39
  • 2.4
  • 2.40
  • 2.41
  • 2.42
  • 2.43
  • 2.44
  • 2.45
  • 2.46
  • 2.47
  • 2.48
  • 2.5
  • 2.6
  • 2.7
  • 2.8
  • 2.9
  • 3.0
  • 3.1
  • 3.10
  • 3.11
  • 3.12
  • 3.2
  • 3.3
  • 3.4
  • 3.5
  • 3.6
  • 3.7
  • 3.8
  • 3.9
  • Lime-Seal
88 results

Target

Select target project
  • goat/gap/geant-service-orchestrator
1 result
Select Git revision
  • 1048-service-config-backfilling
  • NAT-1154-import-edge-port-update
  • develop
  • feature/10GGBS-NAT-980
  • feature/NAT-1260-refactor-precheck-command
  • feature/NAT-732-ias-to-re-interconnect
  • feature/add-modify-note
  • feature/add-moodi-wf-to-router
  • feature/nat-1211-edgeport-lacp-xmit
  • fix/NAT-1009/fix-redeploy-base-config-if-there-is-a-vprn
  • fix/l3-imports
  • fix/nat-1120-sdp-validation
  • master
  • 0.1
  • 0.2
  • 0.3
  • 0.4
  • 0.5
  • 0.6
  • 0.7
  • 0.8
  • 0.9
  • 1.0
  • 1.1
  • 1.4
  • 1.5
  • 2.0
  • 2.1
  • 2.10
  • 2.11
  • 2.12
  • 2.13
  • 2.14
  • 2.15
  • 2.16
  • 2.17
  • 2.18
  • 2.19
  • 2.2
  • 2.20
  • 2.21
  • 2.22
  • 2.23
  • 2.24
  • 2.25
  • 2.26
  • 2.27
  • 2.28
  • 2.29
  • 2.3
  • 2.31
  • 2.32
  • 2.33
  • 2.34
  • 2.35
  • 2.36
  • 2.37
  • 2.38
  • 2.39
  • 2.4
  • 2.40
  • 2.41
  • 2.42
  • 2.43
  • 2.44
  • 2.45
  • 2.46
  • 2.47
  • 2.48
  • 2.5
  • 2.6
  • 2.7
  • 2.8
  • 2.9
  • 3.0
  • 3.1
  • 3.10
  • 3.11
  • 3.12
  • 3.2
  • 3.3
  • 3.4
  • 3.5
  • 3.6
  • 3.7
  • 3.8
  • 3.9
  • Lime-Seal
88 results
Show changes
Hide whitespace changes
Inline Side-by-side
Showing
with 1041 additions and 665 deletions
gso/workflows/router/modify_connection_strategy.py
View file @ f3cdbbcc
......@@ -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
View file @ f3cdbbcc
......@@ -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
View file @ f3cdbbcc
......@@ -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
View file @ f3cdbbcc
......@@ -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
View file @ f3cdbbcc
......@@ -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
View file @ f3cdbbcc
......@@ -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
View file @ f3cdbbcc
......@@ -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.py deleted 100644 → 0
View file @ 49d3e1a6
import ipaddress
from collections.abc import Generator
from typing import Any
from uuid import uuid4
import pytest
from orchestrator import step, workflow
from orchestrator.config.assignee import Assignee
from orchestrator.db import (
ProductTable,
SubscriptionInstanceTable,
SubscriptionInstanceValueTable,
SubscriptionTable,
db,
)
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
from orchestrator.utils.datetime import nowtz
from orchestrator.workflow import done, init, inputstep
from pydantic_forms.core import FormPage
from pydantic_forms.types import FormGenerator, SubscriptionMapping
from pydantic_forms.validators import Choice
from gso.products import ProductName
from gso.products.product_blocks.iptrunk import (
IptrunkInterfaceBlock,
IptrunkSideBlock,
IptrunkType,
)
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
from gso.services import subscriptions
from gso.utils.shared_enums import Vendor
from gso.utils.types.interfaces import PhysicalPortCapacity
from test.workflows import WorkflowInstanceForTests
@pytest.fixture()
def site_subscription_factory(faker, geant_partner):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
site_name=None,
site_city=None,
site_country=None,
site_country_code=None,
site_latitude=None,
site_longitude=None,
site_bgp_community_id=None,
site_internal_id=None,
site_tier=SiteTier.TIER1,
site_ts_address=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 "Site Subscription"
site_name = site_name or faker.site_name()
site_city = site_city or faker.city()
site_country = site_country or faker.country()
site_country_code = site_country_code or faker.country_code()
site_latitude = site_latitude or str(faker.latitude())
site_longitude = site_longitude or str(faker.longitude())
site_bgp_community_id = site_bgp_community_id or faker.pyint()
site_internal_id = site_internal_id or faker.pyint()
site_ts_address = site_ts_address or faker.ipv4()
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.SITE)
site_subscription = SiteInactive.from_product_id(product_id, customer_id=partner["partner_id"], insync=True)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_SITE)
site_subscription = ImportedSiteInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
site_subscription.site.site_city = site_city
site_subscription.site.site_name = site_name
site_subscription.site.site_country = site_country
site_subscription.site.site_country_code = site_country_code
site_subscription.site.site_latitude = site_latitude
site_subscription.site.site_longitude = site_longitude
site_subscription.site.site_bgp_community_id = site_bgp_community_id
site_subscription.site.site_internal_id = site_internal_id
site_subscription.site.site_tier = site_tier
site_subscription.site.site_ts_address = site_ts_address
site_subscription = SubscriptionModel.from_other_lifecycle(site_subscription, SubscriptionLifecycle.ACTIVE)
site_subscription.description = description
site_subscription.start_date = start_date
if status:
site_subscription.status = status
site_subscription.save()
db.session.commit()
return str(site_subscription.subscription_id)
return subscription_create
@pytest.fixture()
def nokia_router_subscription_factory(site_subscription_factory, faker, geant_partner):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
router_fqdn=None,
router_ts_port=None,
router_access_via_ts=None,
router_lo_ipv4_address=None,
router_lo_ipv6_address=None,
router_lo_iso_address=None,
router_role=RouterRole.PE,
router_site=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)
router_fqdn = router_fqdn or faker.domain_name(levels=4)
router_ts_port = router_ts_port or faker.random_int(min=1, max=49151)
router_access_via_ts = router_access_via_ts or faker.boolean()
router_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
router_lo_ipv6_address = router_lo_ipv6_address or ipaddress.IPv6Address(faker.ipv6())
router_lo_iso_address = router_lo_iso_address or faker.word()
router_site = router_site or site_subscription_factory()
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.ROUTER)
router_subscription = RouterInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_ROUTER)
router_subscription = ImportedRouterInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
router_subscription.router.router_fqdn = router_fqdn
router_subscription.router.router_ts_port = router_ts_port
router_subscription.router.router_access_via_ts = router_access_via_ts
router_subscription.router.router_lo_ipv4_address = router_lo_ipv4_address
router_subscription.router.router_lo_ipv6_address = router_lo_ipv6_address
router_subscription.router.router_lo_iso_address = router_lo_iso_address
router_subscription.router.router_role = router_role
router_subscription.router.router_site = Site.from_subscription(router_site).site
router_subscription.router.vendor = Vendor.NOKIA
router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
router_subscription.insync = True
router_subscription.description = description
router_subscription.start_date = start_date
if status:
router_subscription.status = status
router_subscription.save()
db.session.commit()
return str(router_subscription.subscription_id)
return subscription_create
@pytest.fixture()
def juniper_router_subscription_factory(site_subscription_factory, faker, geant_partner):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
router_fqdn=None,
router_ts_port=None,
router_access_via_ts=None,
router_lo_ipv4_address=None,
router_lo_ipv6_address=None,
router_lo_iso_address=None,
router_role=RouterRole.PE,
router_site=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)
router_fqdn = router_fqdn or faker.domain_name(levels=4)
router_ts_port = router_ts_port or faker.random_int(min=1, max=49151)
router_access_via_ts = router_access_via_ts or faker.boolean()
router_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
router_lo_ipv6_address = router_lo_ipv6_address or ipaddress.IPv6Address(faker.ipv6())
router_lo_iso_address = router_lo_iso_address or faker.word()
router_site = router_site or site_subscription_factory()
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.ROUTER)
router_subscription = RouterInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_ROUTER)
router_subscription = ImportedRouterInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
router_subscription.router.router_fqdn = router_fqdn
router_subscription.router.router_ts_port = router_ts_port
router_subscription.router.router_access_via_ts = router_access_via_ts
router_subscription.router.router_lo_ipv4_address = router_lo_ipv4_address
router_subscription.router.router_lo_ipv6_address = router_lo_ipv6_address
router_subscription.router.router_lo_iso_address = router_lo_iso_address
router_subscription.router.router_role = router_role
router_subscription.router.router_site = Site.from_subscription(router_site).site
router_subscription.router.vendor = Vendor.JUNIPER
router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
router_subscription.description = description
router_subscription.start_date = start_date
if status:
router_subscription.status = status
router_subscription.save()
db.session.commit()
return str(router_subscription.subscription_id)
return subscription_create
@pytest.fixture()
def iptrunk_side_subscription_factory(nokia_router_subscription_factory, faker):
def subscription_create(
iptrunk_side_node=None,
iptrunk_side_ae_iface=None,
iptrunk_side_ae_geant_a_sid=None,
iptrunk_side_ae_members=None,
iptrunk_side_ae_members_description=None,
) -> IptrunkSideBlock:
iptrunk_side_node_id = iptrunk_side_node or nokia_router_subscription_factory()
iptrunk_side_node = Router.from_subscription(iptrunk_side_node_id).router
iptrunk_side_ae_iface = iptrunk_side_ae_iface or faker.pystr()
iptrunk_side_ae_geant_a_sid = iptrunk_side_ae_geant_a_sid or faker.geant_sid()
iptrunk_side_ae_members = iptrunk_side_ae_members or [
IptrunkInterfaceBlock.new(
faker.uuid4(),
interface_name=faker.network_interface(),
interface_description=faker.sentence(),
),
IptrunkInterfaceBlock.new(
faker.uuid4(),
interface_name=faker.network_interface(),
interface_description=faker.sentence(),
),
]
return IptrunkSideBlock.new(
faker.uuid4(),
iptrunk_side_node=iptrunk_side_node,
iptrunk_side_ae_iface=iptrunk_side_ae_iface,
iptrunk_side_ae_geant_a_sid=iptrunk_side_ae_geant_a_sid,
iptrunk_side_ae_members=iptrunk_side_ae_members,
iptrunk_side_ae_members_description=iptrunk_side_ae_members_description,
)
return subscription_create
@pytest.fixture()
def iptrunk_subscription_factory(iptrunk_side_subscription_factory, faker, geant_partner):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
geant_s_sid=None,
iptrunk_description=None,
iptrunk_type=IptrunkType.LEASED,
iptrunk_speed=PhysicalPortCapacity.ONE_GIGABIT_PER_SECOND,
iptrunk_isis_metric=None,
iptrunk_ipv4_network=None,
iptrunk_ipv6_network=None,
iptrunk_sides=None,
status: SubscriptionLifecycle | None = None,
partner: dict | None = None,
*,
is_imported: bool | None = True,
) -> UUIDstr:
if partner is None:
partner = geant_partner
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.IP_TRUNK)
iptrunk_subscription = IptrunkInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_IP_TRUNK)
iptrunk_subscription = ImportedIptrunkInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
description = description or faker.sentence()
geant_s_sid = geant_s_sid or faker.geant_sid()
iptrunk_description = iptrunk_description or faker.sentence()
iptrunk_isis_metric = iptrunk_isis_metric or faker.pyint()
iptrunk_ipv4_network = iptrunk_ipv4_network or faker.ipv4_network(max_subnet=31)
iptrunk_ipv6_network = iptrunk_ipv6_network or faker.ipv6_network(max_subnet=126)
iptrunk_minimum_links = 1
iptrunk_side_a = iptrunk_side_subscription_factory()
iptrunk_side_b = iptrunk_side_subscription_factory()
iptrunk_sides = iptrunk_sides or [iptrunk_side_a, iptrunk_side_b]
iptrunk_subscription.iptrunk.geant_s_sid = geant_s_sid
iptrunk_subscription.iptrunk.iptrunk_description = iptrunk_description
iptrunk_subscription.iptrunk.iptrunk_type = iptrunk_type
iptrunk_subscription.iptrunk.iptrunk_speed = iptrunk_speed
iptrunk_subscription.iptrunk.iptrunk_minimum_links = iptrunk_minimum_links
iptrunk_subscription.iptrunk.iptrunk_isis_metric = iptrunk_isis_metric
iptrunk_subscription.iptrunk.iptrunk_ipv4_network = iptrunk_ipv4_network
iptrunk_subscription.iptrunk.iptrunk_ipv6_network = iptrunk_ipv6_network
iptrunk_subscription.iptrunk.iptrunk_sides = iptrunk_sides
iptrunk_subscription = SubscriptionModel.from_other_lifecycle(
iptrunk_subscription,
SubscriptionLifecycle.ACTIVE,
)
if status:
iptrunk_subscription.status = status
iptrunk_subscription.description = description
iptrunk_subscription.start_date = start_date
iptrunk_subscription.save()
db.session.commit()
return str(iptrunk_subscription.subscription_id)
return subscription_create
@pytest.fixture()
def office_router_subscription_factory(site_subscription_factory, faker, geant_partner):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
office_router_fqdn=None,
office_router_ts_port=None,
office_router_lo_ipv4_address=None,
office_router_lo_ipv6_address=None,
office_router_site=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)
office_router_fqdn = office_router_fqdn or faker.domain_name(levels=4)
office_router_ts_port = office_router_ts_port or faker.random_int(min=1, max=49151)
office_router_lo_ipv4_address = office_router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
office_router_lo_ipv6_address = office_router_lo_ipv6_address or ipaddress.IPv6Address(faker.ipv6())
office_router_site = office_router_site or site_subscription_factory()
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.OFFICE_ROUTER)
office_router_subscription = OfficeRouterInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_OFFICE_ROUTER)
office_router_subscription = ImportedOfficeRouterInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
office_router_subscription.office_router.office_router_fqdn = office_router_fqdn
office_router_subscription.office_router.office_router_ts_port = office_router_ts_port
office_router_subscription.office_router.office_router_lo_ipv4_address = office_router_lo_ipv4_address
office_router_subscription.office_router.office_router_lo_ipv6_address = office_router_lo_ipv6_address
office_router_subscription.office_router.office_router_site = Site.from_subscription(office_router_site).site
office_router_subscription.office_router.vendor = Vendor.NOKIA
office_router_subscription = SubscriptionModel.from_other_lifecycle(
office_router_subscription, SubscriptionLifecycle.ACTIVE
)
office_router_subscription.description = description
office_router_subscription.start_date = start_date
if status:
office_router_subscription.status = status
office_router_subscription.save()
db.session.commit()
return str(office_router_subscription.subscription_id)
return subscription_create
@pytest.fixture()
def super_pop_switch_subscription_factory(site_subscription_factory, faker, geant_partner):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
super_pop_switch_fqdn=None,
super_pop_switch_ts_port=None,
super_pop_switch_mgmt_ipv4_address=None,
super_pop_switch_site=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)
super_pop_switch_fqdn = super_pop_switch_fqdn or faker.domain_name(levels=4)
super_pop_switch_ts_port = super_pop_switch_ts_port or faker.random_int(min=1, max=49151)
super_pop_switch_mgmt_ipv4_address = super_pop_switch_mgmt_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
super_pop_switch_site = super_pop_switch_site or site_subscription_factory()
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.SUPER_POP_SWITCH)
super_pop_switch_subscription = SuperPopSwitchInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_SUPER_POP_SWITCH)
super_pop_switch_subscription = ImportedSuperPopSwitchInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
super_pop_switch_subscription.super_pop_switch.super_pop_switch_fqdn = super_pop_switch_fqdn
super_pop_switch_subscription.super_pop_switch.super_pop_switch_ts_port = super_pop_switch_ts_port
super_pop_switch_subscription.super_pop_switch.super_pop_switch_mgmt_ipv4_address = (
super_pop_switch_mgmt_ipv4_address
)
super_pop_switch_subscription.super_pop_switch.super_pop_switch_site = Site.from_subscription(
super_pop_switch_site
).site
super_pop_switch_subscription.super_pop_switch.vendor = Vendor.NOKIA
super_pop_switch_subscription = SubscriptionModel.from_other_lifecycle(
super_pop_switch_subscription, SubscriptionLifecycle.ACTIVE
)
super_pop_switch_subscription.description = description
super_pop_switch_subscription.start_date = start_date
if status:
super_pop_switch_subscription.status = status
super_pop_switch_subscription.save()
db.session.commit()
return str(super_pop_switch_subscription.subscription_id)
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
@step("Insert UUID in state")
def insert_object():
return {"subscription_id": str(uuid4()), "model": generic_product_one.from_subscription(generic_subscription_1)}
@step("Test that it is a string now")
def check_object(subscription_id: Any, model: dict) -> None:
# This is actually a test. It would be nicer to have this in a proper test but that takes to much setup that
# already happens here. So we hijack this fixture and run this test for all tests that use this fixture
# (which should not be an issue)
assert isinstance(subscription_id, str)
assert isinstance(model, dict)
@inputstep("Modify", assignee=Assignee.CHANGES)
def modify(subscription_id: UUIDstr) -> FormGenerator:
class TestChoice(Choice):
A = "A"
B = "B"
C = "C"
class TestForm(FormPage):
generic_select: TestChoice
user_input = yield TestForm
return user_input.model_dump()
@workflow("Workflow")
def workflow_for_testing_processes_py():
return init >> insert_object >> check_object >> modify >> done
with WorkflowInstanceForTests(workflow_for_testing_processes_py, "workflow_for_testing_processes_py") as wf:
yield wf
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/fixtures/__init__.py 0 → 100644
View file @ f3cdbbcc
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 0 → 100644
View file @ f3cdbbcc
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/fixtures/edge_port_fixtures.py 0 → 100644
View file @ f3cdbbcc
import pytest
from orchestrator.db import (
db,
)
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
from gso.products import ProductName, Router
from gso.products.product_blocks.edge_port import (
EdgePortAEMemberBlock,
EdgePortType,
EncapsulationType,
)
from gso.products.product_types.edge_port import EdgePortInactive, ImportedEdgePortInactive
from gso.services import subscriptions
from gso.utils.types.interfaces import PhysicalPortCapacity
@pytest.fixture()
def edge_port_subscription_factory(faker, partner_factory, nokia_router_subscription_factory):
def subscription_create(
description=None,
partner: dict | None = None,
start_date="2023-05-24T00:00:00+00:00",
node=None,
name=None,
edge_port_description=None,
encapsulation=EncapsulationType.DOT1Q,
mac_address=None,
member_speed=PhysicalPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
minimum_links=None,
edge_port_type=EdgePortType.PUBLIC,
geant_ga_id=None,
edge_port_ae_members=None,
status: SubscriptionLifecycle | None = None,
*,
enable_lacp=True,
ignore_if_down=False,
is_imported=True,
) -> UUIDstr:
partner = partner or partner_factory()
edge_port_node = Router.from_subscription(nokia_router_subscription_factory()).router
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.EDGE_PORT)
edge_port_subscription = EdgePortInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_EDGE_PORT)
edge_port_subscription = ImportedEdgePortInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
edge_port_subscription.edge_port.edge_port_description = description or faker.text(max_nb_chars=30)
edge_port_subscription.edge_port.edge_port_geant_ga_id = geant_ga_id or faker.geant_sid()
edge_port_subscription.edge_port.edge_port_node = node or edge_port_node
edge_port_subscription.edge_port.edge_port_name = name or f"lag-{faker.pyint(21, 50)}"
edge_port_subscription.edge_port.edge_port_description = edge_port_description or faker.sentence()
edge_port_subscription.edge_port.edge_port_enable_lacp = enable_lacp
edge_port_subscription.edge_port.edge_port_encapsulation = encapsulation
edge_port_subscription.edge_port.edge_port_mac_address = mac_address or faker.mac_address()
edge_port_subscription.edge_port.edge_port_member_speed = member_speed
edge_port_subscription.edge_port.edge_port_minimum_links = minimum_links or faker.pyint(1, 2)
edge_port_subscription.edge_port.edge_port_type = edge_port_type
edge_port_subscription.edge_port.edge_port_ignore_if_down = ignore_if_down
edge_port_subscription.edge_port.edge_port_geant_ga_id = geant_ga_id
edge_port_subscription.edge_port.edge_port_ae_members = edge_port_ae_members or [
EdgePortAEMemberBlock.new(
faker.uuid4(),
interface_name="Interface2",
interface_description=faker.sentence(),
),
EdgePortAEMemberBlock.new(
faker.uuid4(),
interface_name="Interface3",
interface_description=faker.sentence(),
),
]
edge_port_subscription = SubscriptionModel.from_other_lifecycle(
edge_port_subscription,
SubscriptionLifecycle.ACTIVE,
)
if status:
edge_port_subscription.status = status
edge_port_subscription.description = description or faker.text(max_nb_chars=30)
edge_port_subscription.start_date = start_date
edge_port_subscription.save()
db.session.commit()
return str(edge_port_subscription.subscription_id)
return subscription_create
test/fixtures/geant_ip_fixtures.py 0 → 100644
View file @ f3cdbbcc
import random
from uuid import uuid4
import pytest
from orchestrator.db import db
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
from gso.products import EdgePort, ProductName
from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
from gso.products.product_blocks.nren_l3_core_service import NRENAccessPort
from gso.products.product_blocks.service_binding_port import ServiceBindingPort
from gso.products.product_types.nren_l3_core_service import GeantIPInactive, ImportedGeantIP
from gso.services import subscriptions
from gso.utils.shared_enums import APType, SBPType
from gso.utils.types.ip_address import IPAddress
@pytest.fixture()
def bgp_session_subscription_factory(faker):
def create_bgp_session(
peer_address: IPAddress | None = None,
bfd_interval: int = 2,
bfd_multiplier: int = 2,
families: list[IPFamily] | None = None,
authentication_key: str | None = None,
*,
is_multi_hop: bool = False,
has_custom_policies: bool = False,
bfd_enabled: bool = True,
multipath_enabled: bool | None = True,
send_default_route: bool | None = True,
is_passive: bool | None = False,
rtbh_enabled: bool | None = False,
):
return BGPSession.new(
subscription_id=uuid4(),
peer_address=peer_address or faker.ipv4(),
bfd_enabled=bfd_enabled,
families=families or [IPFamily.V4UNICAST],
has_custom_policies=has_custom_policies,
authentication_key=authentication_key or faker.password(),
multipath_enabled=multipath_enabled,
send_default_route=send_default_route,
is_multi_hop=is_multi_hop,
bfd_interval=bfd_interval,
bfd_multiplier=bfd_multiplier,
rtbh_enabled=rtbh_enabled,
is_passive=is_passive,
)
return create_bgp_session
@pytest.fixture()
def service_binding_port_factory(faker, bgp_session_subscription_factory, edge_port_subscription_factory):
def create_service_binding_port(
sbp_bgp_session_list: list | None = None,
geant_sid: str | None = None,
sbp_type: SBPType = SBPType.L3,
ipv4_address: str | None = None,
ipv4_mask: int | None = None,
ipv6_address: str | None = None,
ipv6_mask: int | None = None,
vlan_id: int | None = None,
edge_port: EdgePort | None = None,
*,
custom_firewall_filters: bool = False,
is_tagged: bool = False,
):
return ServiceBindingPort.new(
subscription_id=uuid4(),
is_tagged=is_tagged,
vlan_id=vlan_id or faker.pyint(min_value=1, max_value=4096),
sbp_type=sbp_type,
ipv4_address=ipv4_address or faker.ipv4(),
ipv4_mask=ipv4_mask or faker.ipv4_netmask(),
ipv6_address=ipv6_address or faker.ipv6(),
ipv6_mask=ipv6_mask or faker.ipv6_netmask(),
custom_firewall_filters=custom_firewall_filters,
geant_sid=geant_sid or faker.geant_sid(),
sbp_bgp_session_list=sbp_bgp_session_list
or [
bgp_session_subscription_factory(families=[IPFamily.V4UNICAST]),
bgp_session_subscription_factory(families=[IPFamily.V6UNICAST], peer_address=faker.ipv6()),
],
edge_port=edge_port or EdgePort.from_subscription(edge_port_subscription_factory()).edge_port,
)
return create_service_binding_port
@pytest.fixture()
def nren_access_port_factory(faker, service_binding_port_factory):
def create_nren_access_port(
nren_ap_type: APType | None = None,
service_binding_port: ServiceBindingPort | None = None,
):
return NRENAccessPort.new(
subscription_id=uuid4(),
nren_ap_type=nren_ap_type or random.choice(list(APType)), # noqa: S311
geant_ip_sbp=service_binding_port or service_binding_port_factory(),
)
return create_nren_access_port
@pytest.fixture()
def geant_ip_subscription_factory(
faker,
partner_factory,
nren_access_port_factory,
):
def create_geant_ip_subscription(
description=None,
partner: dict | None = None,
nren_ap_list: list[NRENAccessPort] | None = None,
start_date="2023-05-24T00:00:00+00:00",
status: SubscriptionLifecycle | None = None,
*,
is_imported=True,
) -> UUIDstr:
partner = partner or partner_factory()
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.GEANT_IP)
geant_ip_subscription = GeantIPInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_GEANT_IP)
geant_ip_subscription = ImportedGeantIP.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
# Default nren_ap_list creation with primary and backup access ports
geant_ip_subscription.geant_ip.geant_ip_ap_list = nren_ap_list or [
nren_access_port_factory(nren_ap_type=APType.PRIMARY),
nren_access_port_factory(nren_ap_type=APType.BACKUP),
]
# Update subscription with description, start date, and status
geant_ip_subscription = SubscriptionModel.from_other_lifecycle(
geant_ip_subscription,
SubscriptionLifecycle.ACTIVE,
)
geant_ip_subscription.description = description or faker.sentence()
geant_ip_subscription.start_date = start_date
geant_ip_subscription.status = status or SubscriptionLifecycle.ACTIVE
geant_ip_subscription.save()
db.session.commit()
return str(geant_ip_subscription.subscription_id)
return create_geant_ip_subscription
test/fixtures/iptrunk_fixtures.py 0 → 100644
View file @ f3cdbbcc
import pytest
from orchestrator.db import (
db,
)
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
from gso.products import ProductName
from gso.products.product_blocks.iptrunk import (
IptrunkInterfaceBlock,
IptrunkSideBlock,
IptrunkType,
)
from gso.products.product_types.iptrunk import ImportedIptrunkInactive, IptrunkInactive
from gso.products.product_types.router import Router
from gso.services import subscriptions
from gso.utils.types.interfaces import PhysicalPortCapacity
@pytest.fixture()
def iptrunk_side_subscription_factory(nokia_router_subscription_factory, faker):
def subscription_create(
iptrunk_side_node=None,
iptrunk_side_ae_iface=None,
iptrunk_side_ae_geant_a_sid=None,
iptrunk_side_ae_members=None,
iptrunk_side_ae_members_description=None,
) -> IptrunkSideBlock:
iptrunk_side_node_id = iptrunk_side_node or nokia_router_subscription_factory()
iptrunk_side_node = Router.from_subscription(iptrunk_side_node_id).router
iptrunk_side_ae_iface = iptrunk_side_ae_iface or faker.pystr()
iptrunk_side_ae_geant_a_sid = iptrunk_side_ae_geant_a_sid or faker.geant_sid()
iptrunk_side_ae_members = iptrunk_side_ae_members or [
IptrunkInterfaceBlock.new(
faker.uuid4(),
interface_name=faker.network_interface(),
interface_description=faker.sentence(),
),
IptrunkInterfaceBlock.new(
faker.uuid4(),
interface_name=faker.network_interface(),
interface_description=faker.sentence(),
),
]
return IptrunkSideBlock.new(
faker.uuid4(),
iptrunk_side_node=iptrunk_side_node,
iptrunk_side_ae_iface=iptrunk_side_ae_iface,
iptrunk_side_ae_geant_a_sid=iptrunk_side_ae_geant_a_sid,
iptrunk_side_ae_members=iptrunk_side_ae_members,
iptrunk_side_ae_members_description=iptrunk_side_ae_members_description,
)
return subscription_create
@pytest.fixture()
def iptrunk_subscription_factory(iptrunk_side_subscription_factory, faker, geant_partner):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
geant_s_sid=None,
iptrunk_description=None,
iptrunk_type=IptrunkType.LEASED,
iptrunk_speed=PhysicalPortCapacity.ONE_GIGABIT_PER_SECOND,
iptrunk_isis_metric=None,
iptrunk_ipv4_network=None,
iptrunk_ipv6_network=None,
iptrunk_sides=None,
status: SubscriptionLifecycle | None = None,
partner: dict | None = None,
*,
is_imported: bool | None = True,
) -> UUIDstr:
if partner is None:
partner = geant_partner
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.IP_TRUNK)
iptrunk_subscription = IptrunkInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_IP_TRUNK)
iptrunk_subscription = ImportedIptrunkInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
description = description or faker.sentence()
geant_s_sid = geant_s_sid or faker.geant_sid()
iptrunk_description = iptrunk_description or faker.sentence()
iptrunk_isis_metric = iptrunk_isis_metric or faker.pyint()
iptrunk_ipv4_network = iptrunk_ipv4_network or faker.ipv4_network(max_subnet=31)
iptrunk_ipv6_network = iptrunk_ipv6_network or faker.ipv6_network(max_subnet=126)
iptrunk_minimum_links = 1
iptrunk_side_a = iptrunk_side_subscription_factory()
iptrunk_side_b = iptrunk_side_subscription_factory()
iptrunk_sides = iptrunk_sides or [iptrunk_side_a, iptrunk_side_b]
iptrunk_subscription.iptrunk.geant_s_sid = geant_s_sid
iptrunk_subscription.iptrunk.iptrunk_description = iptrunk_description
iptrunk_subscription.iptrunk.iptrunk_type = iptrunk_type
iptrunk_subscription.iptrunk.iptrunk_speed = iptrunk_speed
iptrunk_subscription.iptrunk.iptrunk_minimum_links = iptrunk_minimum_links
iptrunk_subscription.iptrunk.iptrunk_isis_metric = iptrunk_isis_metric
iptrunk_subscription.iptrunk.iptrunk_ipv4_network = iptrunk_ipv4_network
iptrunk_subscription.iptrunk.iptrunk_ipv6_network = iptrunk_ipv6_network
iptrunk_subscription.iptrunk.iptrunk_sides = iptrunk_sides
iptrunk_subscription = SubscriptionModel.from_other_lifecycle(
iptrunk_subscription,
SubscriptionLifecycle.ACTIVE,
)
if status:
iptrunk_subscription.status = status
iptrunk_subscription.description = description
iptrunk_subscription.start_date = start_date
iptrunk_subscription.save()
db.session.commit()
return str(iptrunk_subscription.subscription_id)
return subscription_create
test/fixtures/office_router_fixtures.py 0 → 100644
View file @ f3cdbbcc
import ipaddress
import pytest
from orchestrator.db import (
db,
)
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
from gso.products import ProductName
from gso.products.product_types.office_router import ImportedOfficeRouterInactive, OfficeRouterInactive
from gso.products.product_types.site import Site
from gso.services import subscriptions
from gso.utils.shared_enums import Vendor
@pytest.fixture()
def office_router_subscription_factory(site_subscription_factory, faker, geant_partner):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
office_router_fqdn=None,
office_router_ts_port=None,
office_router_lo_ipv4_address=None,
office_router_lo_ipv6_address=None,
office_router_site=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)
office_router_fqdn = office_router_fqdn or faker.domain_name(levels=4)
office_router_ts_port = office_router_ts_port or faker.random_int(min=1, max=49151)
office_router_lo_ipv4_address = office_router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
office_router_lo_ipv6_address = office_router_lo_ipv6_address or ipaddress.IPv6Address(faker.ipv6())
office_router_site = office_router_site or site_subscription_factory()
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.OFFICE_ROUTER)
office_router_subscription = OfficeRouterInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_OFFICE_ROUTER)
office_router_subscription = ImportedOfficeRouterInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
office_router_subscription.office_router.office_router_fqdn = office_router_fqdn
office_router_subscription.office_router.office_router_ts_port = office_router_ts_port
office_router_subscription.office_router.office_router_lo_ipv4_address = office_router_lo_ipv4_address
office_router_subscription.office_router.office_router_lo_ipv6_address = office_router_lo_ipv6_address
office_router_subscription.office_router.office_router_site = Site.from_subscription(office_router_site).site
office_router_subscription.office_router.vendor = Vendor.NOKIA
office_router_subscription = SubscriptionModel.from_other_lifecycle(
office_router_subscription, SubscriptionLifecycle.ACTIVE
)
office_router_subscription.description = description
office_router_subscription.start_date = start_date
if status:
office_router_subscription.status = status
office_router_subscription.save()
db.session.commit()
return str(office_router_subscription.subscription_id)
return subscription_create
test/fixtures/opengear_fixtures.py 0 → 100644
View file @ f3cdbbcc
import pytest
from orchestrator.db import (
db,
)
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
from gso.products import ProductName
from gso.products.product_types.opengear import ImportedOpengearInactive, OpengearInactive
from gso.products.product_types.site import Site
from gso.services import subscriptions
@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
test/fixtures/router_fixtures.py 0 → 100644
View file @ f3cdbbcc
import ipaddress
import pytest
from orchestrator.db import (
db,
)
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
from gso.products import ProductName
from gso.products.product_blocks.router import RouterRole
from gso.products.product_types.router import ImportedRouterInactive, RouterInactive
from gso.products.product_types.site import Site
from gso.services import subscriptions
from gso.utils.shared_enums import Vendor
@pytest.fixture()
def nokia_router_subscription_factory(site_subscription_factory, faker, geant_partner):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
router_fqdn=None,
router_ts_port=None,
router_access_via_ts=None,
router_lo_ipv4_address=None,
router_lo_ipv6_address=None,
router_lo_iso_address=None,
router_role=RouterRole.PE,
router_site=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)
router_fqdn = router_fqdn or faker.domain_name(levels=4)
router_ts_port = router_ts_port or faker.random_int(min=1, max=49151)
router_access_via_ts = router_access_via_ts or faker.boolean()
router_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
router_lo_ipv6_address = router_lo_ipv6_address or ipaddress.IPv6Address(faker.ipv6())
router_lo_iso_address = router_lo_iso_address or faker.word()
router_site = router_site or site_subscription_factory()
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.ROUTER)
router_subscription = RouterInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_ROUTER)
router_subscription = ImportedRouterInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
router_subscription.router.router_fqdn = router_fqdn
router_subscription.router.router_ts_port = router_ts_port
router_subscription.router.router_access_via_ts = router_access_via_ts
router_subscription.router.router_lo_ipv4_address = router_lo_ipv4_address
router_subscription.router.router_lo_ipv6_address = router_lo_ipv6_address
router_subscription.router.router_lo_iso_address = router_lo_iso_address
router_subscription.router.router_role = router_role
router_subscription.router.router_site = Site.from_subscription(router_site).site
router_subscription.router.vendor = Vendor.NOKIA
router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
router_subscription.insync = True
router_subscription.description = description
router_subscription.start_date = start_date
if status:
router_subscription.status = status
router_subscription.save()
db.session.commit()
return str(router_subscription.subscription_id)
return subscription_create
@pytest.fixture()
def juniper_router_subscription_factory(site_subscription_factory, faker, geant_partner):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
router_fqdn=None,
router_ts_port=None,
router_access_via_ts=None,
router_lo_ipv4_address=None,
router_lo_ipv6_address=None,
router_lo_iso_address=None,
router_role=RouterRole.PE,
router_site=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)
router_fqdn = router_fqdn or faker.domain_name(levels=4)
router_ts_port = router_ts_port or faker.random_int(min=1, max=49151)
router_access_via_ts = router_access_via_ts or faker.boolean()
router_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
router_lo_ipv6_address = router_lo_ipv6_address or ipaddress.IPv6Address(faker.ipv6())
router_lo_iso_address = router_lo_iso_address or faker.word()
router_site = router_site or site_subscription_factory()
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.ROUTER)
router_subscription = RouterInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_ROUTER)
router_subscription = ImportedRouterInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
router_subscription.router.router_fqdn = router_fqdn
router_subscription.router.router_ts_port = router_ts_port
router_subscription.router.router_access_via_ts = router_access_via_ts
router_subscription.router.router_lo_ipv4_address = router_lo_ipv4_address
router_subscription.router.router_lo_ipv6_address = router_lo_ipv6_address
router_subscription.router.router_lo_iso_address = router_lo_iso_address
router_subscription.router.router_role = router_role
router_subscription.router.router_site = Site.from_subscription(router_site).site
router_subscription.router.vendor = Vendor.JUNIPER
router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
router_subscription.description = description
router_subscription.start_date = start_date
if status:
router_subscription.status = status
router_subscription.save()
db.session.commit()
return str(router_subscription.subscription_id)
return subscription_create
test/fixtures/site_fixtures.py 0 → 100644
View file @ f3cdbbcc
import pytest
from orchestrator.db import (
db,
)
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
from gso.products import ProductName
from gso.products.product_blocks.site import SiteTier
from gso.products.product_types.site import ImportedSiteInactive, SiteInactive
from gso.services import subscriptions
@pytest.fixture()
def site_subscription_factory(faker, geant_partner):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
site_name=None,
site_city=None,
site_country=None,
site_country_code=None,
site_latitude=None,
site_longitude=None,
site_bgp_community_id=None,
site_internal_id=None,
site_tier=SiteTier.TIER1,
site_ts_address=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 "Site Subscription"
site_name = site_name or faker.site_name()
site_city = site_city or faker.city()
site_country = site_country or faker.country()
site_country_code = site_country_code or faker.country_code()
site_latitude = site_latitude or str(faker.latitude())
site_longitude = site_longitude or str(faker.longitude())
site_bgp_community_id = site_bgp_community_id or faker.pyint()
site_internal_id = site_internal_id or faker.pyint()
site_ts_address = site_ts_address or faker.ipv4()
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.SITE)
site_subscription = SiteInactive.from_product_id(product_id, customer_id=partner["partner_id"], insync=True)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_SITE)
site_subscription = ImportedSiteInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
site_subscription.site.site_city = site_city
site_subscription.site.site_name = site_name
site_subscription.site.site_country = site_country
site_subscription.site.site_country_code = site_country_code
site_subscription.site.site_latitude = site_latitude
site_subscription.site.site_longitude = site_longitude
site_subscription.site.site_bgp_community_id = site_bgp_community_id
site_subscription.site.site_internal_id = site_internal_id
site_subscription.site.site_tier = site_tier
site_subscription.site.site_ts_address = site_ts_address
site_subscription = SubscriptionModel.from_other_lifecycle(site_subscription, SubscriptionLifecycle.ACTIVE)
site_subscription.description = description
site_subscription.start_date = start_date
if status:
site_subscription.status = status
site_subscription.save()
db.session.commit()
return str(site_subscription.subscription_id)
return subscription_create
test/fixtures/super_pop_switch_fixtures.py 0 → 100644
View file @ f3cdbbcc
import ipaddress
import pytest
from orchestrator.db import (
db,
)
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
from gso.products import ProductName
from gso.products.product_types.site import Site
from gso.products.product_types.super_pop_switch import ImportedSuperPopSwitchInactive, SuperPopSwitchInactive
from gso.services import subscriptions
from gso.utils.shared_enums import Vendor
@pytest.fixture()
def super_pop_switch_subscription_factory(site_subscription_factory, faker, geant_partner):
def subscription_create(
description=None,
start_date="2023-05-24T00:00:00+00:00",
super_pop_switch_fqdn=None,
super_pop_switch_ts_port=None,
super_pop_switch_mgmt_ipv4_address=None,
super_pop_switch_site=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)
super_pop_switch_fqdn = super_pop_switch_fqdn or faker.domain_name(levels=4)
super_pop_switch_ts_port = super_pop_switch_ts_port or faker.random_int(min=1, max=49151)
super_pop_switch_mgmt_ipv4_address = super_pop_switch_mgmt_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
super_pop_switch_site = super_pop_switch_site or site_subscription_factory()
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.SUPER_POP_SWITCH)
super_pop_switch_subscription = SuperPopSwitchInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_SUPER_POP_SWITCH)
super_pop_switch_subscription = ImportedSuperPopSwitchInactive.from_product_id(
product_id, customer_id=partner["partner_id"], insync=True
)
super_pop_switch_subscription.super_pop_switch.super_pop_switch_fqdn = super_pop_switch_fqdn
super_pop_switch_subscription.super_pop_switch.super_pop_switch_ts_port = super_pop_switch_ts_port
super_pop_switch_subscription.super_pop_switch.super_pop_switch_mgmt_ipv4_address = (
super_pop_switch_mgmt_ipv4_address
)
super_pop_switch_subscription.super_pop_switch.super_pop_switch_site = Site.from_subscription(
super_pop_switch_site
).site
super_pop_switch_subscription.super_pop_switch.vendor = Vendor.NOKIA
super_pop_switch_subscription = SubscriptionModel.from_other_lifecycle(
super_pop_switch_subscription, SubscriptionLifecycle.ACTIVE
)
super_pop_switch_subscription.description = description
super_pop_switch_subscription.start_date = start_date
if status:
super_pop_switch_subscription.status = status
super_pop_switch_subscription.save()
db.session.commit()
return str(super_pop_switch_subscription.subscription_id)
return subscription_create
test/services/conftest.py
View file @ f3cdbbcc
......@@ -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
View file @ f3cdbbcc
......@@ -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()
......