Skip to content

Modify layer 2 circuit

A modification workflow for a Layer 2 Circuit subscription.

initial_input_form_generator(subscription_id)

Get input from the operator about the modifications to be made to a Layer 2 Circuit subscription.

Source code in gso/workflows/l2_circuit/modify_layer_2_circuit.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
    """Get input from the operator about the modifications to be made to a Layer 2 Circuit subscription."""
    subscription = Layer2Circuit.from_subscription(subscription_id)
    product_name = subscription.product.name

    class ModifyL2CircuitForm(FormPage):
        model_config = ConfigDict(title=f"Modify {product_name}")

        tt_number: TTNumber
        divider: Divider = Field(None, exclude=True)

        # FIXME: str workaround for pre-fill in type hint below
        layer_2_circuit_type: Layer2CircuitType | str = subscription.layer_2_circuit.layer_2_circuit_type
        policer_enabled: bool = subscription.layer_2_circuit.policer_enabled
        custom_service_name: str | None = subscription.layer_2_circuit.custom_service_name

        label: Label = Field("Should this workflow execute Ansible playbooks on routers?", exclude=True)
        run_ansible_steps: bool = True

    layer_2_circuit_input = yield ModifyL2CircuitForm

    class ModifyLayer2CircuitServiceSidesPage(SubmitFormPage):
        model_config = ConfigDict(title=f"{product_name} - Configure Edge Ports")

        vlan_range_label: Label = Field("Please set a VLAN range, bounds including.", exclude=True)
        if layer_2_circuit_input.layer_2_circuit_type == Layer2CircuitType.VLAN:
            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]

        vlan_divider: Divider = Field(None, exclude=True)
        if layer_2_circuit_input.policer_enabled:
            policer_bandwidth: BandwidthString = subscription.layer_2_circuit.bandwidth  # type: ignore[assignment]
            policer_burst_rate: BandwidthString = subscription.layer_2_circuit.policer_burst_rate  # type: ignore[assignment]
            policer_divider: Divider = Field(None, exclude=True)

        class L2CircuitSideA(BaseModel):
            partner: partner_choice(form_title="Partner for Side A") = (  # type: ignore[valid-type]
                subscription.layer_2_circuit.layer_2_circuit_sides[0].partner_id
            )
            # TODO: Uncomment when is_primary is needed
            # is_primary: bool = subscription.layer_2_circuit.layer_2_circuit_sides[0].is_primary  # noqa: ERA001
            if layer_2_circuit_input.layer_2_circuit_type == Layer2CircuitType.ETHERNET:
                vlan_id: VLAN_ID = subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.vlan_id  # type: ignore[assignment]
            edge_port: read_only_field(  # type: ignore[valid-type]
                EdgePort.from_subscription(
                    subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.edge_port.owner_subscription_id
                ).description,
            )

        class L2CircuitSideB(BaseModel):
            partner: partner_choice(form_title="Partner for Side B") = (  # type: ignore[valid-type]
                subscription.layer_2_circuit.layer_2_circuit_sides[1].partner_id
            )
            # TODO: Uncomment when is_primary is needed
            # is_primary: bool = subscription.layer_2_circuit.layer_2_circuit_sides[1].is_primary  # noqa: ERA001
            if layer_2_circuit_input.layer_2_circuit_type == Layer2CircuitType.ETHERNET:
                vlan_id: VLAN_ID = subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.vlan_id  # type: ignore[assignment]
            edge_port: read_only_field(  # type: ignore[valid-type]
                EdgePort.from_subscription(
                    subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.edge_port.owner_subscription_id
                ).description,
            )

        layer_2_circuit_side_a: L2CircuitSideA
        side_divider: Divider = Field(None, exclude=True)
        layer_2_circuit_side_b: L2CircuitSideB

        # TODO: Uncomment when is_primary is added back to the form and make all is_primary required in all access
        # @model_validator(mode="after")
        # def only_one_side_can_be_primary(self) -> Self:
        #     """Ensure that both sides do not have is_primary set to True."""
        #     if self.layer_2_circuit_side_a.is_primary ^ self.layer_2_circuit_side_b.is_primary:
        #         msg = "Only one side of the Layer 2 Circuit can be marked as primary."  # noqa: ERA001
        #         raise ValueError(msg)  # noqa: ERA001
        #     return self  # noqa: ERA001

    layer_2_circuit_sides = yield ModifyLayer2CircuitServiceSidesPage

    return (
        {
            "product_name": product_name,
            # The four values of `None` are required such that the state will contain the required keys.
            # If they are set in the input form by the user, they will be overwritten by the data coming from the form.
            "policer_bandwidth": None,
            "policer_burst_rate": None,
            "vlan_range_upper_bound": None,
            "vlan_range_lower_bound": None,
        }
        | layer_2_circuit_input.model_dump()
        | layer_2_circuit_sides.model_dump()
    )

modify_layer_2_circuit_subscription(subscription, layer_2_circuit_type, vlan_range_lower_bound, vlan_range_upper_bound, policer_enabled, policer_bandwidth, policer_burst_rate, custom_service_name, layer_2_circuit_side_a, layer_2_circuit_side_b)

Update the Layer 2 Circuit subscription with the new values.

Source code in gso/workflows/l2_circuit/modify_layer_2_circuit.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
@step("Update Layer 2 Circuit subscription")
def modify_layer_2_circuit_subscription(
    subscription: Layer2Circuit,
    layer_2_circuit_type: Layer2CircuitType,
    vlan_range_lower_bound: VLAN_ID | None,
    vlan_range_upper_bound: VLAN_ID | None,
    policer_enabled: bool,  # noqa: FBT001
    policer_bandwidth: BandwidthString | None,
    policer_burst_rate: BandwidthString | None,
    custom_service_name: str | None,
    layer_2_circuit_side_a: dict,
    layer_2_circuit_side_b: dict,
) -> 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
    subscription.layer_2_circuit.vlan_range_upper_bound = vlan_range_upper_bound
    subscription.layer_2_circuit.policer_enabled = policer_enabled
    subscription.layer_2_circuit.bandwidth = policer_bandwidth
    subscription.layer_2_circuit.policer_burst_rate = policer_burst_rate
    subscription.layer_2_circuit.custom_service_name = custom_service_name

    # Update `partner_id` and `is_primary` for each side
    subscription.layer_2_circuit.layer_2_circuit_sides[0].partner_id = layer_2_circuit_side_a["partner"]
    subscription.layer_2_circuit.layer_2_circuit_sides[0].is_primary = layer_2_circuit_side_a.get(
        "is_primary", False
    )  # Default to False when is_primary is not in form
    subscription.layer_2_circuit.layer_2_circuit_sides[1].partner_id = layer_2_circuit_side_b["partner"]
    subscription.layer_2_circuit.layer_2_circuit_sides[1].is_primary = layer_2_circuit_side_b.get(
        "is_primary", False
    )  # Default to False when is_primary is not in form

    if layer_2_circuit_type == Layer2CircuitType.VLAN:
        subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.vlan_id = vlan_range_lower_bound
        subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.vlan_id = vlan_range_lower_bound
    else:
        subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.vlan_id = layer_2_circuit_side_a.get("vlan_id")
        subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.vlan_id = layer_2_circuit_side_b.get("vlan_id")
    for layer_2_circuit_side in subscription.layer_2_circuit.layer_2_circuit_sides:
        layer_2_circuit_side.sbp.is_tagged = layer_2_circuit_type == Layer2CircuitType.VLAN

    partner_a = get_partner_by_id(layer_2_circuit_side_a["partner"]).name
    partner_b = get_partner_by_id(layer_2_circuit_side_b["partner"]).name
    subscription.description = (
        f"{subscription.product.name} - {subscription.layer_2_circuit.custom_service_name} ({partner_a} - {partner_b})"
    )

    return {"subscription": subscription}

modify_layer_2_circuit()

Modify an existing Layer 2 Circuit service subscription.

Source code in gso/workflows/l2_circuit/modify_layer_2_circuit.py 
172
173
174
175
176
177
178
179
180
181
182
@modify_workflow("Modify Layer 2 Circuit Service", initial_input_form=initial_input_form_generator)
def modify_layer_2_circuit() -> StepList:
    """Modify an existing Layer 2 Circuit service subscription."""
    return (
        begin
        >> modify_layer_2_circuit_subscription
        >> run_ansible_steps(generate_fqdn_list)
        >> run_ansible_steps(annotate_edge_ports_with_partner_names)
        >> run_ansible_steps(lso_interaction(provision_l2circuit_dry))
        >> run_ansible_steps(lso_interaction(provision_l2circuit_real))
    )