Skip to content
Snippets Groups Projects
Commit b4ab4186 authored by Neda Moeini's avatar Neda Moeini
Browse files

Make Ruff happy

parent 5fddea40
No related branches found
No related tags found
1 merge request!307Feature/l2circuits
...@@ -48,7 +48,7 @@ def initial_input_generator(product_name: str) -> FormGenerator: ...@@ -48,7 +48,7 @@ def initial_input_generator(product_name: str) -> FormGenerator:
) )
vlan_id: VLAN_ID vlan_id: VLAN_ID
def _vlan_range_field(*, is_tagged: bool) -> VLAN_ID | int: def _vlan_range_field(*, is_tagged: bool) -> VLAN_ID:
"""Return the appropriate field type based on whether the circuit is tagged.""" """Return the appropriate field type based on whether the circuit is tagged."""
return VLAN_ID if is_tagged else ReadOnlyField(None, default_type=int) return VLAN_ID if is_tagged else ReadOnlyField(None, default_type=int)
......
...@@ -21,6 +21,7 @@ __all__ = [ ...@@ -21,6 +21,7 @@ __all__ = [
"iptrunk_side_subscription_factory", "iptrunk_side_subscription_factory",
"iptrunk_subscription_factory", "iptrunk_subscription_factory",
"lan_switch_interconnect_subscription_factory", "lan_switch_interconnect_subscription_factory",
"layer_2_circuit_subscription_factory",
"nren_access_port_factory", "nren_access_port_factory",
"nren_l3_core_service_subscription_factory", "nren_l3_core_service_subscription_factory",
"office_router_subscription_factory", "office_router_subscription_factory",
...@@ -30,5 +31,4 @@ __all__ = [ ...@@ -30,5 +31,4 @@ __all__ = [
"site_subscription_factory", "site_subscription_factory",
"super_pop_switch_subscription_factory", "super_pop_switch_subscription_factory",
"switch_subscription_factory", "switch_subscription_factory",
"layer_2_circuit_subscription_factory",
] ]
...@@ -5,7 +5,7 @@ from orchestrator.db import db ...@@ -5,7 +5,7 @@ from orchestrator.db import db
from orchestrator.domain import SubscriptionModel from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr from orchestrator.types import SubscriptionLifecycle, UUIDstr
from gso.products import ProductName, EdgePort from gso.products import EdgePort, ProductName
from gso.products.product_blocks.layer_2_circuit import Layer2CircuitSideBlockInactive, Layer2CircuitType from gso.products.product_blocks.layer_2_circuit import Layer2CircuitSideBlockInactive, Layer2CircuitType
from gso.products.product_blocks.service_binding_port import ServiceBindingPortInactive from gso.products.product_blocks.service_binding_port import ServiceBindingPortInactive
from gso.products.product_types.layer_2_circuit import ( from gso.products.product_types.layer_2_circuit import (
...@@ -23,24 +23,23 @@ from gso.utils.types.virtual_identifiers import VLAN_ID ...@@ -23,24 +23,23 @@ from gso.utils.types.virtual_identifiers import VLAN_ID
@pytest.fixture() @pytest.fixture()
def layer_2_circuit_subscription_factory(faker, geant_partner, edge_port_subscription_factory): def layer_2_circuit_subscription_factory(faker, geant_partner, edge_port_subscription_factory):
def create_subscription( def create_subscription(
description: str | None = None, description: str | None = None,
partner: dict | None = None, partner: dict | None = None,
status: SubscriptionLifecycle | None = None, status: SubscriptionLifecycle | None = None,
start_date: str | None = "2024-01-01T10:20:30+01:02", start_date: str | None = "2024-01-01T10:20:30+01:02",
layer_2_circuit_service_type: Layer2CircuitServiceType | None = None, layer_2_circuit_service_type: Layer2CircuitServiceType | None = None,
layer_2_circuit_type: Layer2CircuitType = Layer2CircuitType.TAGGED, layer_2_circuit_type: Layer2CircuitType = Layer2CircuitType.TAGGED,
vlan_range_lower_bound: VLAN_ID | None = None, vlan_range_lower_bound: VLAN_ID | None = None,
vlan_range_upper_bound: VLAN_ID | None = None, vlan_range_upper_bound: VLAN_ID | None = None,
policer_bandwidth: BandwidthString | None = None, policer_bandwidth: BandwidthString | None = None,
layer_2_circuit_side_a_edgeport: UUIDstr | None = None, layer_2_circuit_side_a_edgeport: UUIDstr | None = None,
vlan_id_side_a: VLAN_ID | None = None, vlan_id_side_a: VLAN_ID | None = None,
layer_2_circuit_side_b_edgeport: UUIDstr | None = None, layer_2_circuit_side_b_edgeport: UUIDstr | None = None,
vlan_id_side_b: VLAN_ID | None = None, vlan_id_side_b: VLAN_ID | None = None,
geant_sid: str | None = None, geant_sid: str | None = None,
*, *,
policer_enabled: bool = False, policer_enabled: bool = False,
) -> UUIDstr: ) -> UUIDstr:
# Assign default partner if none provided # Assign default partner if none provided
if partner is None: if partner is None:
partner = geant_partner partner = geant_partner
...@@ -100,7 +99,9 @@ def layer_2_circuit_subscription_factory(faker, geant_partner, edge_port_subscri ...@@ -100,7 +99,9 @@ def layer_2_circuit_subscription_factory(faker, geant_partner, edge_port_subscri
subscription.layer_2_circuit.policer_enabled = policer_enabled subscription.layer_2_circuit.policer_enabled = policer_enabled
subscription.layer_2_circuit.bandwidth = policer_bandwidth if policer_enabled else None subscription.layer_2_circuit.bandwidth = policer_bandwidth if policer_enabled else None
subscription.description = description or f"{subscription.product.name} - {subscription.layer_2_circuit.virtual_circuit_id}" subscription.description = description or (
f"{subscription.product.name} - " f"{subscription.layer_2_circuit.virtual_circuit_id}"
)
subscription = SubscriptionModel.from_other_lifecycle(subscription, SubscriptionLifecycle.ACTIVE) subscription = SubscriptionModel.from_other_lifecycle(subscription, SubscriptionLifecycle.ACTIVE)
subscription.insync = True subscription.insync = True
......
...@@ -10,11 +10,11 @@ from test.workflows import assert_complete, extract_state, run_workflow ...@@ -10,11 +10,11 @@ from test.workflows import assert_complete, extract_state, run_workflow
@pytest.mark.parametrize("layer_2_circuit_service_type", [ProductName.GEANT_PLUS, ProductName.EXPRESSROUTE]) @pytest.mark.parametrize("layer_2_circuit_service_type", [ProductName.GEANT_PLUS, ProductName.EXPRESSROUTE])
@pytest.mark.workflow() @pytest.mark.workflow()
def test_modify_layer_2_circuit_change_policer_bandwidth( def test_modify_layer_2_circuit_change_policer_bandwidth(
layer_2_circuit_service_type, layer_2_circuit_service_type,
layer_2_circuit_subscription_factory, layer_2_circuit_subscription_factory,
faker, faker,
partner_factory, partner_factory,
data_config_filename, data_config_filename,
): ):
subscription_id = layer_2_circuit_subscription_factory(layer_2_circuit_service_type=layer_2_circuit_service_type) subscription_id = layer_2_circuit_subscription_factory(layer_2_circuit_service_type=layer_2_circuit_service_type)
subscription = Layer2Circuit.from_subscription(subscription_id) subscription = Layer2Circuit.from_subscription(subscription_id)
...@@ -29,7 +29,7 @@ def test_modify_layer_2_circuit_change_policer_bandwidth( ...@@ -29,7 +29,7 @@ def test_modify_layer_2_circuit_change_policer_bandwidth(
"vlan_range_lower_bound": subscription.layer_2_circuit.vlan_range_lower_bound, "vlan_range_lower_bound": subscription.layer_2_circuit.vlan_range_lower_bound,
"vlan_range_upper_bound": subscription.layer_2_circuit.vlan_range_upper_bound, "vlan_range_upper_bound": subscription.layer_2_circuit.vlan_range_upper_bound,
"policer_bandwidth": None, "policer_bandwidth": None,
} },
] ]
result, _, _ = run_workflow("modify_layer_2_circuit", input_form_data) result, _, _ = run_workflow("modify_layer_2_circuit", input_form_data)
assert_complete(result) assert_complete(result)
...@@ -41,11 +41,11 @@ def test_modify_layer_2_circuit_change_policer_bandwidth( ...@@ -41,11 +41,11 @@ def test_modify_layer_2_circuit_change_policer_bandwidth(
@pytest.mark.parametrize("layer_2_circuit_service_type", [ProductName.GEANT_PLUS, ProductName.EXPRESSROUTE]) @pytest.mark.parametrize("layer_2_circuit_service_type", [ProductName.GEANT_PLUS, ProductName.EXPRESSROUTE])
@pytest.mark.workflow() @pytest.mark.workflow()
def test_modify_layer_2_circuit_change_circuit_type( def test_modify_layer_2_circuit_change_circuit_type(
layer_2_circuit_service_type, layer_2_circuit_service_type,
layer_2_circuit_subscription_factory, layer_2_circuit_subscription_factory,
faker, faker,
partner_factory, partner_factory,
data_config_filename, data_config_filename,
): ):
subscription_id = layer_2_circuit_subscription_factory(layer_2_circuit_service_type=layer_2_circuit_service_type) subscription_id = layer_2_circuit_subscription_factory(layer_2_circuit_service_type=layer_2_circuit_service_type)
subscription = Layer2Circuit.from_subscription(subscription_id) subscription = Layer2Circuit.from_subscription(subscription_id)
...@@ -59,7 +59,7 @@ def test_modify_layer_2_circuit_change_circuit_type( ...@@ -59,7 +59,7 @@ def test_modify_layer_2_circuit_change_circuit_type(
"vlan_range_lower_bound": None, "vlan_range_lower_bound": None,
"vlan_range_upper_bound": None, "vlan_range_upper_bound": None,
"policer_bandwidth": subscription.layer_2_circuit.bandwidth, "policer_bandwidth": subscription.layer_2_circuit.bandwidth,
} },
] ]
result, _, _ = run_workflow("modify_layer_2_circuit", input_form_data) result, _, _ = run_workflow("modify_layer_2_circuit", input_form_data)
assert_complete(result) assert_complete(result)
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment