Skip to content

Create layer 2 circuit

Workflow for creating a new Layer 2 Circuit.

initial_input_generator(product_name)

Gather input from the operator about a new Layer 2 Circuit subscription.

Source code in gso/workflows/l2_circuit/create_layer_2_circuit.py 
 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
def initial_input_generator(product_name: str) -> FormGenerator:
    """Gather input from the operator about a new Layer 2 Circuit subscription."""
    geant_partner_id = get_partner_by_name("GEANT").partner_id

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

        tt_number: TTNumber
        partner: partner_choice() = geant_partner_id  # type: ignore[valid-type]
        divider: Divider = Field(None, exclude=True)

        layer_2_circuit_type: Layer2CircuitType
        policer_enabled: bool = False
        custom_service_name: str | None = None

    initial_user_input = yield CreateLayer2CircuitServicePage

    def _generate_circuit_side_selection_form(*, is_side_a: bool) -> type[BaseModel]:
        """Generate a selector for picking an Edge Port.

        If the service is an ExpressRoute, we limit the choice for Edge Port on side A. Only the Edge Ports which
        have the ExpressRoute Edge port label can be chosen.
        """

        class Layer2CircuitSideSelection(BaseModel):
            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,
                port_label_name=PortLabelName.EXPRESSROUTE_EDGE
                if product_name == Layer2CircuitServiceType.EXPRESSROUTE and is_side_a
                else None,
            )
            if initial_user_input.layer_2_circuit_type != Layer2CircuitType.VLAN:
                vlan_id: VLAN_ID

        return Layer2CircuitSideSelection

    def _vlan_range_field(*, is_vlan: bool) -> VLAN_ID:
        """Return the appropriate field type based on whether the circuit is VLAN."""
        return VLAN_ID if is_vlan else read_only_field(None)  # type: ignore[return-value]

    def _policer_field(*, policer_enabled: bool) -> BandwidthString:
        """Return the appropriate field type based on whether the policer is enabled."""
        return BandwidthString if policer_enabled else read_only_field(None)  # type: ignore[return-value]

    class Layer2CircuitServiceSidesPage(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)
        vlan_range_lower_bound: _vlan_range_field(  # type: ignore[valid-type]
            is_vlan=initial_user_input.layer_2_circuit_type == Layer2CircuitType.VLAN
        )
        vlan_range_upper_bound: _vlan_range_field(  # type: ignore[valid-type]
            is_vlan=initial_user_input.layer_2_circuit_type == Layer2CircuitType.VLAN
        )
        vlan_divider: Divider = Field(None, exclude=True)
        policer_bandwidth: _policer_field(policer_enabled=initial_user_input.policer_enabled)  # type: ignore[valid-type]
        policer_burst_rate: _policer_field(policer_enabled=initial_user_input.policer_enabled)  # type: ignore[valid-type]
        layer_2_circuit_side_a: _generate_circuit_side_selection_form(is_side_a=True)  # type: ignore[valid-type]
        side_divider: Divider = Field(None, exclude=True)
        layer_2_circuit_side_b: _generate_circuit_side_selection_form(is_side_a=False)  # type: ignore[valid-type]

        @model_validator(mode="after")
        def check_unique_sides(self) -> Self:
            if self.layer_2_circuit_side_a.edge_port == self.layer_2_circuit_side_b.edge_port:
                msg = "Both sides of the circuit cannot be connected to the same edge port"
                raise ValueError(msg)
            return self

    layer_2_circuit_input = yield Layer2CircuitServiceSidesPage

    return {"product_name": product_name} | initial_user_input.model_dump() | layer_2_circuit_input.model_dump()

create_subscription(product, partner)

Create a new subscription object in the database.

Source code in gso/workflows/l2_circuit/create_layer_2_circuit.py 
109
110
111
112
113
114
@step("Create subscription")
def create_subscription(product: UUIDstr, partner: str) -> State:
    """Create a new subscription object in the database."""
    subscription = Layer2CircuitInactive.from_product_id(product, partner)

    return {"subscription": subscription, "subscription_id": subscription.subscription_id}

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

Build a subscription object from all user input.

Source code in gso/workflows/l2_circuit/create_layer_2_circuit.py 
117
118
119
120
121
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
@step("Initialize subscription")
def initialize_subscription(
    subscription: Layer2CircuitInactive,
    layer_2_circuit_side_a: dict[str, Any],
    layer_2_circuit_side_b: dict[str, Any],
    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,
) -> State:
    """Build a subscription object from all user input."""
    layer_2_circuit_sides = []
    gs_id = generate_unique_id(prefix="GS")
    for circuit_side_data in [layer_2_circuit_side_a, layer_2_circuit_side_b]:
        sbp = ServiceBindingPortInactive.new(
            uuid4(),
            edge_port=EdgePort.from_subscription(subscription_id=circuit_side_data["edge_port"]).edge_port,
            sbp_type=SBPType.L2,
            vlan_id=circuit_side_data.get("vlan_id", vlan_range_lower_bound),
            gs_id=gs_id,
            is_tagged=layer_2_circuit_type == Layer2CircuitType.VLAN,
            custom_firewall_filters=False,
        )
        layer2_circuit_side = Layer2CircuitSideBlockInactive.new(uuid4(), sbp=sbp)
        layer_2_circuit_sides.append(layer2_circuit_side)
    subscription.layer_2_circuit.layer_2_circuit_sides = layer_2_circuit_sides
    subscription.layer_2_circuit.layer_2_circuit_type = layer_2_circuit_type
    subscription.layer_2_circuit.virtual_circuit_id = generate_unique_vc_id(
        l2c_type=subscription.layer_2_circuit.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
    subscription.description = f"{subscription.product.name} - {subscription.layer_2_circuit.custom_service_name}"

    subscription = Layer2Circuit.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
    fqdn_list = [side.sbp.edge_port.node.router_fqdn for side in subscription.layer_2_circuit.layer_2_circuit_sides]

    return {"subscription": subscription, "fqdn_list": fqdn_list}

create_layer_2_circuit()

Create a new Layer 2 Circuit service subscription.

Source code in gso/workflows/l2_circuit/create_layer_2_circuit.py 
164
165
166
167
168
169
170
171
172
173
174
175
@create_workflow("Create Layer 2 Circuit Service", initial_input_form=initial_input_generator)
def create_layer_2_circuit() -> StepList:
    """Create a new Layer 2 Circuit service subscription."""
    return (
        begin
        >> create_subscription
        >> store_process_subscription()
        >> initialize_subscription
        >> annotate_edge_ports_with_partner_names
        >> lso_interaction(provision_l2circuit_dry)
        >> lso_interaction(provision_l2circuit_real)
    )