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
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
        partner: ReadOnlyField(get_partner_by_id(subscription.customer_id).name, default_type=str)  # type: ignore[valid-type]
        divider: Divider = Field(None, exclude=True)

        # FIXME: remove str workaround 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]
        else:
            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  # type: ignore[assignment]
            policer_burst_rate: BandwidthString = subscription.layer_2_circuit.policer_burst_rate  # type: ignore[assignment]
        else:
            policer_bandwidth: ReadOnlyField(None, default_type=str)  # type: ignore[no-redef, valid-type]
            policer_burst_rate: ReadOnlyField(None, default_type=str)  # type: ignore[no-redef, valid-type]
        policer_divider: Divider = Field(None, exclude=True)

        class L2CircuitSideA(BaseModel):
            if layer_2_circuit_input.layer_2_circuit_type != Layer2CircuitType.VLAN:
                vlan_id: VLAN_ID = subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.vlan_id  # type: ignore[assignment]
            edge_port: ReadOnlyField(  # type: ignore[valid-type]
                EdgePort.from_subscription(
                    subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.edge_port.owner_subscription_id
                ).description,
                default_type=str,
            )

        class L2CircuitSideB(BaseModel):
            if layer_2_circuit_input.layer_2_circuit_type != Layer2CircuitType.VLAN:
                vlan_id: VLAN_ID = subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.vlan_id  # type: ignore[assignment]
            edge_port: ReadOnlyField(  # type: ignore[valid-type]
                EdgePort.from_subscription(
                    subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.edge_port.owner_subscription_id
                ).description,
                default_type=str,
            )

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

    layer_2_circuit_sides = yield ModifyLayer2CircuitServiceSidesPage

    return {"product_name": product_name} | 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 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@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
    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

    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 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@workflow(
    "Modify Layer 2 Circuit Service",
    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
    target=Target.MODIFY,
)
def modify_layer_2_circuit() -> StepList:
    """Modify an existing Layer 2 Circuit service subscription."""
    run_ansible_steps = conditional(lambda state: state["run_ansible_steps"])

    return (
        begin
        >> store_process_subscription(Target.MODIFY)
        >> unsync
        >> run_ansible_steps(generate_fqdn_list)
        >> run_ansible_steps(extract_partner_name_from_edge_port)
        >> modify_layer_2_circuit_subscription
        >> run_ansible_steps(lso_interaction(provision_l2circuit_dry))
        >> run_ansible_steps(lso_interaction(provision_l2circuit_real))
        >> resync
        >> done
    )