Initial cleanup: remove all trunk leftovers

geant_service_orchestrator/workflows/trunk_config/terminate_trunk_config.py

-from orchestrator.forms import FormPage
-from orchestrator.forms.validators import Label
-from orchestrator.targets import Target
-from orchestrator.types import InputForm, SubscriptionLifecycle, UUIDstr
-from orchestrator.workflow import done, init, step, workflow
-from orchestrator.workflows.steps \
-    import resync, set_status, store_process_subscription, unsync
-from orchestrator.workflows.utils import wrap_modify_initial_input_form
-
-from products import TrunkConfig
-
-
-def initial_input_form_generator(
-        subscription_id: UUIDstr,
-        organisation: UUIDstr) -> InputForm:
-    subscription = TrunkConfig.from_subscription(subscription_id)
-
-    class TerminateForm(FormPage):
-        are_you_sure: Label = \
-            "Are you sure you want to remove " \
-            f"{subscription.description}?"  # type:ignore
-
-    return TerminateForm
-
-
-def _deprovision_in_gap(trunk_config_id: int) -> int:
-    pass
-
-
-@step("Deprovision trunk config")
-def deprovision_trunk_config(subscription: TrunkConfig) -> None:
-    _deprovision_in_gap(subscription.trunk_config.trunk_config_id)
-
-
-@workflow(
-    "Terminate trunk config",
-    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),  # noqa: E501
-    target=Target.TERMINATE,
-)
-def terminate_trunk_config():
-
-    return (
-        init
-        >> store_process_subscription(Target.TERMINATE)
-        >> unsync
-        >> deprovision_trunk_config
-        >> set_status(SubscriptionLifecycle.TERMINATED)
-        >> resync
-        >> done
-    )

geant_service_orchestrator/workflows/trunk_config_common/create_trunk_config_common.py

-from random import randint
-from typing import List
-from uuid import uuid4
-
-from orchestrator.db.models import ProductTable, SubscriptionTable
-from orchestrator.forms import FormPage
-from orchestrator.forms.validators import Choice, choice_list
-from orchestrator.targets import Target
-from orchestrator.types \
-    import FormGenerator, State, SubscriptionLifecycle, UUIDstr
-from orchestrator.workflow import done, init, step, workflow
-from orchestrator.workflows.steps \
-    import resync, set_status, store_process_subscription
-from orchestrator.workflows.utils import wrap_create_initial_input_form
-
-from geant_service_orchestrator.products.product_types.trunk_config_common \
-    import TrunkConfigCommonInactive, TrunkConfigCommonProvisioning
-from geant_service_orchestrator.products.product_types.trunk_config \
-    import TrunkConfig
-
-
-def trunk_config_selector() -> list:
-    trunk_config_subscriptions = {}
-    for trunk_config_id, trunk_config_description in (
-        SubscriptionTable.query.join(ProductTable)
-        .filter(
-            ProductTable.product_type == "TrunkConfig",
-            SubscriptionTable.status == "active",
-        )
-        .with_entities(
-            SubscriptionTable.subscription_id,
-            SubscriptionTable.description)
-        .all()
-    ):
-        trunk_config_subscriptions[
-            str(trunk_config_id)] = trunk_config_description
-
-    return choice_list(
-        Choice("TrunkConfigEnum", zip(
-            trunk_config_subscriptions.keys(),
-            trunk_config_subscriptions.items())),  # type:ignore
-        min_items=1,
-        max_items=1,
-    )
-
-
-def initial_input_form_generator(product_name: str) -> FormGenerator:
-    class CreateTrunkConfigCommonForm(FormPage):
-        class Config:
-            title = product_name
-        speed: str
-        is_leased_line: bool
-        isis_metric: int
-        minimum_links: int
-        trunk_config_ids: trunk_config_selector()  # type:ignore
-
-    user_input = yield CreateTrunkConfigCommonForm
-    return user_input.dict()
-
-
-def _provision_in_gap() -> int:
-
-    return randint(0, 2147483648)
-
-
-@step("Create subscription")
-def create_subscription(product: UUIDstr) -> State:
-    subscription \
-        = TrunkConfigCommonInactive.from_product_id(product, uuid4())
-
-    return {
-        "subscription": subscription,
-        "subscription_id": subscription.subscription_id
-    }
-
-
-@step("Initialize subscription")
-def initialize_subscription(
-        subscription: TrunkConfigCommonInactive,
-        speed: str,
-        is_leased_line: bool,
-        isis_metric: int,
-        minimum_links: int,
-        trunk_config_ids: List[str]) -> State:
-    subscription.trunk_config_common.speed = speed
-    subscription.trunk_config_common.is_leased_line = is_leased_line
-    subscription.trunk_config_common.isis_metric = isis_metric
-    subscription.trunk_config_common.minimum_links = minimum_links
-    subscription.trunk_config_common.trunk_config \
-        = TrunkConfig.from_subscription(
-            trunk_config_ids[0]).trunk_config
-    subscription.description = \
-        "TrunkConfigCommon from TrunkConfig " \
-        f"{subscription.trunk_config_common.trunk_config.trunk_config_id}"
-    subscription = TrunkConfigCommonProvisioning.from_other_lifecycle(
-        subscription, SubscriptionLifecycle.PROVISIONING)
-
-    return {"subscription": subscription}
-
-
-@step("Provision trunk config common")
-def provision_trunk_config_common(
-        subscription: TrunkConfigCommonProvisioning) -> State:
-    trunk_config_common_id = _provision_in_gap()
-    subscription.trunk_config_common.trunk_config_common_id \
-        = trunk_config_common_id
-
-    return {"subscription": subscription}
-
-
-@workflow(
-    "Create trunk config common",
-    initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),  # noqa: E501
-    target=Target.CREATE,
-)
-def create_trunk_config_common():
-
-    return (
-        init
-        >> create_subscription
-        >> store_process_subscription(Target.CREATE)
-        >> initialize_subscription
-        >> provision_trunk_config_common
-        >> set_status(SubscriptionLifecycle.ACTIVE)
-        >> resync
-        >> done
-    )

geant_service_orchestrator/workflows/trunk_config_common/terminate_trunk_config_common.py

-from orchestrator.forms import FormPage
-from orchestrator.forms.validators import Label
-from orchestrator.targets import Target
-from orchestrator.types import InputForm, SubscriptionLifecycle, UUIDstr
-from orchestrator.workflow import done, init, step, workflow
-from orchestrator.workflows.steps \
-    import resync, set_status, store_process_subscription, unsync
-from orchestrator.workflows.utils import wrap_modify_initial_input_form
-
-from geant_service_orchestrator.products import TrunkConfigCommon
-
-
-def initial_input_form_generator(
-        subscription_id: UUIDstr,
-        organisation: UUIDstr) -> InputForm:
-    subscription = TrunkConfigCommon.from_subscription(subscription_id)
-
-    class TerminateForm(FormPage):
-        are_you_sure: Label = \
-            "Are you sure you want to remove " \
-            f"{subscription.description}?"  # type:ignore
-
-    return TerminateForm
-
-
-def _deprovision_in_gap(trunk_config_common_id: int) -> int:
-    pass
-
-
-@step("Deprovision trunk config common")
-def deprovision_trunk_config_common(
-        subscription: TrunkConfigCommon) -> None:
-    _deprovision_in_gap(
-        subscription.trunk_config_common.trunk_config_common_id)
-
-
-@workflow(
-    "Terminate trunk config common",
-    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),  # noqa: E501
-    target=Target.TERMINATE,
-)
-def terminate_trunk_config_common():
-
-    return (
-        init
-        >> store_process_subscription(Target.TERMINATE)
-        >> unsync
-        >> deprovision_trunk_config_common
-        >> set_status(SubscriptionLifecycle.TERMINATED)
-        >> resync
-        >> done
-    )

geant_service_orchestrator/workflows/trunk_config_side/create_trunk_config_side.py

-from typing import List, Optional
-from uuid import uuid4
-
-from orchestrator.db.models import ProductTable, SubscriptionTable
-from orchestrator.forms import FormPage
-from orchestrator.forms.validators import Choice, choice_list
-from orchestrator.targets import Target
-from orchestrator.types \
-    import FormGenerator, State, SubscriptionLifecycle, UUIDstr
-from orchestrator.workflow import done, init, step, workflow
-from orchestrator.workflows.steps \
-    import resync, set_status, store_process_subscription
-from orchestrator.workflows.utils import wrap_create_initial_input_form
-
-from geant_service_orchestrator.products.product_types.trunk_config_side \
-    import TrunkConfigSideInactive, TrunkConfigSideProvisioning
-from geant_service_orchestrator.products.product_types.trunk_config \
-    import TrunkConfig
-
-
-def trunk_config_selector() -> list:
-    trunk_config_subscriptions = {}
-    for trunk_config_id, trunk_config_description in (
-        SubscriptionTable.query.join(ProductTable)
-        .filter(
-            ProductTable.product_type == "TrunkConfig",
-            SubscriptionTable.status == "active",
-        )
-        .with_entities(
-            SubscriptionTable.subscription_id,
-            SubscriptionTable.description)
-        .all()
-    ):
-        trunk_config_subscriptions[
-            str(trunk_config_id)] = trunk_config_description
-
-    return choice_list(
-        Choice("TrunkConfigEnum", zip(
-            trunk_config_subscriptions.keys(),
-            trunk_config_subscriptions.items())),  # type:ignore
-        min_items=1,
-        max_items=1,
-    )
-
-
-def member_selector() -> list:
-    interfaces_dict = {
-        "ge-0/0/1": "ge-0/0/1",
-        "ge-0/0/2": "ge-0/0/2",
-        "ge-0/0/3": "ge-0/0/3",
-        "ge-0/0/4": "ge-0/0/4",
-        "ge-0/0/5": "ge-0/0/5"
-    }
-
-    return choice_list(
-        Choice("Member Interfaces", zip(
-            interfaces_dict.keys(),
-            interfaces_dict.items())),  # type:ignore
-        min_items=1,
-        max_items=4,
-        unique_items=True,
-    )
-
-
-def initial_input_form_generator(product_name: str) -> FormGenerator:
-    class CreateTrunkConfigSideForm(FormPage):
-        class Config:
-            title = product_name
-
-        fqdn: str
-        ae_name: str
-        geant_a_sid: str
-        ipv4_address: Optional[str]
-        ipv6_address: Optional[str]
-        members: member_selector()  # type:ignore
-        trunk_config_ids: trunk_config_selector()  # type:ignore
-
-    user_input = yield CreateTrunkConfigSideForm
-
-    return user_input.dict()
-
-
-def _provision_in_gap(fqdn: str) -> int:
-    return abs(hash(fqdn))
-
-
-@step("Create subscription")
-def create_subscription(product: UUIDstr) -> State:
-    subscription \
-        = TrunkConfigSideInactive.from_product_id(product, uuid4())
-
-    return {
-        "subscription": subscription,
-        "subscription_id": subscription.subscription_id
-    }
-
-
-@step("Initialize subscription")
-def initialize_subscription(
-        subscription: TrunkConfigSideInactive,
-        fqdn: str,
-        ae_name: str,
-        geant_a_sid: str,
-        ipv4_address: str,
-        ipv6_address: str,
-        members: list,
-        trunk_config_ids: List[str]) -> State:
-    subscription.trunk_config_side.fqdn = fqdn
-    subscription.trunk_config_side.ae_name = ae_name
-    subscription.trunk_config_side.geant_a_sid = geant_a_sid
-    subscription.trunk_config_side.ipv4_address = ipv4_address
-    subscription.trunk_config_side.ipv6_address = ipv6_address
-    subscription.trunk_config_side.members = members
-    subscription.trunk_config_side.trunk_config \
-        = TrunkConfig.from_subscription(trunk_config_ids[0]).trunk_config
-    subscription.description = \
-        "TrunkConfigSide from TrunkConfig " \
-        f"{subscription.trunk_config_side.trunk_config.trunk_config_id}"
-    subscription = TrunkConfigSideProvisioning.from_other_lifecycle(
-        subscription, SubscriptionLifecycle.PROVISIONING)
-
-    return {"subscription": subscription}
-
-
-@step("Provision trunk config side")
-def provision_trunk_config_side(
-        subscription: TrunkConfigSideProvisioning, fqdn: str) -> State:
-    trunk_config_side_id = _provision_in_gap(fqdn)
-    subscription.trunk_config_side.trunk_config_side_id = trunk_config_side_id
-
-    return {"subscription": subscription}
-
-
-@workflow(
-    "Create trunk config side",
-    initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),  # noqa: E501
-    target=Target.CREATE,
-)
-def create_trunk_config_side():
-
-    return (
-        init
-        >> create_subscription
-        >> store_process_subscription(Target.CREATE)
-        >> initialize_subscription
-        >> provision_trunk_config_side
-        >> set_status(SubscriptionLifecycle.ACTIVE)
-        >> resync
-        >> done
-    )

geant_service_orchestrator/workflows/trunk_config_side/terminate_trunk_config_side.py

-from orchestrator.forms import FormPage
-from orchestrator.forms.validators import Label
-from orchestrator.targets import Target
-from orchestrator.types import InputForm, SubscriptionLifecycle, UUIDstr
-from orchestrator.workflow import done, init, step, workflow
-from orchestrator.workflows.steps \
-    import resync, set_status, store_process_subscription, unsync
-from orchestrator.workflows.utils import wrap_modify_initial_input_form
-
-from geant_service_orchestrator.products import TrunkConfigSide
-
-
-def initial_input_form_generator(
-        subscription_id: UUIDstr,
-        organisation: UUIDstr) -> InputForm:
-    subscription = TrunkConfigSide.from_subscription(subscription_id)
-
-    class TerminateForm(FormPage):
-        are_you_sure: Label = \
-            "Are you sure you want to remove " \
-            f"{subscription.description}?"  # type:ignore
-
-    return TerminateForm
-
-
-def _deprovision_in_gap(trunk_config_side_id: int) -> int:
-    pass
-
-
-@step("Deprovision trunk config side")
-def deprovision_trunk_config_side(
-        subscription: TrunkConfigSide) -> None:
-    _deprovision_in_gap(subscription.trunk_config_side.trunk_config_side_id)
-
-
-@workflow(
-    "Terminate trunk config side",
-    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),  # noqa: E501
-    target=Target.TERMINATE,
-)
-def terminate_trunk_config_side():
-
-    return (
-        init
-        >> store_process_subscription(Target.TERMINATE)
-        >> unsync
-        >> deprovision_trunk_config_side
-        >> set_status(SubscriptionLifecycle.TERMINATED)
-        >> resync
-        >> done
-    )