Skip to content

Create imported office router

A creation workflow that adds existing office routers to the coreDB.

create_subscription(partner)

Create a new subscription object.

Source code in gso/workflows/office_router/create_imported_office_router.py 
20
21
22
23
24
25
26
27
28
29
30
@step("Create subscription")
def create_subscription(partner: str) -> State:
    """Create a new subscription object."""
    partner_id = get_partner_by_name(partner).partner_id
    product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_OFFICE_ROUTER)
    subscription = ImportedOfficeRouterInactive.from_product_id(product_id, partner_id)

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

initial_input_form_generator()

Generate a form that is filled in using information passed through the API endpoint.

Source code in gso/workflows/office_router/create_imported_office_router.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def initial_input_form_generator() -> FormGenerator:
    """Generate a form that is filled in using information passed through the API endpoint."""

    class ImportOfficeRouter(SubmitFormPage):
        model_config = ConfigDict(title="Import an office router")

        partner: str
        office_router_site: str
        office_router_fqdn: str
        office_router_ts_port: PortNumber
        office_router_lo_ipv4_address: IPv4AddressType
        office_router_lo_ipv6_address: IPv6AddressType

    user_input = yield ImportOfficeRouter

    return user_input.model_dump()

initialize_subscription(subscription, office_router_fqdn, office_router_ts_port, office_router_site, office_router_lo_ipv4_address=None, office_router_lo_ipv6_address=None)

Initialise the office router subscription using input data.

Source code in gso/workflows/office_router/create_imported_office_router.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@step("Initialize subscription")
def initialize_subscription(
    subscription: ImportedOfficeRouterInactive,
    office_router_fqdn: str,
    office_router_ts_port: PortNumber,
    office_router_site: str,
    office_router_lo_ipv4_address: IPv4AddressType | None = None,
    office_router_lo_ipv6_address: IPv6AddressType | None = None,
) -> State:
    """Initialise the office router subscription using input data."""
    subscription.office_router.office_router_ts_port = office_router_ts_port
    site_obj = get_site_by_name(office_router_site).site
    subscription.office_router.office_router_site = site_obj
    subscription.office_router.office_router_fqdn = office_router_fqdn
    subscription.description = f"Office router {office_router_fqdn}"
    subscription.office_router.office_router_lo_ipv4_address = office_router_lo_ipv4_address
    subscription.office_router.office_router_lo_ipv6_address = office_router_lo_ipv6_address
    subscription.office_router.vendor = Vendor.JUNIPER

    return {"subscription": subscription}

create_imported_office_router()

Import an office router without provisioning it.

Source code in gso/workflows/office_router/create_imported_office_router.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@workflow(
    "Import office router",
    initial_input_form=initial_input_form_generator,
    target=Target.CREATE,
)
def create_imported_office_router() -> StepList:
    """Import an office router without provisioning it."""
    return (
        begin
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )