From a8a9d82b38dae6782b3bd24333c7e9b334018629 Mon Sep 17 00:00:00 2001 From: Hans Trompert Date: Thu, 11 Sep 2025 15:21:26 +0200 Subject: [PATCH] SURF NSISTP implementation to be used as basis --- products/product_blocks/sn8_nsistp.py | 46 ++++++ products/product_types/sn8_nsistp.py | 37 +++++ workflows/nsistp/__init__.py | 0 workflows/nsistp/sn8/__init__.py | 0 workflows/nsistp/sn8/create_nsistp.py | 122 +++++++++++++++ workflows/nsistp/sn8/modify_nsistp.py | 157 +++++++++++++++++++ workflows/nsistp/sn8/shared/__init__.py | 0 workflows/nsistp/sn8/shared/forms.py | 190 +++++++++++++++++++++++ workflows/nsistp/sn8/terminate_nsistp.py | 72 +++++++++ workflows/nsistp/sn8/validate_nsistp.py | 31 ++++ 10 files changed, 655 insertions(+) create mode 100644 products/product_blocks/sn8_nsistp.py create mode 100644 products/product_types/sn8_nsistp.py create mode 100644 workflows/nsistp/__init__.py create mode 100644 workflows/nsistp/sn8/__init__.py create mode 100644 workflows/nsistp/sn8/create_nsistp.py create mode 100644 workflows/nsistp/sn8/modify_nsistp.py create mode 100644 workflows/nsistp/sn8/shared/__init__.py create mode 100644 workflows/nsistp/sn8/shared/forms.py create mode 100644 workflows/nsistp/sn8/terminate_nsistp.py create mode 100644 workflows/nsistp/sn8/validate_nsistp.py diff --git a/products/product_blocks/sn8_nsistp.py b/products/product_blocks/sn8_nsistp.py new file mode 100644 index 0000000..560aa5d --- /dev/null +++ b/products/product_blocks/sn8_nsistp.py @@ -0,0 +1,46 @@ +# Copyright 2019-2024 SURF. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from pydantic import computed_field + +from orchestrator.domain.base import ProductBlockModel, SubscriptionModel +from orchestrator.types import SubscriptionLifecycle +from surf.products.product_blocks.sap_sn8 import Sn8ServiceAttachPointBlock, Sn8ServiceAttachPointBlockInactive + + +class NsistpBlockInactive(ProductBlockModel, product_block_name="NSISTP Service Settings"): + sap: Sn8ServiceAttachPointBlockInactive + topology: str | None = None + stp_id: str | None = None + stp_description: str | None = None + is_alias_in: str | None = None + is_alias_out: str | None = None + expose_in_topology: bool = False + bandwidth: int | None = None + + +class NsistpBlock(NsistpBlockInactive, lifecycle=[SubscriptionLifecycle.ACTIVE, SubscriptionLifecycle.PROVISIONING]): + sap: Sn8ServiceAttachPointBlock + topology: str + stp_id: str + stp_description: str | None = None + is_alias_in: str | None = None + is_alias_out: str | None = None + expose_in_topology: bool = False + bandwidth: int | None = None + + @computed_field # type: ignore[prop-decorator] + @property + def title(self) -> str: + subscription = SubscriptionModel.from_subscription(self.sap.owner_subscription_id) + return f"{self.tag} {self.topology} {self.stp_id} {subscription.description} vlan {self.sap.vlanrange}" diff --git a/products/product_types/sn8_nsistp.py b/products/product_types/sn8_nsistp.py new file mode 100644 index 0000000..81af8f5 --- /dev/null +++ b/products/product_types/sn8_nsistp.py @@ -0,0 +1,37 @@ +# Copyright 2019-2024 SURF. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from nwastdlib.vlans import VlanRanges +from orchestrator.domain.base import SubscriptionModel +from orchestrator.types import SubscriptionLifecycle +from surf.products.product_blocks.sn8_nsistp import NsistpBlock, NsistpBlockInactive +from surf.products.product_types.fixed_input_types import Domain + + +class NsistpInactive(SubscriptionModel, is_base=True): + domain: Domain + settings: NsistpBlockInactive + + +class NsistpProvisioning(NsistpInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]): + domain: Domain + settings: NsistpBlock + + +class Nsistp(NsistpProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]): + domain: Domain + settings: NsistpBlock + + @property + def vlan_range(self) -> VlanRanges: + return self.settings.sap.vlanrange diff --git a/workflows/nsistp/__init__.py b/workflows/nsistp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/workflows/nsistp/sn8/__init__.py b/workflows/nsistp/sn8/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/workflows/nsistp/sn8/create_nsistp.py b/workflows/nsistp/sn8/create_nsistp.py new file mode 100644 index 0000000..795eec9 --- /dev/null +++ b/workflows/nsistp/sn8/create_nsistp.py @@ -0,0 +1,122 @@ +# Copyright 2019-2024 SURF. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from typing import TypeAlias, cast +from uuid import UUID + +from pydantic import ConfigDict, model_validator +from pydantic_forms.types import FormGenerator, State, UUIDstr + +from orchestrator.forms import FormPage +from orchestrator.forms.validators import CustomerId, Divider, Label, ListOfOne +from orchestrator.targets import Target +from orchestrator.types import SubscriptionLifecycle +from orchestrator.workflow import StepList, begin, step +from orchestrator.workflows.steps import store_process_subscription +from surf.forms.validator.bandwidth import ServiceSpeed +from surf.forms.validator.service_port import ServicePort +from surf.forms.validators import JiraTicketId +from surf.products.product_types.sn8_nsistp import NsistpInactive, NsistpProvisioning +from surf.products.services.subscription import subscription_description +from surf.workflows.nsistp.sn8.shared.forms import ( + IsAlias, + StpDescription, + StpId, + Topology, + nsistp_fill_sap, + nsistp_service_port, + validate_both_aliases_empty_or_not, +) +from surf.workflows.shared.summary_form import base_summary +from surf.workflows.workflow import create_workflow + + +def initial_input_form_generator(product_name: str) -> FormGenerator: + FormNsistpServicePort: TypeAlias = cast(type[ServicePort], nsistp_service_port()) + + SingleServicePort = ListOfOne[FormNsistpServicePort] + + class CreateNsiStpForm(FormPage): + model_config = ConfigDict(title=product_name) + + customer_id: CustomerId + ticket_id: JiraTicketId + + label_nsistp_settings: Label + divider: Divider + + service_ports: SingleServicePort + + topology: Topology + + stp_id: StpId + stp_description: StpDescription | None = None + + is_alias_in: IsAlias | None = None + is_alias_out: IsAlias | None = None + + expose_in_topology: bool = True + + bandwidth_info: ServiceSpeed + + @model_validator(mode="after") + def validate_is_alias_in_out(self) -> "CreateNsiStpForm": + validate_both_aliases_empty_or_not(self.is_alias_in, self.is_alias_out) + return self + + user_input = yield CreateNsiStpForm + + user_input_dict = user_input.model_dump() + yield from base_summary(product_name, user_input_dict) + + return user_input_dict + + +@step("Create subscription") +def construct_nsistp_subscription( + customer_id: UUIDstr, + product: UUID, + service_ports: list[dict], + topology: str, + stp_id: str, + stp_description: str | None, + is_alias_in: str, + is_alias_out: str, + expose_in_topology: bool, + bandwidth_info: int, +) -> State: + nsistp = NsistpInactive.from_product_id(product, customer_id) + + settings = nsistp.settings + settings.topology = topology + settings.stp_id = stp_id + settings.stp_description = stp_description + settings.is_alias_in = is_alias_in + settings.is_alias_out = is_alias_out + settings.expose_in_topology = expose_in_topology + settings.bandwidth = bandwidth_info + + nsistp_fill_sap(nsistp, service_ports) + + nsistp = NsistpProvisioning.from_other_lifecycle(nsistp, SubscriptionLifecycle.PROVISIONING) + nsistp.description = subscription_description(nsistp) + + return { + "subscription": nsistp, + "subscription_id": nsistp.subscription_id, + "subscription_description": nsistp.description, + } + + +@create_workflow("Create NSISTP", initial_input_form=initial_input_form_generator) +def create_nsistp() -> StepList: + return begin >> construct_nsistp_subscription >> store_process_subscription(Target.CREATE) diff --git a/workflows/nsistp/sn8/modify_nsistp.py b/workflows/nsistp/sn8/modify_nsistp.py new file mode 100644 index 0000000..aee34af --- /dev/null +++ b/workflows/nsistp/sn8/modify_nsistp.py @@ -0,0 +1,157 @@ +# Copyright 2019-2024 SURF. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from functools import partial +from typing import Annotated, TypeAlias, cast +from uuid import UUID + +from more_itertools import flatten +from more_itertools.more import one +from pydantic import AfterValidator, model_validator +from pydantic_forms.types import FormGenerator, State +from pydantic_forms.validators import ReadOnlyField + +from nwastdlib.vlans import VlanRanges +from orchestrator.forms import FormPage +from orchestrator.forms.validators import Divider, Label, ListOfOne +from orchestrator.workflow import StepList, begin, step +from surf.forms.validator.bandwidth import ServiceSpeed +from surf.forms.validator.service_port import ServicePort +from surf.forms.validators import JiraTicketId +from surf.products.product_types.sn8_nsistp import Nsistp, NsistpProvisioning +from surf.products.services.nsistp import nsi_lp_get_by_port_id +from surf.utils.exceptions import PortsValueError +from surf.workflows.nsistp.sn8.shared.forms import ( + IsAlias, + StpDescription, + StpId, + Topology, + nsistp_fill_sap, + nsistp_service_port, + validate_both_aliases_empty_or_not, + validate_stp_id_uniqueness, +) +from surf.workflows.shared.steps import update_subscription_description +from surf.workflows.shared.summary_form import base_summary +from surf.workflows.shared.validate_subscriptions import subscription_update +from surf.workflows.workflow import modify_workflow + + +def validate_service_port_vlan( + current_sp_id: UUID, nsi_lp_vlanrange: VlanRanges, service_ports: list[ServicePort] +) -> list: + return validate_sp_vlan_in_use_by_nsi_lp(service_ports, current_sp_id, nsi_lp_vlanrange) + + +def validate_sp_vlan_in_use_by_nsi_lp( + service_ports: list[ServicePort], current_sp_id: UUID, nsi_lp_vlanrange: VlanRanges +) -> list: + sp = one(service_ports) + nsi_lp_vlanrange_not_used = nsi_lp_vlanrange - sp.vlan + + if nsi_lp_vlanrange_not_used: + if current_sp_id != sp.subscription_id: + raise PortsValueError( + f"Can't change service port when VLAN's are in use by NSI light paths: ({current_sp_id}: {nsi_lp_vlanrange})" + ) + raise PortsValueError(f"VLAN range must include VLAN's currently in use by NSI light paths: {nsi_lp_vlanrange}") + return service_ports + + +def get_vlans_in_use_by_nsi_lp(nsistp: Nsistp) -> VlanRanges: + nsi_lps = nsi_lp_get_by_port_id(nsistp.settings.sap.port_subscription_id) + used_vlans = VlanRanges(flatten(sap.vlanrange for nsi_lp in nsi_lps for sap in nsi_lp.vc.saps)) + used_vlans_list = set(used_vlans) + + return VlanRanges([vlan for vlan in nsistp.vlan_range if vlan in used_vlans_list]) + + +def initial_input_form_generator(subscription_id: UUID) -> FormGenerator: + nsistp = Nsistp.from_subscription(subscription_id) + settings = nsistp.settings + sap = settings.sap + + port_subscription_id = sap.port.owner_subscription_id + current_service_port = {"subscription_id": port_subscription_id, "vlan": str(sap.vlanrange)} + nsi_lp_vlans = get_vlans_in_use_by_nsi_lp(nsistp) + FormNsistpServicePort: TypeAlias = cast(type[ServicePort], nsistp_service_port(current=[current_service_port])) + + ModifyStpId = Annotated[StpId, AfterValidator(partial(validate_stp_id_uniqueness, subscription_id))] + + service_port_validator = AfterValidator(partial(validate_service_port_vlan, port_subscription_id, nsi_lp_vlans)) + ServicePorts = Annotated[ListOfOne[FormNsistpServicePort], service_port_validator] + + class ModifyNsiStpForm(FormPage): + customer_id: ReadOnlyField(UUID(nsistp.customer_id)) # type: ignore + ticket_id: JiraTicketId + + label_nsistp_settings: Label + divider: Divider + + service_ports: ServicePorts = [current_service_port] + + topology: Topology = settings.topology + + stp_id: ModifyStpId = settings.stp_id + stp_description: StpDescription | None = settings.stp_description + + is_alias_in: IsAlias | None = settings.is_alias_in + is_alias_out: IsAlias | None = settings.is_alias_out + + expose_in_topology: bool = settings.expose_in_topology + + bandwidth_info: ServiceSpeed | None = settings.bandwidth + + @model_validator(mode="after") + def validate_is_alias_in_out(self) -> "ModifyNsiStpForm": + validate_both_aliases_empty_or_not(self.is_alias_in, self.is_alias_out) + return self + + before_user_input_dict = ModifyNsiStpForm().model_dump() # type: ignore + user_input = yield ModifyNsiStpForm + user_input_dict = user_input.model_dump() + + yield from base_summary(nsistp.product.name, user_input_dict, old_data=before_user_input_dict) + + return user_input_dict | {"subscription": nsistp} + + +@subscription_update +@step("Update subscription") +def update_subscription( + subscription: NsistpProvisioning, + service_ports: list[dict], + topology: str, + stp_id: str, + stp_description: str, + is_alias_in: str | None, + is_alias_out: str | None, + expose_in_topology: bool, + bandwidth_info: int | None, +) -> State: + nsistp_fill_sap(subscription, service_ports) + + settings = subscription.settings + settings.topology = topology + settings.stp_id = stp_id + settings.stp_description = stp_description + settings.is_alias_in = is_alias_in + settings.is_alias_out = is_alias_out + settings.expose_in_topology = expose_in_topology + settings.bandwidth = bandwidth_info + + return {"subscription": subscription} + + +@modify_workflow("Modify NSISTP", initial_input_form=initial_input_form_generator) +def modify_nsistp() -> StepList: + return begin >> update_subscription >> update_subscription_description diff --git a/workflows/nsistp/sn8/shared/__init__.py b/workflows/nsistp/sn8/shared/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/workflows/nsistp/sn8/shared/forms.py b/workflows/nsistp/sn8/shared/forms.py new file mode 100644 index 0000000..7356b51 --- /dev/null +++ b/workflows/nsistp/sn8/shared/forms.py @@ -0,0 +1,190 @@ +# Copyright 2019-2024 SURF. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +import re +from collections.abc import Iterator +from datetime import datetime +from functools import partial +from typing import Annotated +from uuid import UUID + +from annotated_types import doc +from more_itertools import one +from pydantic import AfterValidator, ValidationInfo +from pydantic_forms.types import State +from sqlalchemy import select + +from orchestrator.db import ProductTable, db +from orchestrator.domain.base import SubscriptionModel +from orchestrator.types import SubscriptionLifecycle +from surf.config.service_port_tags import SN8_PORT_TAGS_AGGSP, SN8_PORT_TAGS_MSC, SN8_PORT_TAGS_SP_ALL +from surf.db import SurfSubscriptionTable +from surf.forms.validator.service_port import service_port +from surf.products.product_types.sn8_nsistp import Nsistp, NsistpInactive +from surf.utils.exceptions import DuplicateValueError, FieldValueError + +TOPOLOGY_REGEX = r"^[-a-z0-9+,.;=_]+$" +STP_ID_REGEX = r"^[-a-z0-9+,.;=_:]+$" +NURN_REGEX = r"^urn:ogf:network:([^:]+):([0-9]+):([a-z0-9+,-.:;_!$()*@~&]*)$" +FQDN_REQEX = ( + r"^(?!.{255}|.{253}[^.])([a-z0-9](?:[-a-z-0-9]{0,61}[a-z0-9])?\.)*[a-z0-9](?:[-a-z0-9]{0,61}[a-z0-9])?[.]?$" +) + + +def nsistp_service_port(current: list[State] | None = None) -> type: + return service_port( + visible_port_mode="tagged", + allowed_tags=SN8_PORT_TAGS_SP_ALL + SN8_PORT_TAGS_AGGSP + SN8_PORT_TAGS_MSC, + current=current, + ) + + +def is_fqdn(hostname: str) -> bool: + return re.match(FQDN_REQEX, hostname, re.IGNORECASE) is not None + + +def valid_date(date: str) -> tuple[bool, str | None]: + def valid_month() -> tuple[bool, str | None]: + month_str = date[4:6] + month = int(month_str) + if month < 1 or month > 12: + return False, f"{month_str} is not a valid month number" + return True, None + + def valid_day() -> tuple[bool, str | None]: + try: + datetime.fromisoformat(f"{date[0:4]}-{date[4:6]}-{date[6:8]}") + except ValueError: + return False, f"`{date}` is not a valid date" + return True, None + + length = len(date) + if length == 4: # year + pass # No checks on reasonable year, so 9999 is allowed + elif length in (6, 8): + valid, message = valid_month() + if not valid: + return valid, message + if length == 8: # year + month + day + return valid_day() + else: + return False, f"date `{date}` has invalid length" + + return True, None + + +def valid_nurn(nurn: str) -> tuple[bool, str | None]: + if not (match := re.match(NURN_REGEX, nurn, re.IGNORECASE)): + return False, "not a valid NSI STP identifier (urn:ogf:network:...)" + + hostname = match.group(1) + if not is_fqdn(hostname): + return False, f"{hostname} is not a valid fqdn" + + date = match.group(2) + valid, message = valid_date(date) + + return valid, message + + +def validate_regex( + regex: str, + message: str, + field: str | None, +) -> str | None: + if field is None: + return field + + if not re.match(regex, field, re.IGNORECASE): + raise FieldValueError(f"{message} must match: {regex}") + + return field + + +def _get_nsistp_subscriptions(subscription_id: UUID | None) -> Iterator[Nsistp]: + query = ( + select(SurfSubscriptionTable.subscription_id) + .join(ProductTable) + .filter( + ProductTable.product_type == "NSISTP", + SurfSubscriptionTable.status == SubscriptionLifecycle.ACTIVE, + SurfSubscriptionTable.subscription_id != subscription_id, + ) + ) + result = db.session.scalars(query).all() + return (Nsistp.from_subscription(subscription_id) for subscription_id in result) + + +def validate_stp_id_uniqueness(subscription_id: UUID | None, stp_id: str, info: ValidationInfo) -> str: + values = info.data + + customer_id = values.get("customer_id") + topology = values.get("topology") + + if customer_id and topology: + + def is_not_unique(nsistp: Nsistp) -> bool: + return ( + nsistp.settings.stp_id.casefold() == stp_id.casefold() + and nsistp.settings.topology.casefold() == topology.casefold() + ) + + subscriptions = _get_nsistp_subscriptions(subscription_id) + if any(is_not_unique(nsistp) for nsistp in subscriptions): + raise DuplicateValueError(f"STP identifier `{stp_id}` already exists for topology `{topology}`") + + return stp_id + + +StpId = Annotated[ + str, + AfterValidator(partial(validate_regex, STP_ID_REGEX, "STP identifier")), + doc("must be unique along the set of NSISTP's in the same TOPOLOGY"), +] + + +def validate_both_aliases_empty_or_not(is_alias_in: str | None, is_alias_out: str | None) -> None: + if bool(is_alias_in) != bool(is_alias_out): + raise FieldValueError("NSI inbound and outbound isAlias should either both have a value or be empty") + + +def validate_nurn(nurn: str | None) -> str | None: + if nurn: + valid, message = valid_nurn(nurn) + if not valid: + raise FieldValueError(message) + + return nurn + + +def nsistp_fill_sap(subscription: NsistpInactive, service_ports: list[dict]) -> None: + sp = one(service_ports) + subscription.settings.sap.vlanrange = sp["vlan"] + # SubscriptionModel can be any type of ServicePort + subscription.settings.sap.port = SubscriptionModel.from_subscription(sp["subscription_id"]).port # type: ignore + + +IsAlias = Annotated[ + str, AfterValidator(validate_nurn), doc("ISALIAS conform https://www.ogf.org/documents/GFD.202.pdf") +] + +StpDescription = Annotated[ + str, + AfterValidator(partial(validate_regex, r"^[^<>&]*$", "STP description")), + doc("STP description may not contain characters from the set [<>&]"), +] + +Topology = Annotated[ + str, + AfterValidator(partial(validate_regex, TOPOLOGY_REGEX, "Topology")), + doc("topology string may only consist of characters from the set [-a-z+,.;=_]"), +] diff --git a/workflows/nsistp/sn8/terminate_nsistp.py b/workflows/nsistp/sn8/terminate_nsistp.py new file mode 100644 index 0000000..dd6f803 --- /dev/null +++ b/workflows/nsistp/sn8/terminate_nsistp.py @@ -0,0 +1,72 @@ +# Copyright 2019-2024 SURF. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from typing import Annotated, TypeAlias +from uuid import UUID + +from more_itertools import flatten +from pydantic import Field, model_validator +from pydantic_forms.types import InputForm, State + +from nwastdlib.vlans import VlanRanges +from orchestrator.forms import FormPage +from orchestrator.forms.validators import DisplaySubscription +from orchestrator.workflow import StepList, begin, step +from surf.forms.validators import JiraTicketId +from surf.products.product_types.nsi_lp import NsiLightPath +from surf.products.product_types.sn8_nsistp import Nsistp +from surf.products.services.nsistp import nsi_lp_get_by_port_id +from surf.workflows.workflow import terminate_workflow + + +@step("Load initial state") +def load_initial_state(subscription: Nsistp) -> State: + return {"subscription": subscription} + + +def get_in_use_by_nsi_lp(nsistp: Nsistp) -> list[NsiLightPath]: + nsi_lps_by_port = nsi_lp_get_by_port_id(nsistp.settings.sap.port_subscription_id) + + def check_in_use_by_nsi_lp(nsi_lp: NsiLightPath) -> bool: + nsi_lp_vlans = VlanRanges(flatten(sap.vlanrange for sap in nsi_lp.vc.saps)) + return (nsi_lp_vlans - nsistp.vlan_range) != nsi_lp_vlans + + return [nsi_lp for nsi_lp in nsi_lps_by_port if check_in_use_by_nsi_lp(nsi_lp)] + + +def validate_not_in_use_by_nsi_lp(subscription_id: UUID) -> None: + nsistp = Nsistp.from_subscription(subscription_id) + if in_use_by_nsi_lps := get_in_use_by_nsi_lp(nsistp): + in_use_by_nsi_lp_ids = ",".join(str(nsi_lp.subscription_id) for nsi_lp in in_use_by_nsi_lps) + raise ValueError( + f"NSISTP cannot be removed with more than 1 vlan in use by NSILPs, NSILP's that still use vlans of NSISTP ({in_use_by_nsi_lp_ids})." + ) + + +def terminate_initial_input_form_generator(subscription_id: UUID) -> InputForm: + SubscriptionId: TypeAlias = Annotated[DisplaySubscription, Field(subscription_id)] + + class TerminateForm(FormPage): + subscription_id: SubscriptionId + ticket_id: JiraTicketId | None = None + + @model_validator(mode="after") + def check_not_in_use_by_nsi_lp(self) -> "TerminateForm": + validate_not_in_use_by_nsi_lp(self.subscription_id) + return self + + return TerminateForm + + +@terminate_workflow("Terminate NSISTP", initial_input_form=terminate_initial_input_form_generator) +def terminate_nsistp() -> StepList: + return begin >> load_initial_state diff --git a/workflows/nsistp/sn8/validate_nsistp.py b/workflows/nsistp/sn8/validate_nsistp.py new file mode 100644 index 0000000..20c986c --- /dev/null +++ b/workflows/nsistp/sn8/validate_nsistp.py @@ -0,0 +1,31 @@ +# Copyright 2019-2024 SURF. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from pydantic_forms.types import State + +from orchestrator.workflow import StepList, begin, step +from surf.products.product_types.sn8_nsistp import Nsistp +from surf.workflows.helpers import validate_subscription_description, validate_subscription_status_is_active +from surf.workflows.workflow import validate_workflow + + +@step("Check data in coredb") +def check_core_db(subscription: Nsistp) -> State: + validate_subscription_status_is_active(subscription) + validate_subscription_description(subscription) + + return {"check_core_db": True} + + +@validate_workflow("Validate NSISTP") +def validate_nsistp() -> StepList: + return begin >> check_core_db