diff options
Diffstat (limited to 'src/onapsdk/sdc/service.py')
| -rw-r--r-- | src/onapsdk/sdc/service.py | 932 |
1 files changed, 932 insertions, 0 deletions
diff --git a/src/onapsdk/sdc/service.py b/src/onapsdk/sdc/service.py new file mode 100644 index 0000000..c866fa4 --- /dev/null +++ b/src/onapsdk/sdc/service.py @@ -0,0 +1,932 @@ +"""Service module.""" +# Copyright 2022 Orange, Deutsche Telekom AG +# +# 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 base64 +import pathlib as Path +import time +from dataclasses import dataclass, field +from enum import Enum +from io import BytesIO, TextIOWrapper +from os import makedirs +from typing import Dict, List, Callable, Iterator, Optional, Type, Union, Any, BinaryIO +from zipfile import ZipFile, BadZipFile + +import oyaml as yaml +from requests import Response + +import onapsdk.constants as const +from onapsdk.exceptions import (ParameterError, RequestError, ResourceNotFound, + StatusError, ValidationError) +from onapsdk.sdc.category_management import ServiceCategory +from onapsdk.sdc.properties import NestedInput, Property +from onapsdk.sdc.sdc_resource import SdcResource +from onapsdk.utils.configuration import (components_needing_distribution, + tosca_path) +from onapsdk.utils.headers_creator import headers_sdc_creator, headers_sdc_artifact_upload +from onapsdk.utils.jinja import jinja_env + + +@dataclass +class VfModule: # pylint: disable=too-many-instance-attributes + """VfModule dataclass.""" + + name: str + group_type: str + model_name: str + model_version_id: str + model_invariant_uuid: str + model_version: str + model_customization_id: str + properties: Iterator[Property] + + +@dataclass +class NodeTemplate: # pylint: disable=too-many-instance-attributes + """Node template dataclass. + + Base class for Vnf, Pnf and Network classes. + """ + + name: str + node_template_type: str + model_name: str + model_version_id: str + model_invariant_id: str + model_version: str + model_customization_id: str + model_instance_name: str + component: "Component" + + @property + def properties(self) -> Iterator["Property"]: + """Node template properties. + + Returns: + Iterator[Property]: Node template properties iterator + + """ + return self.component.properties + + +@dataclass +class Vnf(NodeTemplate): + """Vnf dataclass.""" + + vf_modules: List[VfModule] = field(default_factory=list) + + +@dataclass +class Pnf(NodeTemplate): + """Pnf dataclass.""" + + +class Network(NodeTemplate): # pylint: disable=too-few-public-methods + """Network dataclass.""" + + +class ServiceInstantiationType(Enum): + """Service instantiation type enum class. + + Service can be instantiated using `A-la-carte` or `Macro` flow. + It has to be determined during design time. That class stores these + two values to set during initialization. + + """ + + A_LA_CARTE = "A-la-carte" + MACRO = "Macro" + + +class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too-many-public-methods + """ + ONAP Service Object used for SDC operations. + + Attributes: + name (str): the name of the service. Defaults to "ONAP-test-Service". + identifier (str): the unique ID of the service from SDC. + status (str): the status of the service from SDC. + version (str): the version ID of the service from SDC. + uuid (str): the UUID of the Service (which is different from + identifier, don't ask why...) + distribution_status (str): the status of distribution in the different + ONAP parts. + distribution_id (str): the ID of the distribution when service is + distributed. + distributed (bool): True if the service is distributed + unique_identifier (str): Yet Another ID, just to puzzle us... + + """ + + SERVICE_PATH = "services" + + def __init__(self, name: str = None, version: str = None, sdc_values: Dict[str, str] = None, # pylint: disable=too-many-arguments + resources: List[SdcResource] = None, properties: List[Property] = None, + inputs: List[Union[Property, NestedInput]] = None, + instantiation_type: Optional[ServiceInstantiationType] = \ + None, + category: str = None, role: str = "", function: str = "", service_type: str = ""): + """ + Initialize service object. + + Args: + name (str, optional): the name of the service + version (str, optional): the version of the service + sdc_values (Dict[str, str], optional): dictionary of values + returned by SDC + resources (List[SdcResource], optional): list of SDC resources + properties (List[Property], optional): list of properties to add to service. + None by default. + inputs (List[Union[Property, NestedInput]], optional): list of inputs + to declare for service. It can be both Property or NestedInput object. + None by default. + instantiation_type (ServiceInstantiationType, optional): service instantiation + type. ServiceInstantiationType.A_LA_CARTE by default + category (str, optional): service category name + role (str, optional): service role + function (str, optional): service function. Empty by default + service_type (str, optional): service type. Empty by default + + """ + super().__init__(sdc_values=sdc_values, version=version, properties=properties, + inputs=inputs, category=category) + self.name: str = name or "ONAP-test-Service" + self.distribution_status = None + self.category_name: str = category + self.role: str = role + self.function: str = function + self.service_type: str = service_type + if sdc_values: + self.distribution_status = sdc_values['distributionStatus'] + self.category_name = sdc_values["category"] + self.resources = resources or [] + self._instantiation_type: Optional[ServiceInstantiationType] = instantiation_type + self._distribution_id: str = None + self._distributed: bool = False + self._resource_type: str = "services" + self._tosca_model: bytes = None + self._tosca_template: str = None + self._vnfs: list = None + self._pnfs: list = None + self._networks: list = None + self._vf_modules: list = None + + @classmethod + def get_by_unique_uuid(cls, unique_uuid: str) -> "Service": + """Get the service model using unique uuid. + + Returns: + Service: object with provided unique_uuid + + Raises: + ResourceNotFound: No service with given unique_uuid exists + + """ + services: List["Service"] = cls.get_all() + for service in services: + if service.unique_uuid == unique_uuid: + return service + raise ResourceNotFound("Service with given unique uuid doesn't exist") + + def onboard(self) -> None: + """Onboard the Service in SDC. + + Raises: + StatusError: service has an invalid status + ParameterError: no resources, no properties for service + in DRAFT status + + """ + # first Lines are equivalent for all onboard functions but it's more + # readable + if not self.status: + # equivalent step as in onboard-function in sdc_resource + self.create() + time.sleep(self._time_wait) + self.onboard() + elif self.status == const.DRAFT: + if not any([self.resources, self._properties_to_add]): + raise ParameterError("No resources nor properties were given") + self.declare_resources_and_properties() + self.checkin() + time.sleep(self._time_wait) + self.onboard() + elif self.status == const.CHECKED_IN: + self.certify() + time.sleep(self._time_wait) + self.onboard() + elif self.status == const.CERTIFIED: + self.distribute() + self.onboard() + elif self.status == const.DISTRIBUTED: + self._logger.info("Service %s onboarded", self.name) + else: + self._logger.error("Service has invalid status: %s", self.status) + raise StatusError(self.status) + + @property + def distribution_id(self) -> str: + """Return and lazy load the distribution_id.""" + if not self._distribution_id: + self.load_metadata() + return self._distribution_id + + @distribution_id.setter + def distribution_id(self, value: str) -> None: + """Set value for distribution_id.""" + self._distribution_id = value + + @property + def distributed(self) -> bool: + """Return and lazy load the distributed state.""" + if not self._distributed: + self._check_distributed() + return self._distributed + + @property + def tosca_template(self) -> str: + """Service tosca template file. + + Get tosca template from service tosca model bytes. + + Returns: + str: Tosca template file + + """ + if not self._tosca_template and self.tosca_model: + self._unzip_csar_file(BytesIO(self.tosca_model), + self._load_tosca_template) + return self._tosca_template + + @property + def tosca_model(self) -> bytes: + """Service's tosca model file. + + Send request to get service TOSCA model, + + Returns: + bytes: TOSCA model file bytes + + """ + if not self._tosca_model: + url = "{}/services/{}/toscaModel".format(self._base_url(), + self.identifier) + headers = self.headers.copy() + headers["Accept"] = "application/octet-stream" + self._tosca_model = self.send_message( + "GET", + "Download Tosca Model for {}".format(self.name), + url, + headers=headers).content + return self._tosca_model + + def create_node_template(self, + node_template_type: Type[NodeTemplate], + component: "Component") -> NodeTemplate: + """Create a node template type object. + + The base of the all node template types objects (Vnf, Pnf, Network) is the + same. The difference is only for the Vnf which can have vf modules associated with. + Vf modules could have "vf_module_label" property with"base_template_dummy_ignore" + value. These vf modules should be ignored/ + + Args: + node_template_type (Type[NodeTemplate]): Node template class type + component (Component): Component on which base node template object should be created + + Returns: + NodeTemplate: Node template object created from component + + """ + node_template: NodeTemplate = node_template_type( + name=component.name, + node_template_type=component.tosca_component_name, + model_name=component.component_name, + model_version_id=component.sdc_resource.identifier, + model_invariant_id=component.sdc_resource.unique_uuid, + model_version=component.sdc_resource.version, + model_customization_id=component.customization_uuid, + model_instance_name=self.name, + component=component + ) + if node_template_type is Vnf: + if component.group_instances: + for vf_module in component.group_instances: + if not any([property_def["name"] == "vf_module_label"] and \ + property_def["value"] == "base_template_dummy_ignore" for \ + property_def in vf_module["properties"]): + node_template.vf_modules.append(VfModule( + name=vf_module["name"], + group_type=vf_module["type"], + model_name=vf_module["groupName"], + model_version_id=vf_module["groupUUID"], + model_invariant_uuid=vf_module["invariantUUID"], + model_version=vf_module["version"], + model_customization_id=vf_module["customizationUUID"], + properties=( + Property( + name=property_def["name"], + property_type=property_def["type"], + description=property_def["description"], + value=property_def["value"] + ) for property_def in vf_module["properties"] \ + if property_def["value"] and not ( + property_def["name"] == "vf_module_label" and \ + property_def["value"] == "base_template_dummy_ignore" + ) + ) + )) + return node_template + + def __has_component_type(self, origin_type: str) -> bool: + """Check if any of Service's component type is provided origin type. + + In template generation is checked if Service has some types of components, + based on that blocks are added to the request template. It's not + the best option to get all components to check if at least one with + given type exists for conditional statement. + + Args: + origin_type (str): Type to check if any component exists. + + Returns: + bool: True if service has at least one component with given origin type, + False otherwise + + """ + return any((component.origin_type == origin_type for component in self.components)) + + @property + def has_vnfs(self) -> bool: + """Check if service has at least one VF component.""" + return self.__has_component_type("VF") + + @property + def has_pnfs(self) -> bool: + """Check if service has at least one PNF component.""" + return self.__has_component_type("PNF") + + @property + def has_vls(self) -> bool: + """Check if service has at least one VL component.""" + return self.__has_component_type("VL") + + @property + def vnfs(self) -> Iterator[Vnf]: + """Service Vnfs. + + Load VNFs from components generator. + It creates a generator of the vf modules as well, but without + vf modules which has "vf_module_label" property value equal + to "base_template_dummy_ignore". + + Returns: + Iterator[Vnf]: Vnf objects iterator + + """ + for component in self.components: + if component.origin_type == "VF": + yield self.create_node_template(Vnf, component) + + @property + def pnfs(self) -> Iterator[Pnf]: + """Service Pnfs. + + Load PNFS from components generator. + + Returns: + Iterator[Pnf]: Pnf objects generator + + """ + for component in self.components: + if component.origin_type == "PNF": + yield self.create_node_template(Pnf, component) + + @property + def networks(self) -> Iterator[Network]: + """Service networks. + + Load networks from service's components generator. + + Returns: + Iterator[Network]: Network objects generator + + """ + for component in self.components: + if component.origin_type == "VL": + yield self.create_node_template(Network, component) + + @property + def deployment_artifacts_url(self) -> str: + """Deployment artifacts url. + + Returns: + str: SdcResource Deployment artifacts url + + """ + return (f"{self._base_create_url()}/services/" + f"{self.unique_identifier}/filteredDataByParams?include=deploymentArtifacts") + + @property + def add_deployment_artifacts_url(self) -> str: + """Add deployment artifacts url. + + Returns: + str: Url used to add deployment artifacts + + """ + return (f"{self._base_create_url()}/services/" + f"{self.unique_identifier}/artifacts") + + @property + def properties_url(self) -> str: + """Properties url. + + Returns: + str: SdcResource properties url + + """ + return (f"{self._base_create_url()}/services/" + f"{self.unique_identifier}/filteredDataByParams?include=properties") + + @property + def metadata_url(self) -> str: + """Metadata url. + + Returns: + str: Service metadata url + + """ + return (f"{self._base_create_url()}/services/" + f"{self.unique_identifier}/filteredDataByParams?include=metadata") + + @property + def resource_inputs_url(self) -> str: + """Service inputs url. + + Returns: + str: Service inputs url + + """ + return (f"{self._base_create_url()}/services/" + f"{self.unique_identifier}") + + @property + def set_input_default_value_url(self) -> str: + """Url to set input default value. + + Returns: + str: SDC API url used to set input default value + + """ + return (f"{self._base_create_url()}/services/" + f"{self.unique_identifier}/update/inputs") + + @property + def origin_type(self) -> str: + """Service origin type. + + Value needed for composition. It's used for adding SDC resource + as an another SDC resource component. + For Service that value has to be set to "ServiceProxy". + + Returns: + str: Service resource origin type + + """ + return "ServiceProxy" + + @property + def instantiation_type(self) -> ServiceInstantiationType: + """Service instantiation type. + + One of `ServiceInstantiationType` enum value. + + Returns: + ServiceInstantiationType: Service instantiation type + + """ + if not self._instantiation_type: + if not self.created(): + self._instantiation_type = ServiceInstantiationType.A_LA_CARTE + else: + response: str = self.send_message_json("GET", + f"Get service {self.name} metadata", + self.metadata_url)["metadata"]\ + ["instantiationType"] + self._instantiation_type = ServiceInstantiationType(response) + return self._instantiation_type + + def create(self) -> None: + """Create the Service in SDC if not already existing.""" + self._create("service_create.json.j2", + name=self.name, + instantiation_type=self.instantiation_type.value, + category=self.category, + role=self.role, service_type=self.service_type, function=self.function) + + def declare_resources_and_properties(self) -> None: + """Delcare resources and properties. + + It declares also inputs. + + """ + for resource in self.resources: + self.add_resource(resource) + for property_to_add in self._properties_to_add: + self.add_property(property_to_add) + for input_to_add in self._inputs_to_add: + self.declare_input(input_to_add) + + def checkin(self) -> None: + """Checkin Service.""" + self._verify_lcm_to_sdc(const.DRAFT, const.CHECKIN) + + def submit(self) -> None: + """Really submit the SDC Service.""" + self._verify_lcm_to_sdc(const.CHECKED_IN, const.SUBMIT_FOR_TESTING) + + def start_certification(self) -> None: + """Start Certification on Service.""" + headers = headers_sdc_creator(SdcResource.headers) + self._verify_lcm_to_sdc(const.CHECKED_IN, + const.START_CERTIFICATION, + headers=headers) + + def certify(self) -> None: + """Certify Service in SDC.""" + headers = headers_sdc_creator(SdcResource.headers) + self._verify_lcm_to_sdc(const.CHECKED_IN, + const.CERTIFY, + headers=headers) + + def approve(self) -> None: + """Approve Service in SDC.""" + headers = headers_sdc_creator(SdcResource.headers) + self._verify_approve_to_sdc(const.CERTIFIED, + const.APPROVE, + headers=headers) + + def distribute(self) -> None: + """Apptove Service in SDC.""" + headers = headers_sdc_creator(SdcResource.headers) + self._verify_distribute_to_sdc(const.CERTIFIED, + const.DISTRIBUTE, + headers=headers) + + def redistribute(self) -> None: + """Apptove Service in SDC.""" + headers = headers_sdc_creator(SdcResource.headers) + self._verify_distribute_to_sdc(const.DISTRIBUTED, + const.DISTRIBUTE, + headers=headers) + + def get_tosca(self) -> None: + """Get Service tosca files and save it.""" + url = "{}/services/{}/toscaModel".format(self._base_url(), + self.identifier) + headers = self.headers.copy() + headers["Accept"] = "application/octet-stream" + result = self.send_message("GET", + "Download Tosca Model for {}".format( + self.name), + url, + headers=headers) + if result: + self._create_tosca_file(result) + + def _create_tosca_file(self, result: Response) -> None: + """Create Service Tosca files from HTTP response.""" + csar_filename = "service-{}-csar.csar".format(self.name) + makedirs(tosca_path(), exist_ok=True) + with open((tosca_path() + csar_filename), 'wb') as csar_file: + for chunk in result.iter_content(chunk_size=128): + csar_file.write(chunk) + try: + self._unzip_csar_file(tosca_path() + csar_filename, + self._write_csar_file) + except BadZipFile as exc: + self._logger.exception(exc) + + def _check_distributed(self) -> bool: + """Check if service is distributed and update status accordingly.""" + url = "{}/services/distribution/{}".format(self._base_create_url(), + self.distribution_id) + headers = headers_sdc_creator(SdcResource.headers) + + status = {} + for component in components_needing_distribution(): + status[component] = False + + try: + result = self.send_message_json("GET", + "Check distribution for {}".format( + self.name), + url, + headers=headers) + except ResourceNotFound: + msg = f"No distributions found for {self.name} of {self.__class__.__name__}." + self._logger.debug(msg) + else: + status = self._update_components_status(status, result) + + for state in status.values(): + if not state: + self._distributed = False + return + self._distributed = True + + def _update_components_status(self, status: Dict[str, bool], + result: Response) -> Dict[str, bool]: + """Update components distribution status.""" + distrib_list = result['distributionStatusList'] + self._logger.debug("[SDC][Get Distribution] distrib_list = %s", + distrib_list) + for elt in distrib_list: + status = self._parse_components_status(status, elt) + return status + + def _parse_components_status(self, status: Dict[str, bool], + element: Dict[str, Any]) -> Dict[str, bool]: + """Parse components distribution status.""" + for key in status: + if ((key in element['omfComponentID']) + and (const.DOWNLOAD_OK in element['status'])): + status[key] = True + self._logger.info(("[SDC][Get Distribution] Service " + "distributed in %s"), key) + return status + + def load_metadata(self) -> None: + """Load Metada of Service and retrieve informations.""" + url = "{}/services/{}/distribution".format(self._base_create_url(), + self.identifier) + headers = headers_sdc_creator(SdcResource.headers) + result = self.send_message_json("GET", + "Get Metadata for {}".format( + self.name), + url, + headers=headers) + if ('distributionStatusOfServiceList' in result + and len(result['distributionStatusOfServiceList']) > 0): + # API changed and the latest distribution is not added to the end + # of distributions list but inserted as the first one. + dist_status = result['distributionStatusOfServiceList'][0] + self._distribution_id = dist_status['distributionID'] + + @classmethod + def _get_all_url(cls) -> str: + """ + Get URL for all elements in SDC. + + Returns: + str: the url + + """ + return "{}/{}".format(cls._base_url(), cls._sdc_path()) + + def _really_submit(self) -> None: + """Really submit the SDC Service in order to enable it.""" + result = self._action_to_sdc(const.CERTIFY, + action_type="lifecycleState") + if result: + self.load() + + def _specific_copy(self, obj: 'Service') -> None: + """ + Copy specific properties from object. + + Args: + obj (Service): the object to "copy" + + """ + super()._specific_copy(obj) + self.category_name = obj.category_name + self.role = obj.role + + def _verify_distribute_to_sdc(self, desired_status: str, + desired_action: str, **kwargs) -> None: + self._verify_action_to_sdc(desired_status, desired_action, + "distribution", **kwargs) + + def _verify_approve_to_sdc(self, desired_status: str, desired_action: str, + **kwargs) -> None: + self._verify_action_to_sdc(desired_status, desired_action, + "distribution-state", **kwargs) + + def _verify_lcm_to_sdc(self, desired_status: str, desired_action: str, + **kwargs) -> None: + self._verify_action_to_sdc(desired_status, desired_action, + "lifecycleState", **kwargs) + + def _verify_action_to_sdc(self, desired_status: str, desired_action: str, + action_type: str, **kwargs) -> None: + """ + Verify action to SDC. + + Verify that object is in right state before launching the action on + SDC. + + Raises: + StatusError: if current status is not the desired status. + + Args: + desired_status (str): the status the object should be + desired_action (str): the action we want to perform + action_type (str): the type of action ('distribution-state' or + 'lifecycleState') + **kwargs: any specific stuff to give to requests + + """ + self._logger.info("attempting to %s Service %s in SDC", desired_action, + self.name) + if self.status == desired_status and self.created(): + self._action_to_sdc(desired_action, + action_type=action_type, + **kwargs) + self.load() + elif not self.created(): + self._logger.warning("Service %s in SDC is not created", self.name) + elif self.status != desired_status: + msg = (f"Service {self.name} in SDC is in status {self.status} " + f"and it should be in status {desired_status}") + raise StatusError(msg) + + @staticmethod + def _unzip_csar_file(zip_file: Union[str, BytesIO], + function: Callable[[str, + TextIOWrapper], None]) -> None: + """ + Unzip Csar File and perform an action on the file. + + Raises: + ValidationError: CSAR file has no service template + + """ + folder = "Definitions" + prefix = "service-" + suffix = "-template.yml" + with ZipFile(zip_file) as myzip: + service_template = None + for name in myzip.namelist(): + if (name[-13:] == suffix + and name[:20] == f"{folder}/{prefix}"): + service_template = name + + if not service_template: + msg = (f"CSAR file has no service template. " + f"Valid path: {folder}/{prefix}*{suffix}") + raise ValidationError(msg) + + with myzip.open(service_template) as template_file: + function(service_template, template_file) + + @staticmethod + def _write_csar_file(service_template: str, + template_file: TextIOWrapper) -> None: + """Write service temple into a file.""" + with open(tosca_path() + service_template[12:], 'wb') as file: + file.write(template_file.read()) + + # _service_template is not used but function generation is generic + # pylint: disable-unused-argument + def _load_tosca_template(self, _service_template: str, + template_file: TextIOWrapper) -> None: + """Load Tosca template.""" + self._tosca_template = yaml.safe_load(template_file.read()) + + @classmethod + def _sdc_path(cls) -> None: + """Give back the end of SDC path.""" + return cls.SERVICE_PATH + + def get_nf_unique_id(self, nf_name: str) -> str: + """ + Get nf (network function) uniqueID. + + Get nf uniqueID from service nf in sdc. + + Args: + nf_name (str): the nf from which we extract the unique ID + + Returns: + the nf unique ID + + Raises: + ResourceNotFound: Couldn't find NF by name. + + """ + url = f"{self._base_create_url()}/services/{self.unique_identifier}" + request_return = self.send_message_json('GET', + 'Get nf unique ID', + url) + + for instance in filter(lambda x: x["componentName"] == nf_name, + request_return["componentInstances"]): + return instance["uniqueId"] + + raise ResourceNotFound(f"NF '{nf_name}'") + + + def add_artifact_to_vf(self, vnf_name: str, artifact_type: str, + artifact_name: str, artifact: BinaryIO = None): + """ + Add artifact to vf. + + Add artifact to vf using payload data. + + Raises: + RequestError: file upload (POST request) for an artifact fails. + + Args: + vnf_name (str): the vnf which we want to add the artifact + artifact_type (str): all SDC artifact types are supported (DCAE_*, HEAT_*, ...) + artifact_name (str): the artifact file name including its extension + artifact (str): binary data to upload + + """ + missing_identifier = self.get_nf_unique_id(vnf_name) + url = (f"{self._base_create_url()}/services/{self.unique_identifier}/" + f"resourceInstance/{missing_identifier}/artifacts") + template = jinja_env().get_template("add_artifact_to_vf.json.j2") + data = template.render(artifact_name=artifact_name, + artifact_label=f"sdk{Path.PurePosixPath(artifact_name).stem}", + artifact_type=artifact_type, + b64_artifact=base64.b64encode(artifact).decode('utf-8')) + headers = headers_sdc_artifact_upload(base_header=self.headers, + data=data) + try: + self.send_message('POST', + 'Add artifact to vf', + url, + headers=headers, + data=data) + except RequestError as exc: + self._logger.error(("an error occured during file upload for an Artifact" + "to VNF %s"), vnf_name) + raise exc + + def get_component_properties_url(self, component: "Component") -> str: + """Url to get component's properties. + + This method is here because component can have different url when + it's a component of another SDC resource type, eg. for service and + for VF components have different urls. + Also for VL origin type components properties url is different than + for the other types. + + Args: + component (Component): Component object to prepare url for + + Returns: + str: Component's properties url + + """ + if component.origin_type == "VL": + return super().get_component_properties_url(component) + return (f"{self.resource_inputs_url}/" + f"componentInstances/{component.unique_id}/{component.actual_component_uid}/inputs") + + def get_component_properties_value_set_url(self, component: "Component") -> str: + """Url to set component property value. + + This method is here because component can have different url when + it's a component of another SDC resource type, eg. for service and + for VF components have different urls. + Also for VL origin type components properties url is different than + for the other types. + + Args: + component (Component): Component object to prepare url for + + Returns: + str: Component's properties url + + """ + if component.origin_type == "VL": + return super().get_component_properties_value_set_url(component) + return (f"{self.resource_inputs_url}/" + f"resourceInstance/{component.unique_id}/inputs") + + def get_category_for_new_resource(self) -> ServiceCategory: + """Get category for service not created in SDC yet. + + If no category values are provided default category is going to be used. + + Returns: + ServiceCategory: Category of the new service + + """ + if not self._category_name: + return ServiceCategory.get(name="Network Service") + return ServiceCategory.get(name=self._category_name) |