Skip to content

Base create placement port

Base functionality for creating a placement port subscription.

initial_input_generator(product_name)

Gather input from the operator about a new placement port subscription.

It is the base input form for IX Port, Private Peer Port, and Transit Provider Port.

Source code in gso/workflows/placement_port/base_create_placement_port.py 
27
28
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
def initial_input_generator(product_name: str) -> FormGenerator:
    """Gather input from the operator about a new placement port subscription.

    It is the base input form for IX Port, Private Peer Port, and Transit Provider Port.
    """
    geant_partner_id = get_partner_by_name("GEANT").partner_id

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

        tt_number: TTNumber
        partner: partner_choice()  # type: ignore[valid-type]

    initial_user_input = yield InitialPlacementPortForm

    class ConfigurePlacementPortForm(SubmitFormPage):
        model_config = ConfigDict(title=f"Configure {product_name}")

        generate_gs_id: bool = True
        gs_id: IMPORTED_GS_ID | None = None
        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
        )
        is_tagged: bool = True
        vlan_id: VLAN_ID | None = None
        divider1: Divider = Field(None, exclude=True)
        ipv4_address: IPv4AddressType | None = None
        ipv4_mask: IPv4Netmask | None = None
        ipv6_address: IPv6AddressType | None = None
        ipv6_mask: IPv6Netmask | None = None
        custom_firewall_filters: bool = False
        divider2: Divider = Field(None, exclude=True)

        @model_validator(mode="after")
        def check_gs_id_required(self) -> Self:
            if not self.generate_gs_id and not self.gs_id:
                error_msg = "GS ID is required when generate_gs_id is False"
                raise ValueError(error_msg)
            return self

    config_input = yield ConfigurePlacementPortForm

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

initialize_subscription(subscription, gs_id, edge_port, generate_gs_id, vlan_id, ipv4_address, ipv4_mask, ipv6_address, ipv6_mask, custom_firewall_filters=None, is_tagged=None)

Initialize the placement port subscription with the provided parameters.

Source code in gso/workflows/placement_port/base_create_placement_port.py 
 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
@step("Initialize subscription")
def initialize_subscription(
    subscription: SubscriptionModel,
    gs_id: str | None,
    edge_port: str,
    generate_gs_id: bool,  # noqa: FBT001
    vlan_id: VLAN_ID | None,
    ipv4_address: IPv4AddressType | None,
    ipv4_mask: IPv4Netmask | None,
    ipv6_address: IPv6AddressType | None,
    ipv6_mask: IPv6Netmask | None,
    custom_firewall_filters: bool | None = None,  # noqa: FBT001
    is_tagged: bool | None = None,  # noqa: FBT001
) -> State:
    """Initialize the placement port subscription with the provided parameters."""
    gs_id = generate_unique_id(prefix="GS") if generate_gs_id else gs_id
    subscription.l3_interface = subscription.l3_interface.new(  # type: ignore[attr-defined]
        uuid4(),
        gs_id=gs_id,
        edge_port=EdgePort.from_subscription(subscription_id=edge_port).edge_port,
        is_tagged=is_tagged,
        vlan_id=vlan_id,
        ipv4_address=ipv4_address,
        ipv4_mask=ipv4_mask,
        ipv6_address=ipv6_address,
        ipv6_mask=ipv6_mask,
        custom_firewall_filters=custom_firewall_filters,
    )

    subscription = SubscriptionModel.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
    subscription.description = f"{subscription.product.name} - {gs_id}"
    partner_name = get_partner_by_id(subscription.customer_id).name

    return {
        "subscription": subscription,
        "partner_name": partner_name,
    }

deploy_l3_port_dry(subscription, partner_name)

Perform a dry run of deploying L3 interface.

Source code in gso/workflows/placement_port/base_create_placement_port.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@step("[DRY RUN] Deploy L3 interface")
def deploy_l3_port_dry(subscription: SubscriptionModel, partner_name: str) -> LSOState:
    """Perform a dry run of deploying L3 interface."""
    extra_vars = {
        "subscription": json.loads(json_dumps(subscription)),
        "partner_name": partner_name,
        "dry_run": True,
        "verb": "deploy",
        "object": "l3_interface",
    }

    return {
        "playbook_name": "gap_ansible/playbooks/l3_interface.yaml",
        "inventory": {"all": {"hosts": subscription.l3_interface.edge_port.node.router_fqdn}},  # type: ignore[attr-defined]
        "extra_vars": extra_vars,
    }