diff --git a/gso/cli/imports.py b/gso/cli/imports.py index cd2e5998dd39ea965f1c476357c1738afeaae372..9afc6efbdf4cb47a7f38bb002ba94ae7d7c2b115 100644 --- a/gso/cli/imports.py +++ b/gso/cli/imports.py @@ -22,7 +22,6 @@ from gso.products.product_blocks.bgp_session import IPFamily 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.service_binding_port import VLAN_ID from gso.products.product_blocks.switch import SwitchModel from gso.products.product_types.nren_l3_core_service import NRENL3CoreServiceType from gso.services.partners import ( @@ -50,6 +49,7 @@ from gso.utils.types.ip_address import ( IPV6Netmask, PortNumber, ) +from gso.utils.types.virtual_identifiers import VLAN_ID app: typer.Typer = typer.Typer() diff --git a/gso/workflows/l2_circuit/create_layer_2_circuit.py b/gso/workflows/l2_circuit/create_layer_2_circuit.py index 59f5983d6dc4ea3890842a31dcca5ba1caf015b9..a3199bc1ad2ea56f6ed62a66f075216fa0e99881 100644 --- a/gso/workflows/l2_circuit/create_layer_2_circuit.py +++ b/gso/workflows/l2_circuit/create_layer_2_circuit.py @@ -43,12 +43,12 @@ def initial_input_generator(product_name: str) -> FormGenerator: initial_user_input = yield CreateLayer2CircuitServicePage class Layer2CircuitSideSelection(BaseModel): - edge_port: active_edge_port_selector( + edge_port: active_edge_port_selector( # type: ignore[valid-type] partner_id=initial_user_input.partner if initial_user_input.partner != geant_partner_id else None ) vlan_id: VLAN_ID - def _vlan_range_field(*, is_tagged: bool) -> VLAN_ID | ReadOnlyField: + def _vlan_range_field(*, is_tagged: bool) -> VLAN_ID: """Return the appropriate field type based on whether the circuit is tagged.""" return VLAN_ID if is_tagged else ReadOnlyField(None, default_type=int) @@ -56,14 +56,14 @@ def initial_input_generator(product_name: str) -> FormGenerator: model_config = ConfigDict(title=f"{product_name} - Configure Edge Ports") vlan_range_label: Label = Field("Please set a VLAN range, bounds including.", exclude=True) - vlan_range_lower_bound: _vlan_range_field( + vlan_range_lower_bound: _vlan_range_field( # type: ignore[valid-type] is_tagged=initial_user_input.layer_2_circuit_type == Layer2CircuitType.TAGGED ) - vlan_range_upper_bound: _vlan_range_field( + vlan_range_upper_bound: _vlan_range_field( # type: ignore[valid-type] is_tagged=initial_user_input.layer_2_circuit_type == Layer2CircuitType.TAGGED ) vlan_divider: Divider = Field(None, exclude=True) - policer_bandwidth: ( + policer_bandwidth: ( # type: ignore[valid-type] BandwidthString if initial_user_input.policer_enabled else ReadOnlyField(None, default_type=str) ) geant_sid: str diff --git a/gso/workflows/l2_circuit/modify_layer_2_circuit.py b/gso/workflows/l2_circuit/modify_layer_2_circuit.py index 21b397f11546de7f5e5d42c72f0790443913ace2..5bb7ba19aea5594d67a8980d874fd40a46082c01 100644 --- a/gso/workflows/l2_circuit/modify_layer_2_circuit.py +++ b/gso/workflows/l2_circuit/modify_layer_2_circuit.py @@ -27,7 +27,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator: class ModifyL2CircuitForm(FormPage): model_config = ConfigDict(title=f"Modify {product_name}") tt_number: TTNumber - partner: ReadOnlyField(get_partner_by_id(subscription.customer_id).name, default_type=str) + partner: ReadOnlyField(get_partner_by_id(subscription.customer_id).name, default_type=str) # type: ignore[valid-type] divider: Divider = Field(None, exclude=True) layer_2_circuit_type: Layer2CircuitType = subscription.layer_2_circuit.layer_2_circuit_type @@ -40,27 +40,27 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator: vlan_range_label: Label = Field("Please set a VLAN range, bounds including.", exclude=True) if layer_2_circuit_input.layer_2_circuit_type == Layer2CircuitType.TAGGED: - vlan_range_lower_bound: VLAN_ID = subscription.layer_2_circuit.vlan_range_lower_bound - vlan_range_upper_bound: VLAN_ID = subscription.layer_2_circuit.vlan_range_upper_bound + vlan_range_lower_bound: VLAN_ID = subscription.layer_2_circuit.vlan_range_lower_bound # type: ignore[assignment] + vlan_range_upper_bound: VLAN_ID = subscription.layer_2_circuit.vlan_range_upper_bound # type: ignore[assignment] else: - vlan_range_lower_bound: ReadOnlyField(None, default_type=int) - vlan_range_upper_bound: ReadOnlyField(None, default_type=int) + vlan_range_lower_bound: ReadOnlyField(None, default_type=int) # type: ignore[no-redef, valid-type] + vlan_range_upper_bound: ReadOnlyField(None, default_type=int) # type: ignore[no-redef, valid-type] vlan_divider: Divider = Field(None, exclude=True) if layer_2_circuit_input.policer_enabled: - policer_bandwidth: BandwidthString = subscription.layer_2_circuit.bandwidth + policer_bandwidth: BandwidthString = subscription.layer_2_circuit.bandwidth # type: ignore[assignment] else: - policer_bandwidth: ReadOnlyField(None, default_type=str) + policer_bandwidth: ReadOnlyField(None, default_type=str) # type: ignore[no-redef, valid-type] policer_divider: Divider = Field(None, exclude=True) - layer_2_circuit_side_a: ReadOnlyField( + layer_2_circuit_side_a: ReadOnlyField( # type: ignore[valid-type] EdgePort.from_subscription( subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.edge_port.owner_subscription_id ).description, default_type=str, ) side_divider: Divider = Field(None, exclude=True) - layer_2_circuit_side_b: ReadOnlyField( + layer_2_circuit_side_b: ReadOnlyField( # type: ignore[valid-type] EdgePort.from_subscription( subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.edge_port.owner_subscription_id ).description, @@ -80,7 +80,7 @@ def modify_layer_2_circuit_subscription( vlan_range_upper_bound: VLAN_ID | None, policer_enabled: bool, # noqa: FBT001 policer_bandwidth: BandwidthString | None, -): +) -> dict: """Update the Layer 2 Circuit subscription with the new values.""" subscription.layer_2_circuit.layer_2_circuit_type = layer_2_circuit_type subscription.layer_2_circuit.vlan_range_lower_bound = vlan_range_lower_bound diff --git a/test/workflows/l2_circuit/test_create_layer_2_circuit.py b/test/workflows/l2_circuit/test_create_layer_2_circuit.py index cc1d33dc80516fb8e0a5600423e7d86e5ac3b62a..7550d969a342dbdd16e30ede7a4ae3a39f9520a9 100644 --- a/test/workflows/l2_circuit/test_create_layer_2_circuit.py +++ b/test/workflows/l2_circuit/test_create_layer_2_circuit.py @@ -15,8 +15,6 @@ def layer_2_circuit_input(faker, partner_factory, edge_port_subscription_factory edge_port_a = edge_port_subscription_factory(partner=partner) edge_port_b = edge_port_subscription_factory(partner=partner) policer_enabled = faker.boolean() - - return [ {"product": product_id}, { @@ -35,9 +33,8 @@ def layer_2_circuit_input(faker, partner_factory, edge_port_subscription_factory }, ] -@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() def test_create_layer_2_circuit_success( layer_2_circuit_service_type, @@ -46,8 +43,7 @@ def test_create_layer_2_circuit_success( partner_factory, data_config_filename, ): - - result, process_stat, step_log = run_workflow("create_layer_2_circuit", layer_2_circuit_input) + result, _, _ = run_workflow("create_layer_2_circuit", layer_2_circuit_input) assert_complete(result) state = extract_state(result) subscription = Layer2Circuit.from_subscription(state["subscription_id"])