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 
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
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

    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
        )
        vlan_id: VLAN_ID

    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)

    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 ReadOnlyField(None, default_type=str)

    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_tagged=initial_user_input.layer_2_circuit_type == Layer2CircuitType.TAGGED
        )
        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_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: Layer2CircuitSideSelection
        side_divider: Divider = Field(None, exclude=True)
        layer_2_circuit_side_b: Layer2CircuitSideSelection

        @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 
90
91
92
93
94
95
@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 
 98
 99
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
131
132
133
134
135
136
137
138
139
@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["vlan_id"],
            gs_id=gs_id,
            is_tagged=layer_2_circuit_type == Layer2CircuitType.TAGGED,
            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.virtual_circuit_id = generate_unique_vc_id()
    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
    subscription.description = f"{subscription.product.name} - {subscription.layer_2_circuit.virtual_circuit_id}"

    subscription = Layer2Circuit.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)

    return {"subscription": subscription}

create_layer_2_circuit()

Create a new Layer 2 Circuit service subscription.

Source code in gso/workflows/l2_circuit/create_layer_2_circuit.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@workflow(
    "Create Layer 2 Circuit Service",
    initial_input_form=wrap_create_initial_input_form(initial_input_generator),
    target=Target.CREATE,
)
def create_layer_2_circuit() -> StepList:
    """Create a new Layer 2 Circuit service subscription."""
    return (
        begin
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )