diff options
Diffstat (limited to 'src/onapsdk/aai/business')
| -rw-r--r-- | src/onapsdk/aai/business/__init__.py | 27 | ||||
| -rw-r--r-- | src/onapsdk/aai/business/customer.py | 603 | ||||
| -rw-r--r-- | src/onapsdk/aai/business/instance.py | 55 | ||||
| -rw-r--r-- | src/onapsdk/aai/business/line_of_business.py | 123 | ||||
| -rw-r--r-- | src/onapsdk/aai/business/network.py | 223 | ||||
| -rw-r--r-- | src/onapsdk/aai/business/owning_entity.py | 154 | ||||
| -rw-r--r-- | src/onapsdk/aai/business/platform.py | 123 | ||||
| -rw-r--r-- | src/onapsdk/aai/business/pnf.py | 267 | ||||
| -rw-r--r-- | src/onapsdk/aai/business/project.py | 123 | ||||
| -rw-r--r-- | src/onapsdk/aai/business/service.py | 484 | ||||
| -rw-r--r-- | src/onapsdk/aai/business/sp_partner.py | 176 | ||||
| -rw-r--r-- | src/onapsdk/aai/business/vf_module.py | 199 | ||||
| -rw-r--r-- | src/onapsdk/aai/business/vnf.py | 536 |
13 files changed, 3093 insertions, 0 deletions
diff --git a/src/onapsdk/aai/business/__init__.py b/src/onapsdk/aai/business/__init__.py new file mode 100644 index 0000000..41f9671 --- /dev/null +++ b/src/onapsdk/aai/business/__init__.py @@ -0,0 +1,27 @@ +"""A&AI business package.""" +# 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. +# +from .customer import Customer, ServiceSubscription +from .instance import Instance +from .line_of_business import LineOfBusiness +from .network import NetworkInstance +from .owning_entity import OwningEntity +from .platform import Platform +from .pnf import PnfInstance +from .project import Project +from .service import ServiceInstance +from .sp_partner import SpPartner +from .vf_module import VfModuleInstance +from .vnf import VnfInstance diff --git a/src/onapsdk/aai/business/customer.py b/src/onapsdk/aai/business/customer.py new file mode 100644 index 0000000..cdefd6f --- /dev/null +++ b/src/onapsdk/aai/business/customer.py @@ -0,0 +1,603 @@ +"""AAI business 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. +from dataclasses import dataclass +from typing import Iterable, Iterator, Optional +from urllib.parse import urlencode + +from onapsdk.utils.jinja import jinja_env +from onapsdk.exceptions import APIError, ParameterError, ResourceNotFound + +from ..aai_element import AaiResource, Relationship +from ..cloud_infrastructure.cloud_region import CloudRegion +from .service import ServiceInstance + + +@dataclass +class ServiceSubscriptionCloudRegionTenantData: + """Dataclass to store cloud regions and tenants data for service subscription.""" + + cloud_owner: str = None + cloud_region_id: str = None + tenant_id: str = None + + +@dataclass +class ServiceSubscription(AaiResource): + """Service subscription class.""" + + service_type: str + resource_version: str + customer: "Customer" + + def __init__(self, customer: "Customer", service_type: str, resource_version: str) -> None: + """Service subscription object initialization. + + Args: + customer (Customer): Customer object + service_type (str): Service type + resource_version (str): Service subscription resource version + """ + super().__init__() + self.customer: "Customer" = customer + self.service_type: str = service_type + self.resource_version: str = resource_version + + def _get_service_instance_by_filter_parameter(self, + filter_parameter_name: str, + filter_parameter_value: str) -> ServiceInstance: + """Call a request to get service instance with given filter parameter and value. + + Args: + filter_parameter_name (str): Name of parameter to filter + filter_parameter_value (str): Value of filter parameter + + Returns: + ServiceInstance: ServiceInstance object + + """ + service_instance: dict = self.send_message_json( + "GET", + f"Get service instance with {filter_parameter_value} {filter_parameter_name}", + f"{self.url}/service-instances?{filter_parameter_name}={filter_parameter_value}" + )["service-instance"][0] + return ServiceInstance( + service_subscription=self, + instance_id=service_instance.get("service-instance-id"), + instance_name=service_instance.get("service-instance-name"), + service_type=service_instance.get("service-type"), + service_role=service_instance.get("service-role"), + environment_context=service_instance.get("environment-context"), + workload_context=service_instance.get("workload-context"), + created_at=service_instance.get("created-at"), + updated_at=service_instance.get("updated-at"), + description=service_instance.get("description"), + model_invariant_id=service_instance.get("model-invariant-id"), + model_version_id=service_instance.get("model-version-id"), + persona_model_version=service_instance.get("persona-model-version"), + widget_model_id=service_instance.get("widget-model-id"), + widget_model_version=service_instance.get("widget-model-version"), + bandwith_total=service_instance.get("bandwidth-total"), + vhn_portal_url=service_instance.get("vhn-portal-url"), + service_instance_location_id=service_instance.get("service-instance-location-id"), + resource_version=service_instance.get("resource-version"), + selflink=service_instance.get("selflink"), + orchestration_status=service_instance.get("orchestration-status"), + input_parameters=service_instance.get("input-parameters") + ) + + @classmethod + def get_all_url(cls, customer: "Customer") -> str: # pylint: disable=arguments-differ + """Return url to get all customers. + + Returns: + str: Url to get all customers + + """ + return (f"{cls.base_url}{cls.api_version}/business/customers/" + f"customer/{customer.global_customer_id}/service-subscriptions/") + + @classmethod + def create_from_api_response(cls, + api_response: dict, + customer: "Customer") -> "ServiceSubscription": + """Create service subscription using API response dict. + + Returns: + ServiceSubscription: ServiceSubscription object. + + """ + return cls( + service_type=api_response.get("service-type"), + resource_version=api_response.get("resource-version"), + customer=customer + ) + + @property + def url(self) -> str: + """Cloud region object url. + + URL used to call CloudRegion A&AI API + + Returns: + str: CloudRegion object url + + """ + return ( + f"{self.base_url}{self.api_version}/business/customers/" + f"customer/{self.customer.global_customer_id}/service-subscriptions/" + f"service-subscription/{self.service_type}" + ) + + @property + def service_instances(self) -> Iterator[ServiceInstance]: + """Service instances. + + Yields: + Iterator[ServiceInstance]: Service instance + + """ + for service_instance in \ + self.send_message_json("GET", + (f"Get all service instances for {self.service_type} service " + f"subscription"), + f"{self.url}/service-instances").get("service-instance", []): + yield ServiceInstance( + service_subscription=self, + instance_id=service_instance.get("service-instance-id"), + instance_name=service_instance.get("service-instance-name"), + service_type=service_instance.get("service-type"), + service_role=service_instance.get("service-role"), + environment_context=service_instance.get("environment-context"), + workload_context=service_instance.get("workload-context"), + created_at=service_instance.get("created-at"), + updated_at=service_instance.get("updated-at"), + description=service_instance.get("description"), + model_invariant_id=service_instance.get("model-invariant-id"), + model_version_id=service_instance.get("model-version-id"), + persona_model_version=service_instance.get("persona-model-version"), + widget_model_id=service_instance.get("widget-model-id"), + widget_model_version=service_instance.get("widget-model-version"), + bandwith_total=service_instance.get("bandwidth-total"), + vhn_portal_url=service_instance.get("vhn-portal-url"), + service_instance_location_id=service_instance.get("service-instance-location-id"), + resource_version=service_instance.get("resource-version"), + selflink=service_instance.get("selflink"), + orchestration_status=service_instance.get("orchestration-status"), + input_parameters=service_instance.get("input-parameters") + ) + + @property + def tenant_relationships(self) -> Iterator["Relationship"]: + """Tenant related relationships. + + Iterate through relationships and get related to tenant. + + Yields: + Relationship: Relationship related to tenant. + + """ + for relationship in self.relationships: + if relationship.related_to == "tenant": + yield relationship + + @property + def cloud_region(self) -> "CloudRegion": + """Cloud region associated with service subscription. + + IT'S DEPRECATED! `cloud_regions` parameter SHOULD BE USED + + Raises: + ParameterError: Service subscription has no associated cloud region. + + Returns: + CloudRegion: CloudRegion object + + """ + try: + return next(self.cloud_regions) + except StopIteration: + msg = f"No cloud region for service subscription '{self.name}'" + raise ParameterError(msg) + + @property + def tenant(self) -> "Tenant": + """Tenant associated with service subscription. + + IT'S DEPRECATED! `tenants` parameter SHOULD BE USED + + Raises: + ParameterError: Service subscription has no associated tenants + + Returns: + Tenant: Tenant object + + """ + try: + return next(self.tenants) + except StopIteration: + msg = f"No tenants for service subscription '{self.name}'" + raise ParameterError(msg) + + @property + def _cloud_regions_tenants_data(self) -> Iterator["ServiceSubscriptionCloudRegionTenantData"]: + for relationship in self.tenant_relationships: + cr_tenant_data: ServiceSubscriptionCloudRegionTenantData = \ + ServiceSubscriptionCloudRegionTenantData() + for data in relationship.relationship_data: + if data["relationship-key"] == "cloud-region.cloud-owner": + cr_tenant_data.cloud_owner = data["relationship-value"] + if data["relationship-key"] == "cloud-region.cloud-region-id": + cr_tenant_data.cloud_region_id = data["relationship-value"] + if data["relationship-key"] == "tenant.tenant-id": + cr_tenant_data.tenant_id = data["relationship-value"] + if all([cr_tenant_data.cloud_owner, + cr_tenant_data.cloud_region_id, + cr_tenant_data.tenant_id]): + yield cr_tenant_data + else: + self._logger.error("Invalid tenant relationship: %s", relationship) + + @property + def cloud_regions(self) -> Iterator["CloudRegion"]: + """Cloud regions associated with service subscription. + + Yields: + CloudRegion: CloudRegion object + + """ + cloud_region_set: set = set() + for cr_data in self._cloud_regions_tenants_data: + cloud_region_set.add((cr_data.cloud_owner, cr_data.cloud_region_id)) + for cloud_region_data in cloud_region_set: + try: + yield CloudRegion.get_by_id(cloud_owner=cloud_region_data[0], + cloud_region_id=cloud_region_data[1]) + except ResourceNotFound: + self._logger.error("Can't get cloud region %s %s", cloud_region_data[0], \ + cloud_region_data[1]) + + @property + def tenants(self) -> Iterator["Tenant"]: + """Tenants associated with service subscription. + + Yields: + Tenant: Tenant object + + """ + for cr_data in self._cloud_regions_tenants_data: + try: + cloud_region: CloudRegion = CloudRegion.get_by_id(cr_data.cloud_owner, + cr_data.cloud_region_id) + yield cloud_region.get_tenant(cr_data.tenant_id) + except ResourceNotFound: + self._logger.error("Can't get %s tenant", cr_data.tenant_id) + + def get_service_instance_by_id(self, service_instance_id) -> ServiceInstance: + """Get service instance using it's ID. + + Args: + service_instance_id (str): ID of the service instance + + Returns: + ServiceInstance: ServiceInstance object + + """ + return self._get_service_instance_by_filter_parameter( + "service-instance-id", + service_instance_id + ) + + def get_service_instance_by_name(self, service_instance_name: str) -> ServiceInstance: + """Get service instance using it's name. + + Args: + service_instance_name (str): Name of the service instance + + Returns: + ServiceInstance: ServiceInstance object + + """ + return self._get_service_instance_by_filter_parameter( + "service-instance-name", + service_instance_name + ) + + def link_to_cloud_region_and_tenant(self, + cloud_region: "CloudRegion", + tenant: "Tenant") -> None: + """Create relationship between object and cloud region with tenant. + + Args: + cloud_region (CloudRegion): Cloud region to link to + tenant (Tenant): Cloud region tenant to link to + """ + relationship: Relationship = Relationship( + related_to="tenant", + related_link=tenant.url, + relationship_data=[ + { + "relationship-key": "cloud-region.cloud-owner", + "relationship-value": cloud_region.cloud_owner, + }, + { + "relationship-key": "cloud-region.cloud-region-id", + "relationship-value": cloud_region.cloud_region_id, + }, + { + "relationship-key": "tenant.tenant-id", + "relationship-value": tenant.tenant_id, + }, + ], + related_to_property=[ + {"property-key": "tenant.tenant-name", "property-value": tenant.name} + ], + ) + self.add_relationship(relationship) + + +class Customer(AaiResource): + """Customer class.""" + + def __init__(self, + global_customer_id: str, + subscriber_name: str, + subscriber_type: str, + resource_version: str = None) -> None: + """Initialize Customer class object. + + Args: + global_customer_id (str): Global customer id used across ONAP to + uniquely identify customer. + subscriber_name (str): Subscriber name, an alternate way to retrieve a customer. + subscriber_type (str): Subscriber type, a way to provide VID with + only the INFRA customers. + resource_version (str, optional): Used for optimistic concurrency. + Must be empty on create, valid on update + and delete. Defaults to None. + + """ + super().__init__() + self.global_customer_id: str = global_customer_id + self.subscriber_name: str = subscriber_name + self.subscriber_type: str = subscriber_type + self.resource_version: str = resource_version + + def __repr__(self) -> str: # noqa + """Customer description. + + Returns: + str: Customer object description + + """ + return (f"Customer(global_customer_id={self.global_customer_id}, " + f"subscriber_name={self.subscriber_name}, " + f"subscriber_type={self.subscriber_type}, " + f"resource_version={self.resource_version})") + + def get_service_subscription_by_service_type(self, service_type: str) -> ServiceSubscription: + """Get subscribed service by service type. + + Call a request to get service subscriptions filtered by service-type parameter. + + Args: + service_type (str): Service type + + Returns: + ServiceSubscription: Service subscription + + """ + response: dict = self.send_message_json( + "GET", + f"Get service subscription with {service_type} service type", + (f"{self.base_url}{self.api_version}/business/customers/" + f"customer/{self.global_customer_id}/service-subscriptions" + f"?service-type={service_type}") + ) + return ServiceSubscription.create_from_api_response(response["service-subscription"][0], + self) + + @classmethod + def get_all_url(cls) -> str: # pylint: disable=arguments-differ + """Return an url to get all customers. + + Returns: + str: URL to get all customers + + """ + return f"{cls.base_url}{cls.api_version}/business/customers" + + @classmethod + def get_all(cls, + global_customer_id: str = None, + subscriber_name: str = None, + subscriber_type: str = None) -> Iterator["Customer"]: + """Get all customers. + + Call an API to retrieve all customers. It can be filtered + by global-customer-id, subscriber-name and/or subsriber-type. + + Args: + global_customer_id (str): global-customer-id to filer customers by. Defaults to None. + subscriber_name (str): subscriber-name to filter customers by. Defaults to None. + subscriber_type (str): subscriber-type to filter customers by. Defaults to None. + + """ + filter_parameters: dict = cls.filter_none_key_values( + { + "global-customer-id": global_customer_id, + "subscriber-name": subscriber_name, + "subscriber-type": subscriber_type, + } + ) + url: str = (f"{cls.get_all_url()}?{urlencode(filter_parameters)}") + for customer in cls.send_message_json("GET", "get customers", url).get("customer", []): + yield Customer( + global_customer_id=customer["global-customer-id"], + subscriber_name=customer["subscriber-name"], + subscriber_type=customer["subscriber-type"], + resource_version=customer["resource-version"], + ) + + @classmethod + def get_by_global_customer_id(cls, global_customer_id: str) -> "Customer": + """Get customer by it's global customer id. + + Args: + global_customer_id (str): global customer ID + + Returns: + Customer: Customer with given global_customer_id + + """ + response: dict = cls.send_message_json( + "GET", + f"Get {global_customer_id} customer", + f"{cls.base_url}{cls.api_version}/business/customers/customer/{global_customer_id}" + ) + return Customer( + global_customer_id=response["global-customer-id"], + subscriber_name=response["subscriber-name"], + subscriber_type=response["subscriber-type"], + resource_version=response["resource-version"], + ) + + @classmethod + def create(cls, + global_customer_id: str, + subscriber_name: str, + subscriber_type: str, + service_subscriptions: Optional[Iterable[str]] = None) -> "Customer": + """Create customer. + + Args: + global_customer_id (str): Global customer id used across ONAP + to uniquely identify customer. + subscriber_name (str): Subscriber name, an alternate way + to retrieve a customer. + subscriber_type (str): Subscriber type, a way to provide + VID with only the INFRA customers. + service_subscriptions (Optional[Iterable[str]], optional): Iterable + of service subscription names should be created for newly + created customer. Defaults to None. + + Returns: + Customer: Customer object. + + """ + url: str = ( + f"{cls.base_url}{cls.api_version}/business/customers/" + f"customer/{global_customer_id}" + ) + cls.send_message( + "PUT", + "declare customer", + url, + data=jinja_env() + .get_template("customer_create.json.j2") + .render( + global_customer_id=global_customer_id, + subscriber_name=subscriber_name, + subscriber_type=subscriber_type, + service_subscriptions=service_subscriptions + ), + ) + response: dict = cls.send_message_json( + "GET", "get created customer", url + ) # Call API one more time to get Customer's resource version + return Customer( + global_customer_id=response["global-customer-id"], + subscriber_name=response["subscriber-name"], + subscriber_type=response["subscriber-type"], + resource_version=response["resource-version"], + ) + + @property + def url(self) -> str: + """Return customer object url. + + Unique url address to get customer's data. + + Returns: + str: Customer object url + + """ + return ( + f"{self.base_url}{self.api_version}/business/customers/customer/" + f"{self.global_customer_id}?resource-version={self.resource_version}" + ) + + @property + def service_subscriptions(self) -> Iterator[ServiceSubscription]: + """Service subscriptions of customer resource. + + Yields: + ServiceSubscription: ServiceSubscription object + + """ + try: + response: dict = self.send_message_json( + "GET", + "get customer service subscriptions", + f"{self.base_url}{self.api_version}/business/customers/" + f"customer/{self.global_customer_id}/service-subscriptions" + ) + for service_subscription in response.get("service-subscription", []): + yield ServiceSubscription.create_from_api_response( + service_subscription, + self + ) + except ResourceNotFound as exc: + self._logger.info( + "Subscriptions are not " \ + "found for a customer: %s", exc) + except APIError as exc: + self._logger.error( + "API returned an error: %s", exc) + + def subscribe_service(self, service_type: str) -> "ServiceSubscription": + """Create SDC Service subscription. + + If service subscription with given service_type already exists it won't create + a new resource but use the existing one. + + Args: + service_type (str): Value defined by orchestration to identify this service + across ONAP. + """ + try: + return self.get_service_subscription_by_service_type(service_type) + except ResourceNotFound: + self._logger.info("Create service subscription for %s customer", + self.global_customer_id) + self.send_message( + "PUT", + "Create service subscription", + (f"{self.base_url}{self.api_version}/business/customers/" + f"customer/{self.global_customer_id}/service-subscriptions/" + f"service-subscription/{service_type}") + ) + return self.get_service_subscription_by_service_type(service_type) + + def delete(self) -> None: + """Delete customer. + + Sends request to A&AI to delete customer object. + + """ + self.send_message( + "DELETE", + "Delete customer", + self.url + ) diff --git a/src/onapsdk/aai/business/instance.py b/src/onapsdk/aai/business/instance.py new file mode 100644 index 0000000..146aee9 --- /dev/null +++ b/src/onapsdk/aai/business/instance.py @@ -0,0 +1,55 @@ +"""Base instance 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. + +from abc import ABC, abstractmethod + +from ..aai_element import AaiResource + + +class Instance(AaiResource, ABC): + """Abstract instance class.""" + + def __init__(self, # pylint: disable=too-many-arguments + resource_version: str = None, + model_invariant_id: str = None, + model_version_id: str = None) -> None: + """Instance initialization. + + Args: + resource_version (str, optional): Used for optimistic concurrency. + Must be empty on create, valid on update and delete. Defaults to None. + model_invariant_id (str, optional): The ASDC model id for this resource or + service model. Defaults to None. + model_version_id (str, optional): The ASDC model version for this resource or + service model. Defaults to None. + """ + super().__init__() + self.resource_version: str = resource_version + self.model_invariant_id: str = model_invariant_id + self.model_version_id: str = model_version_id + + @abstractmethod + def delete(self, a_la_carte: bool = True) -> "DeletionRequest": + """Create instance deletion request. + + Send request to delete instance + + Args: + a_la_carte (boolean): deletion mode + + Returns: + DeletionRequest: Deletion request + + """ diff --git a/src/onapsdk/aai/business/line_of_business.py b/src/onapsdk/aai/business/line_of_business.py new file mode 100644 index 0000000..61fc0f8 --- /dev/null +++ b/src/onapsdk/aai/business/line_of_business.py @@ -0,0 +1,123 @@ +"""A&AI line of business 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. + +from typing import Any, Dict, Iterator + +from onapsdk.utils.jinja import jinja_env + +from ..aai_element import AaiResource + + +class LineOfBusiness(AaiResource): + """Line of business class.""" + + def __init__(self, name: str, resource_version: str) -> None: + """Line of business object initialization. + + Args: + name (str): Line of business name + resource_version (str): resource version + """ + super().__init__() + self.name: str = name + self.resource_version: str = resource_version + + @property + def url(self) -> str: + """Line of business's url. + + Returns: + str: Resource's url + + """ + return (f"{self.base_url}{self.api_version}/business/lines-of-business/" + f"line-of-business/{self.name}") + + @classmethod + def get_all_url(cls) -> str: # pylint: disable=arguments-differ + """Return url to get all lines of business. + + Returns: + str: Url to get all lines of business + + """ + return f"{cls.base_url}{cls.api_version}/business/lines-of-business/" + + def __repr__(self) -> str: + """Line of business object representation. + + Returns: + str: Line of business object representation + + """ + return f"LineOfBusiness(name={self.name})" + + @classmethod + def get_all(cls) -> Iterator["LineOfBusiness"]: + """Get all line of business. + + Yields: + LineOfBusiness: LineOfBusiness object + + """ + url: str = f"{cls.base_url}{cls.api_version}/business/lines-of-business" + for line_of_business in cls.send_message_json("GET", + "Get A&AI lines of business", + url).get("line-of-business", []): + yield cls( + line_of_business.get("line-of-business-name"), + line_of_business.get("resource-version") + ) + + @classmethod + def create(cls, name: str) -> "LineOfBusiness": + """Create line of business A&AI resource. + + Args: + name (str): line of business name + + Returns: + LineOfBusiness: Created LineOfBusiness object + + """ + cls.send_message( + "PUT", + "Declare A&AI line of business", + (f"{cls.base_url}{cls.api_version}/business/lines-of-business/" + f"line-of-business/{name}"), + data=jinja_env().get_template("aai_line_of_business_create.json.j2").render( + line_of_business_name=name + ) + ) + return cls.get_by_name(name) + + @classmethod + def get_by_name(cls, name: str) -> "LineOfBusiness": + """Get line of business resource by it's name. + + Raises: + ResourceNotFound: Line of business requested by a name does not exist. + + Returns: + LineOfBusiness: Line of business requested by a name. + + """ + url = (f"{cls.base_url}{cls.api_version}/business/lines-of-business/" + f"line-of-business/{name}") + response: Dict[str, Any] = \ + cls.send_message_json("GET", + f"Get {name} line of business", + url) + return cls(response["line-of-business-name"], response["resource-version"]) diff --git a/src/onapsdk/aai/business/network.py b/src/onapsdk/aai/business/network.py new file mode 100644 index 0000000..e36cf62 --- /dev/null +++ b/src/onapsdk/aai/business/network.py @@ -0,0 +1,223 @@ +"""Network instance 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. + +from onapsdk.so.deletion import NetworkDeletionRequest + +from .instance import Instance + + +class NetworkInstance(Instance): # pylint: disable=too-many-instance-attributes + """Network instance class.""" + + def __init__(self, # pylint: disable=too-many-arguments, too-many-locals + service_instance: "ServiceInstance", + network_id: str, + is_bound_to_vpn: bool, + is_provider_network: bool, + is_shared_network: bool, + is_external_network: bool, + network_name: str = None, + network_type: str = None, + network_role: str = None, + network_technology: str = None, + neutron_network_id: str = None, + service_id: str = None, + network_role_instance: str = None, + resource_version: str = None, + orchestration_status: str = None, + heat_stack_id: str = None, + mso_catalog_key: str = None, + model_invariant_id: str = None, + contrail_network_fqdn: str = None, + persona_model_version: str = None, + model_version_id: str = None, + model_customization_id: str = None, + widget_model_id: str = None, + physical_network_name: str = None, + widget_model_version: str = None, + selflink: str = None, + operational_status: str = None, + is_trunked: bool = None): + """Network instance object initialization. + + Args: + service_instance (ServiceInstance): Service instance object + network_id (str): Network ID, should be uuid. Unique across A&AI. + is_bound_to_vpn (bool): Set to true if bound to VPN + is_provider_network (bool): boolean indicatating whether or not network + is a provider network. + is_shared_network (bool): boolean indicatating whether + or not network is a shared network. + is_external_network (bool): boolean indicatating whether + or not network is an external network. + network_name (str, optional): Name of the network, governed by some naming convention. + Defaults to None. + network_type (str, optional): Type of the network. Defaults to None. + network_role (str, optional): Role the network. Defaults to None. + network_technology (str, optional): Network technology. Defaults to None. + neutron_network_id (str, optional): Neutron network id of this Interface. + Defaults to None. + service_id (str, optional): Unique identifier of service from ASDC. + Does not strictly map to ASDC services. Defaults to None. + network_role_instance (str, optional): network role instance. Defaults to None. + resource_version (str, optional): Used for optimistic concurrency. + Must be empty on create, valid on update and delete. Defaults to None. + orchestration_status (str, optional): Orchestration status of this VNF, + mastered by MSO. Defaults to None. + heat_stack_id (str, optional): Heat stack id corresponding to this instance, + managed by MSO. Defaults to None. + mso_catalog_key (str, optional): Corresponds to the SDN-C catalog id used to + configure this VCE. Defaults to None. + contrail_network_fqdn (str, optional): Contrail FQDN for the network. Defaults to None. + model_invariant_id (str, optional): the ASDC model id for this resource + or service model. Defaults to None. + model_version_id (str, optional): the ASDC model version for this resource + or service model. Defaults to None. + persona_model_version (str, optional): the ASDC model version for this resource + or service model. Defaults to None. + model_customization_id (str, optional): captures the id of all the configuration + used to customize the resource for the service. Defaults to None. + widget_model_id (str, optional): the ASDC data dictionary widget model. + This maps directly to the A&AI widget. Defaults to None. + widget_model_version (str, optional): the ASDC data dictionary version of + the widget model. This maps directly to the A&AI version of the widget. + Defaults to None. + physical_network_name (str, optional): Name associated with the physical network. + Defaults to None. + selflink (str, optional): Path to the controller object. Defaults to None. + operational_status (str, optional): Indicator for whether the resource is considered + operational. Defaults to None. + is_trunked (bool, optional): Trunked network indication. Defaults to None. + """ + super().__init__(resource_version=resource_version, + model_version_id=model_version_id, + model_invariant_id=model_invariant_id) + self.service_instance: "ServiceInstance" = service_instance + self.network_id: str = network_id + self.is_bound_to_vpn: bool = is_bound_to_vpn + self.is_provider_network: bool = is_provider_network + self.is_shared_network: bool = is_shared_network + self.is_external_network: bool = is_external_network + self.network_name: str = network_name + self.network_type: str = network_type + self.network_role: str = network_role + self.network_technology: str = network_technology + self.neutron_network_id: str = neutron_network_id + self.service_id: str = service_id + self.network_role_instance: str = network_role_instance + self.orchestration_status: str = orchestration_status + self.heat_stack_id: str = heat_stack_id + self.mso_catalog_key: str = mso_catalog_key + self.contrail_network_fqdn: str = contrail_network_fqdn + self.model_customization_id: str = model_customization_id + self.physical_network_name: str = physical_network_name + self.selflink: str = selflink + self.operational_status: str = operational_status + self.is_trunked: bool = is_trunked + self.persona_model_version: str = persona_model_version + self.widget_model_id: str = widget_model_id + self.widget_model_version: str = widget_model_version + + def __repr__(self) -> str: + """Network instance object representation. + + Returns: + str: Human readable network instance representation + + """ + return (f"NetworkInstance(network_id={self.network_id}, " + f"network_name={self.network_name}, " + f"is_bound_to_vpn={self.is_bound_to_vpn}, " + f"is_provider_network={self.is_provider_network}, " + f"is_shared_network={self.is_shared_network}, " + f"is_external_network={self.is_external_network}, " + f"orchestration_status={self.orchestration_status})") + + @classmethod + def get_all_url(cls) -> str: # pylint: disable=arguments-differ + """Return url to get all networks. + + Returns: + str: Url to get all networks + + """ + return f"{cls.base_url}{cls.api_version}/network/l3-networks/" + + @property + def url(self) -> str: + """Network instance url. + + Returns: + str: NetworkInstance url + + """ + return f"{self.base_url}{self.api_version}/network/l3-networks/l3-network/{self.network_id}" + + @classmethod + def create_from_api_response(cls, api_response: dict, + service_instance: "ServiceInstance") -> "NetworkInstance": + """Create network instance object using HTTP API response dictionary. + + Args: + api_response (dict): A&AI API response dictionary + service_instance (ServiceInstance): Service instance with which network is related + + Returns: + VnfInstance: VnfInstance object + + """ + return cls(service_instance=service_instance, + network_id=api_response["network-id"], + is_bound_to_vpn=api_response["is-bound-to-vpn"], + is_provider_network=api_response["is-provider-network"], + is_shared_network=api_response["is-shared-network"], + is_external_network=api_response["is-external-network"], + network_name=api_response.get("network-name"), + network_type=api_response.get("network-type"), + network_role=api_response.get("network-role"), + network_technology=api_response.get("network-technology"), + neutron_network_id=api_response.get("neutron-network-id"), + service_id=api_response.get("service-id"), + network_role_instance=api_response.get("network-role-instance"), + resource_version=api_response.get("resource-version"), + orchestration_status=api_response.get("orchestration-status"), + heat_stack_id=api_response.get("heat-stack-id"), + mso_catalog_key=api_response.get("mso-catalog-key"), + model_invariant_id=api_response.get("model-invariant-id"), + contrail_network_fqdn=api_response.get("contrail-network-fqdn"), + model_version_id=api_response.get("model-version-id"), + model_customization_id=api_response.get("model-customization-id"), + widget_model_id=api_response.get("widget-model-id"), + persona_model_version=api_response.get("persona-model-version"), + physical_network_name=api_response.get("physical-network-name"), + selflink=api_response.get("selflink"), + widget_model_version=api_response.get("widget-model-version"), + operational_status=api_response.get("operational-status"), + is_trunked=api_response.get("is-trunked")) + + def delete(self, a_la_carte: bool = True) -> "NetworkDeletionRequest": + """Create network deletion request. + + Send request to delete network instance + + Args: + a_la_carte (boolean): deletion mode + + Returns: + NetworkDeletionRequest: Deletion request + + """ + self._logger.debug("Delete %s network", self.network_id) + return NetworkDeletionRequest.send_request(self, a_la_carte) diff --git a/src/onapsdk/aai/business/owning_entity.py b/src/onapsdk/aai/business/owning_entity.py new file mode 100644 index 0000000..bf1e7c1 --- /dev/null +++ b/src/onapsdk/aai/business/owning_entity.py @@ -0,0 +1,154 @@ +"""A&AI owning entity 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. + +from uuid import uuid4 +from typing import Iterator + +from onapsdk.utils.jinja import jinja_env +from onapsdk.exceptions import ResourceNotFound + +from ..aai_element import AaiResource + + +class OwningEntity(AaiResource): + """Owning entity class.""" + + def __init__(self, name: str, owning_entity_id: str, resource_version: str) -> None: + """Owning entity object initialization. + + Args: + name (str): Owning entity name + owning_entity_id (str): owning entity ID + resource_version (str): resource version + """ + super().__init__() + self.name: str = name + self.owning_entity_id: str = owning_entity_id + self.resource_version: str = resource_version + + def __repr__(self) -> str: + """Owning entity object representation. + + Returns: + str: Owning entity object representation + + """ + return f"OwningEntity(name={self.name}, owning_entity_id={self.owning_entity_id})" + + @property + def url(self) -> str: + """Owning entity object url. + + Returns: + str: Url + + """ + return (f"{self.base_url}{self.api_version}/business/owning-entities/owning-entity/" + f"{self.owning_entity_id}") + + @classmethod + def get_all_url(cls) -> str: # pylint: disable=arguments-differ + """Return url to get all owning entities. + + Returns: + str: Url to get all owning entities + + """ + return f"{cls.base_url}{cls.api_version}/business/owning-entities" + + @classmethod + def get_all(cls) -> Iterator["OwningEntity"]: + """Get all owning entities. + + Yields: + OwningEntity: OwningEntity object + + """ + url: str = cls.get_all_url() + for owning_entity in cls.send_message_json("GET", + "Get A&AI owning entities", + url).get("owning-entity", []): + yield cls( + owning_entity.get("owning-entity-name"), + owning_entity.get("owning-entity-id"), + owning_entity.get("resource-version") + ) + + @classmethod + def get_by_owning_entity_id(cls, owning_entity_id: str) -> "OwningEntity": + """Get owning entity by it's ID. + + Args: + owning_entity_id (str): owning entity object id + + Returns: + OwningEntity: OwningEntity object + + """ + response: dict = cls.send_message_json( + "GET", + "Get A&AI owning entity", + (f"{cls.base_url}{cls.api_version}/business/owning-entities/" + f"owning-entity/{owning_entity_id}") + ) + return cls( + response.get("owning-entity-name"), + response.get("owning-entity-id"), + response.get("resource-version") + ) + + @classmethod + def get_by_owning_entity_name(cls, owning_entity_name: str) -> "OwningEntity": + """Get owning entity resource by it's name. + + Raises: + ResourceNotFound: Owning entity requested by a name does not exist. + + Returns: + OwningEntity: Owning entity requested by a name. + + """ + for owning_entity in cls.get_all(): + if owning_entity.name == owning_entity_name: + return owning_entity + + msg = f'Owning entity "{owning_entity_name}" does not exist.' + raise ResourceNotFound(msg) + + @classmethod + def create(cls, name: str, owning_entity_id: str = None) -> "OwningEntity": + """Create owning entity A&AI resource. + + Args: + name (str): owning entity name + owning_entity_id (str): owning entity ID. Defaults to None. + + Returns: + OwningEntity: Created OwningEntity object + + """ + if not owning_entity_id: + owning_entity_id = str(uuid4()) + cls.send_message( + "PUT", + "Declare A&AI owning entity", + (f"{cls.base_url}{cls.api_version}/business/owning-entities/" + f"owning-entity/{owning_entity_id}"), + data=jinja_env().get_template("aai_owning_entity_create.json.j2").render( + owning_entity_name=name, + owning_entity_id=owning_entity_id + ) + ) + return cls.get_by_owning_entity_id(owning_entity_id) diff --git a/src/onapsdk/aai/business/platform.py b/src/onapsdk/aai/business/platform.py new file mode 100644 index 0000000..5c12ba8 --- /dev/null +++ b/src/onapsdk/aai/business/platform.py @@ -0,0 +1,123 @@ +"""A&AI platform 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. + +from typing import Any, Dict, Iterator + +from onapsdk.utils.jinja import jinja_env + +from ..aai_element import AaiResource + + +class Platform(AaiResource): + """Platform class.""" + + def __init__(self, name: str, resource_version: str) -> None: + """Platform object initialization. + + Args: + name (str): Platform name + resource_version (str): resource version + """ + super().__init__() + self.name: str = name + self.resource_version: str = resource_version + + def __repr__(self) -> str: + """Platform object representation. + + Returns: + str: Platform object representation + + """ + return f"Platform(name={self.name})" + + @property + def url(self) -> str: + """Platform's url. + + Returns: + str: Resource's url + + """ + return (f"{self.base_url}{self.api_version}/business/platforms/" + f"platform/{self.name}") + + @classmethod + def get_all_url(cls) -> str: # pylint: disable=arguments-differ + """Return url to get all platforms. + + Returns: + str: Url to get all platforms + + """ + return f"{cls.base_url}{cls.api_version}/business/platforms" + + @classmethod + def get_all(cls) -> Iterator["Platform"]: + """Get all platform. + + Yields: + Platform: Platform object + + """ + url: str = cls.get_all_url() + for platform in cls.send_message_json("GET", + "Get A&AI platforms", + url).get("platform", []): + yield cls( + platform.get("platform-name"), + platform.get("resource-version") + ) + + @classmethod + def create(cls, name: str) -> "Platform": + """Create platform A&AI resource. + + Args: + name (str): platform name + + Returns: + Platform: Created Platform object + + """ + cls.send_message( + "PUT", + "Declare A&AI platform", + (f"{cls.base_url}{cls.api_version}/business/platforms/" + f"platform/{name}"), + data=jinja_env().get_template("aai_platform_create.json.j2").render( + platform_name=name + ) + ) + return cls.get_by_name(name) + + @classmethod + def get_by_name(cls, name: str) -> "Platform": + """Get platform resource by it's name. + + Raises: + ResourceNotFound: Platform requested by a name does not exist. + + Returns: + Platform: Platform requested by a name. + + """ + url = (f"{cls.base_url}{cls.api_version}/business/platforms/" + f"platform/{name}") + response: Dict[str, Any] = \ + cls.send_message_json("GET", + f"Get {name} platform", + url) + return cls(response["platform-name"], response["resource-version"]) diff --git a/src/onapsdk/aai/business/pnf.py b/src/onapsdk/aai/business/pnf.py new file mode 100644 index 0000000..9061ebf --- /dev/null +++ b/src/onapsdk/aai/business/pnf.py @@ -0,0 +1,267 @@ +"""Pnf instance 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. + +from typing import Iterator, Optional, TYPE_CHECKING + +from onapsdk.exceptions import ResourceNotFound +from .instance import Instance + +if TYPE_CHECKING: + from .service import ServiceInstance # pylint: disable=cyclic-import + +class PnfInstance(Instance): # pylint: disable=too-many-instance-attributes + """Pnf instance class.""" + + def __init__(self, # pylint: disable=too-many-arguments, too-many-locals + service_instance: "ServiceInstance", + pnf_name: str, + in_maint: bool, + selflink: str = None, + pnf_id: str = None, + equip_type: str = None, + equip_vendor: str = None, + equip_model: str = None, + management_option: str = None, + orchestration_status: str = None, + ipaddress_v4_oam: str = None, + sw_version: str = None, + frame_id: str = None, + serial_number: str = None, + ipaddress_v4_loopback_0: str = None, + ipaddress_v6_loopback_0: str = None, + ipaddress_v4_aim: str = None, + ipaddress_v6_aim: str = None, + ipaddress_v6_oam: str = None, + inv_status: str = None, + resource_version: str = None, + prov_status: str = None, + nf_role: str = None, + admin_status: str = None, + operational_status: str = None, + model_customization_id: str = None, + model_invariant_id: str = None, + model_version_id: str = None, + pnf_ipv4_address: str = None, + pnf_ipv6_address: str = None) -> None: + """Pnf instance object initialization. + + Args: + service_instance (ServiceInstance): Service instance object + pnf_name (str): unique name of Physical Network Function + in_maint (bool): Used to indicate whether or not this object is in maintenance mode + (maintenance mode = True). This field (in conjunction with prov_status) + is used to suppress alarms and vSCL on VNFs/VMs. + selflink (str, optional): URL to endpoint where AAI can get more details. + Defaults to None. + pnf_id (str, optional): id of pnf. Defaults to None. + equip_type (str, optional): Equipment type. Source of truth should define valid values. + Defaults to None. + equip_vendor (str, optional): Equipment vendor. Source of truth should define + valid values. Defaults to None. + equip_model (str, optional): Equipment model. Source of truth should define + valid values. Defaults to None. + management_option (str, optional): identifier of managed customer. Defaults to None. + orchestration_status (str, optional): Orchestration status of this pnf. + Defaults to None. + ipaddress_v4_oam (str, optional): ipv4-oam-address with new naming + convention for IP addresses. Defaults to None. + sw_version (str, optional): sw-version is the version of SW for the hosted + application on the PNF. Defaults to None. + frame_id (str, optional): ID of the physical frame (relay rack) where pnf is installed. + Defaults to None. + serial_number (str, optional): Serial number of the device. Defaults to None. + ipaddress_v4_loopback_0 (str, optional): IPV4 Loopback 0 address. Defaults to None. + ipaddress_v6_loopback_0 (str, optional): IPV6 Loopback 0 address. Defaults to None. + ipaddress_v4_aim (str, optional): IPV4 AIM address. Defaults to None. + ipaddress_v6_aim (str, optional): IPV6 AIM address. Defaults to None. + ipaddress_v6_oam (str, optional): IPV6 OAM address. Defaults to None. + inv_status (str, optional): CANOPI's inventory status. Only set with values exactly + as defined by CANOPI. Defaults to None. + resource_version (str, optional): Used for optimistic concurrency. + Must be empty on create, valid on update and delete. Defaults to None. + prov_status (str, optional): Prov Status of this device (not under canopi control) + Valid values [PREPROV/NVTPROV/PROV]. Defaults to None. + nf_role (str, optional): Nf Role is the role performed by this instance in the network. + Defaults to None. + admin_status (str, optional): admin Status of this PNF. Defaults to None. + operational_status (str, optional): Store the operational-status for this object. + Defaults to None. + model_customization_id (str, optional): Store the model-customization-id + for this object. Defaults to None. + model_invariant_id (str, optional): The ASDC model id for this resource model. + Defaults to None. + model_version_id (str, optional): The ASDC model version for this resource model. + Defaults to None. + pnf_ipv4_address (str, optional): This is the IP address (IPv4) for the PNF itself. + This is the IPv4 address that the PNF iself can be accessed at. Defaults to None. + pnf_ipv6_address (str, optional): This is the IP address (IPv6) for the PNF itself. + This is the IPv6 address that the PNF iself can be accessed at. Defaults to None. + """ + super().__init__(resource_version=resource_version, + model_invariant_id=model_invariant_id, + model_version_id=model_version_id) + self.service_instance: "ServiceInstance" = service_instance + self.pnf_name: str = pnf_name + self.in_maint: bool = in_maint + self.selflink: Optional[str] = selflink + self.pnf_id: Optional[str] = pnf_id + self.equip_type: Optional[str] = equip_type + self.equip_vendor: Optional[str] = equip_vendor + self.equip_model: Optional[str] = equip_model + self.management_option: Optional[str] = management_option + self.orchestration_status: Optional[str] = orchestration_status + self.ipaddress_v4_oam: Optional[str] = ipaddress_v4_oam + self.sw_version: Optional[str] = sw_version + self.frame_id: Optional[str] = frame_id + self.serial_number: Optional[str] = serial_number + self.ipaddress_v4_loopback_0: Optional[str] = ipaddress_v4_loopback_0 + self.ipaddress_v6_loopback_0: Optional[str] = ipaddress_v6_loopback_0 + self.ipaddress_v4_aim: Optional[str] = ipaddress_v4_aim + self.ipaddress_v6_aim: Optional[str] = ipaddress_v6_aim + self.ipaddress_v6_oam: Optional[str] = ipaddress_v6_oam + self.inv_status: Optional[str] = inv_status + self.prov_status: Optional[str] = prov_status + self.nf_role: Optional[str] = nf_role + self.admin_status: Optional[str] = admin_status + self.operational_status: Optional[str] = operational_status + self.model_customization_id: Optional[str] = model_customization_id + self.pnf_ipv4_address: Optional[str] = pnf_ipv4_address + self.pnf_ipv6_address: Optional[str] = pnf_ipv6_address + + self._pnf: "Pnf" = None + + def __repr__(self) -> str: + """Pnf instance object representation. + + Returns: + str: Human readable pnf instance representation + + """ + return f"PnfInstance(pnf_name={self.pnf_name})" + + @classmethod + def get_all_url(cls) -> str: # pylint: disable=arguments-differ + """Return an url to get all pnfs. + + Returns: + str: Url to get all pnfs + + """ + return f"{cls.base_url}{cls.api_version}/network/pnfs/" + + @classmethod + def get_all(cls) -> Iterator["PnfInstance"]: + """Get all PNF instances. + + Yields: + PnfInstance: Pnf instance + + """ + for pnf_data in cls.send_message_json( \ + "GET", \ + "Get all pnf instances", \ + cls.get_all_url() \ + ).get("pnf", []): + yield cls.create_from_api_response(pnf_data, None) + + @property + def url(self) -> str: + """Network instance url. + + Returns: + str: NetworkInstance url + + """ + return f"{self.base_url}{self.api_version}/network/pnfs/pnf/{self.pnf_name}" + + @property + def pnf(self) -> "Pnf": + """Pnf associated with that pnf instance. + + Raises: + ResourceNotFound: Could not find PNF for that PNF instance + + Returns: + Pnf: Pnf object associated with Pnf instance + + """ + if not self._pnf: + for pnf in self.service_instance.sdc_service.pnfs: + if pnf.model_version_id == self.model_version_id: + self._pnf = pnf + return self._pnf + + msg = ( + f'Could not find PNF for the PNF instance' + f' with model version ID "{self.model_version_id}"' + ) + raise ResourceNotFound(msg) + return self._pnf + + @classmethod + def create_from_api_response(cls, api_response: dict, + service_instance: "ServiceInstance") -> "PnfInstance": + """Create pnf instance object using HTTP API response dictionary. + + Args: + api_response (dict): A&AI API response dictionary + service_instance (ServiceInstance): Service instance with which network is related + + Returns: + PnfInstance: PnfInstance object + + """ + return cls(service_instance=service_instance, + pnf_name=api_response["pnf-name"], + in_maint=api_response["in-maint"], + selflink=api_response.get("selflink"), + pnf_id=api_response.get("pnf-id"), + equip_type=api_response.get("equip-type"), + equip_vendor=api_response.get("equip-vendor"), + equip_model=api_response.get("equip-model"), + management_option=api_response.get("management-option"), + orchestration_status=api_response.get("orchestration-status"), + ipaddress_v4_oam=api_response.get("ipaddress-v4-oam"), + sw_version=api_response.get("sw-version"), + frame_id=api_response.get("frame-id"), + serial_number=api_response.get("serial-number"), + ipaddress_v4_loopback_0=api_response.get("ipaddress-v4-loopback-0"), + ipaddress_v6_loopback_0=api_response.get("ipaddress-v6-loopback-0"), + ipaddress_v4_aim=api_response.get("ipaddress-v4-aim"), + ipaddress_v6_aim=api_response.get("ipaddress-v6-aim"), + ipaddress_v6_oam=api_response.get("ipaddress-v6-oam"), + inv_status=api_response.get("inv-status"), + resource_version=api_response.get("resource-version"), + prov_status=api_response.get("prov-status"), + nf_role=api_response.get("nf-role"), + admin_status=api_response.get("admin-status"), + operational_status=api_response.get("operational-status"), + model_customization_id=api_response.get("model-customization-id"), + model_invariant_id=api_response.get("model-invariant-id"), + model_version_id=api_response.get("model-version-id"), + pnf_ipv4_address=api_response.get("pnf-ipv4-address"), + pnf_ipv6_address=api_response.get("pnf-ipv6-address")) + + def delete(self, a_la_carte: bool = True) -> None: + """Delete Pnf instance. + + PNF deletion it's just A&AI resource deletion. That's difference between another instances. + You don't have to wait for that task finish, because it's not async task. + + """ + self._logger.debug("Delete %s pnf", self.pnf_name) + self.send_message("DELETE", + f"Delete {self.pnf_name} PNF", + f"{self.url}?resource-version={self.resource_version}") diff --git a/src/onapsdk/aai/business/project.py b/src/onapsdk/aai/business/project.py new file mode 100644 index 0000000..989444a --- /dev/null +++ b/src/onapsdk/aai/business/project.py @@ -0,0 +1,123 @@ +"""A&AI project 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. + +from typing import Any, Dict, Iterator + +from onapsdk.utils.jinja import jinja_env + +from ..aai_element import AaiResource + + +class Project(AaiResource): + """Project class.""" + + def __init__(self, name: str, resource_version: str) -> None: + """Project object initialization. + + Args: + name (str): Project name + resource_version (str): resource version + """ + super().__init__() + self.name: str = name + self.resource_version: str = resource_version + + @classmethod + def get_all(cls) -> Iterator["Project"]: + """Get all project. + + Yields: + Project: Project object + + """ + url: str = cls.get_all_url() + for project in cls.send_message_json("GET", + "Get A&AI projects", + url).get("project", []): + yield cls( + project.get("project-name"), + project.get("resource-version") + ) + + @classmethod + def get_all_url(cls) -> str: # pylint: disable=arguments-differ + """Return url to get all projects. + + Returns: + str: Url to get all projects + + """ + return f"{cls.base_url}{cls.api_version}/business/projects" + + def __repr__(self) -> str: + """Project object representation. + + Returns: + str: Project object representation + + """ + return f"Project(name={self.name})" + + @property + def url(self) -> str: + """Project's url. + + Returns: + str: Resource's url + + """ + return (f"{self.base_url}{self.api_version}/business/projects/" + f"project/{self.name}") + + @classmethod + def create(cls, name: str) -> "Project": + """Create project A&AI resource. + + Args: + name (str): project name + + Returns: + Project: Created Project object + + """ + cls.send_message( + "PUT", + "Declare A&AI project", + (f"{cls.base_url}{cls.api_version}/business/projects/" + f"project/{name}"), + data=jinja_env().get_template("aai_project_create.json.j2").render( + project_name=name + ) + ) + return cls.get_by_name(name) + + @classmethod + def get_by_name(cls, name: str) -> "Project": + """Get project resource by it's name. + + Raises: + ResourceNotFound: Project requested by a name does not exist. + + Returns: + Project: Project requested by a name. + + """ + url = (f"{cls.base_url}{cls.api_version}/business/projects/" + f"project/{name}") + response: Dict[str, Any] = \ + cls.send_message_json("GET", + f"Get {name} project", + url) + return cls(response["project-name"], response["resource-version"]) diff --git a/src/onapsdk/aai/business/service.py b/src/onapsdk/aai/business/service.py new file mode 100644 index 0000000..fe3b34d --- /dev/null +++ b/src/onapsdk/aai/business/service.py @@ -0,0 +1,484 @@ +"""Service instance 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. + +from typing import Iterator, Type, Union, Iterable, Optional + +from onapsdk.exceptions import StatusError, ParameterError +from onapsdk.sdc.service import Service +from onapsdk.so.deletion import ServiceDeletionRequest +from onapsdk.so.instantiation import NetworkInstantiation, VnfInstantiation +from onapsdk.utils.jinja import jinja_env + +from .instance import Instance +from .network import NetworkInstance +from .pnf import PnfInstance +from .vnf import VnfInstance + + +class ServiceInstance(Instance): # pylint: disable=too-many-instance-attributes + """Service instanve class.""" + + def __init__(self, # pylint: disable=too-many-arguments, too-many-locals + service_subscription: "ServiceSubscription", + instance_id: str, + instance_name: str = None, + service_type: str = None, + service_role: str = None, + environment_context: str = None, + workload_context: str = None, + created_at: str = None, + updated_at: str = None, + resource_version: str = None, + description: str = None, + model_invariant_id: str = None, + model_version_id: str = None, + persona_model_version: str = None, + widget_model_id: str = None, + widget_model_version: str = None, + bandwith_total: str = None, + vhn_portal_url: str = None, + service_instance_location_id: str = None, + selflink: str = None, + orchestration_status: str = None, + input_parameters: str = None) -> None: + """Service instance object initialization. + + Args: + service_subscription (ServiceSubscription): service subscription which is belongs to + instance_id (str): Uniquely identifies this instance of a service + instance_name (str, optional): This field will store a name assigned to + the service-instance. Defaults to None. + service_type (str, optional): String capturing type of service. Defaults to None. + service_role (str, optional): String capturing the service role. Defaults to None. + environment_context (str, optional): This field will store the environment context + assigned to the service-instance. Defaults to None. + workload_context (str, optional): This field will store the workload context assigned to + the service-instance. Defaults to None. + created_at (str, optional): Create time of Network Service. Defaults to None. + updated_at (str, optional): Last update of Network Service. Defaults to None. + description (str, optional): Short description for service-instance. Defaults to None. + model_invariant_id (str, optional): The ASDC model id for this resource or + service model. Defaults to None. + model_version_id (str, optional): The ASDC model version for this resource or + service model. Defaults to None. + persona_model_version (str, optional): The ASDC model version for this resource or + service model. Defaults to None. + widget_model_id (str, optional): The ASDC data dictionary widget model. This maps + directly to the A&AI widget. Defaults to None. + widget_model_version (str, optional): The ASDC data dictionary version of the widget + model. This maps directly to the A&AI version of the widget. Defaults to None. + bandwith_total (str, optional): Indicates the total bandwidth to be used for this + service. Defaults to None. + vhn_portal_url (str, optional): URL customers will use to access the vHN Portal. + Defaults to None. + service_instance_location_id (str, optional): An identifier that customers assign to + the location where this service is being used. Defaults to None. + resource_version (str, optional): Used for optimistic concurrency. Must be empty on + create, valid on update and delete. Defaults to None. + selflink (str, optional): Path to the controller object. Defaults to None. + orchestration_status (str, optional): Orchestration status of this service. + Defaults to None. + input_parameters (str, optional): String capturing request parameters from SO to + pass to Closed Loop. Defaults to None. + """ + super().__init__(resource_version=resource_version, + model_invariant_id=model_invariant_id, + model_version_id=model_version_id) + self.service_subscription: "ServiceSubscription" = service_subscription + self.instance_id: str = instance_id + self.instance_name: str = instance_name + self.service_type: str = service_type + self.service_role: str = service_role + self.environment_context: str = environment_context + self.workload_context: str = workload_context + self.created_at: str = created_at + self.updated_at: str = updated_at + self.description: str = description + self.bandwith_total: str = bandwith_total + self.vhn_portal_url: str = vhn_portal_url + self.service_instance_location_id: str = service_instance_location_id + self.selflink: str = selflink + self.orchestration_status: str = orchestration_status + self.input_parameters: str = input_parameters + self.persona_model_version: str = persona_model_version + self.widget_model_id: str = widget_model_id + self.widget_model_version: str = widget_model_version + self._sdc_service: Optional[Service] = None + + def __repr__(self) -> str: + """Service instance object representation. + + Returns: + str: Human readable service instance representation + + """ + return (f"ServiceInstance(instance_id={self.instance_id}, " + f"instance_name={self.instance_name})") + + def _get_related_instance(self, + related_instance_class: Union[Type[NetworkInstance], + Type[VnfInstance]], + relationship_related_to_type: str) -> Iterator[\ + Union[NetworkInstance, + VnfInstance]]: + """Iterate through related service instances. + + This is method which for given `relationship_related_to_type` creates iterator + it iterate through objects which are related with service. + + Args: + related_instance_class (Union[Type[NetworkInstance], Type[VnfInstance]]): Class object + to create required object instances + relationship_related_to_type (str): Has to be "generic-vnf" or "l3-network" + + Raises: + ParameterError: relationship_related_to_type does not satisfy the requirements + + Yields: + Iterator[ Union[NetworkInstance, VnfInstance]]: [description] + + """ + if not relationship_related_to_type in ["l3-network", "generic-vnf", "pnf"]: + msg = ( + f'Invalid "relationship_related_to_type" value. ' + f'Provided "{relationship_related_to_type}". ' + f'Has to be "l3-network" or "generic-vnf".' + ) + raise ParameterError(msg) + for relationship in self.relationships: + if relationship.related_to == relationship_related_to_type: + yield related_instance_class.create_from_api_response(\ + self.send_message_json("GET", + (f"Get {self.instance_id} " + f"{related_instance_class.__class__.__name__}"), + f"{self.base_url}{relationship.related_link}"), + self) + + @classmethod + def create(cls, service_subscription: "ServiceSubscription", # pylint: disable=too-many-arguments, too-many-locals + instance_id: str, + instance_name: str = None, + service_type: str = None, + service_role: str = None, + environment_context: str = None, + workload_context: str = None, + created_at: str = None, + updated_at: str = None, + resource_version: str = None, + description: str = None, + model_invariant_id: str = None, + model_version_id: str = None, + persona_model_version: str = None, + widget_model_id: str = None, + widget_model_version: str = None, + bandwith_total: str = None, + vhn_portal_url: str = None, + service_instance_location_id: str = None, + selflink: str = None, + orchestration_status: str = None, + input_parameters: str = None): + """Service instance creation. + + Args: + service_subscription (ServiceSubscription): service subscription which is belongs to + instance_id (str): Uniquely identifies this instance of a service + instance_name (str, optional): This field will store a name assigned to + the service-instance. Defaults to None. + service_type (str, optional): String capturing type of service. Defaults to None. + service_role (str, optional): String capturing the service role. Defaults to None. + environment_context (str, optional): This field will store the environment context + assigned to the service-instance. Defaults to None. + workload_context (str, optional): This field will store the workload context assigned to + the service-instance. Defaults to None. + created_at (str, optional): Create time of Network Service. Defaults to None. + updated_at (str, optional): Last update of Network Service. Defaults to None. + description (str, optional): Short description for service-instance. Defaults to None. + model_invariant_id (str, optional): The ASDC model id for this resource or + service model. Defaults to None. + model_version_id (str, optional): The ASDC model version for this resource or + service model. Defaults to None. + persona_model_version (str, optional): The ASDC model version for this resource or + service model. Defaults to None. + widget_model_id (str, optional): The ASDC data dictionary widget model. This maps + directly to the A&AI widget. Defaults to None. + widget_model_version (str, optional): The ASDC data dictionary version of the widget + model. This maps directly to the A&AI version of the widget. Defaults to None. + bandwith_total (str, optional): Indicates the total bandwidth to be used for this + service. Defaults to None. + vhn_portal_url (str, optional): URL customers will use to access the vHN Portal. + Defaults to None. + service_instance_location_id (str, optional): An identifier that customers assign to + the location where this service is being used. Defaults to None. + resource_version (str, optional): Used for optimistic concurrency. Must be empty on + create, valid on update and delete. Defaults to None. + selflink (str, optional): Path to the controller object. Defaults to None. + orchestration_status (str, optional): Orchestration status of this service. + Defaults to None. + input_parameters (str, optional): String capturing request parameters from SO to + pass to Closed Loop. Defaults to None. + """ + service_instance: "ServiceInstance" = cls( + service_subscription, + instance_id, + instance_name, + service_type, + service_role, + environment_context, + workload_context, + created_at, + updated_at, + resource_version, + description, + model_invariant_id, + model_version_id, + persona_model_version, + widget_model_id, + widget_model_version, + bandwith_total, + vhn_portal_url, + service_instance_location_id, + selflink, + orchestration_status, + input_parameters + ) + cls.send_message("PUT", + f"Create service instance {instance_id} for "\ + f"{service_subscription.service_type} service subscription", + f"{service_subscription.url}/service-instances/service-instance/"\ + f"{instance_id}", + data=jinja_env() + .get_template("aai_service_instance_create.json.j2") + .render( + service_instance=service_instance + )) + return service_instance + + @classmethod + def get_all_url(cls, service_subscription: "ServiceSubscription") -> str: # pylint: disable=arguments-differ + """Return an url to get all service instances for service subscription. + + Args: + service_subscription (ServiceSubscription): Service subscription object + + Returns: + str: Url to get all service instances for service subscription + + """ + return f"{service_subscription.url}/service-instances/" + + @property + def url(self) -> str: + """Service instance resource URL. + + Returns: + str: Service instance url + + """ + return ( + f"{self.service_subscription.url}/service-instances/service-instance/{self.instance_id}" + ) + + @property + def vnf_instances(self) -> Iterator[VnfInstance]: + """Vnf instances associated with service instance. + + Returns iterator of VnfInstances representing VNF instantiated for that service + + Yields: + VnfInstance: VnfInstance object + + """ + return self._get_related_instance(VnfInstance, "generic-vnf") + + @property + def network_instances(self) -> Iterator[NetworkInstance]: + """Network instances associated with service instance. + + Returns iterator of NetworkInstance representing network instantiated for that service + + Yields: + NetworkInstance: NetworkInstance object + + """ + return self._get_related_instance(NetworkInstance, "l3-network") + + @property + def pnfs(self) -> Iterator[PnfInstance]: + """Pnfs associated with service instance. + + Returns iterator of PnfInstance representing pnfs instantiated for that service + + Yields: + PnfInstance: PnfInstance object + + """ + return self._get_related_instance(PnfInstance, "pnf") + + @property + def sdc_service(self) -> Service: + """Sdc service related with that instance. + + Sdc service model which was used to create that instance. + + Raises: + ResourceNotFound: Service model not found + + """ + if not self._sdc_service: + self._sdc_service = Service.get_by_unique_uuid(self.model_invariant_id) + return self._sdc_service + + @property + def active(self) -> bool: + """Information if service instance's orchestration status is active.""" + return self.orchestration_status == "Active" + + def add_vnf(self, # pylint: disable=too-many-arguments + vnf: "Vnf", + line_of_business: "LineOfBusiness", + platform: "Platform", + cloud_region: "CloudRegion" = None, + tenant: "Tenant" = None, + vnf_instance_name: str = None, + vnf_parameters: Iterable["InstantiationParameter"] = None, + so_vnf: "SoServiceVnf" = None, + a_la_carte: bool = True + ) -> "VnfInstantiation": + """Add vnf into service instance. + + Instantiate VNF. + + Args: + vnf (Vnf): Vnf from service configuration to instantiate + line_of_business (LineOfBusiness): LineOfBusiness to use in instantiation request + platform (Platform): Platform to use in instantiation request + cloud_region (CloudRegion, optional): Cloud region to use in instantiation request. + Defaults to None. + THAT PROPERTY WILL BE REQUIRED IN ONE OF THE FUTURE RELEASE. REFACTOR YOUR CODE + TO USE IT!. + tenant (Tenant, optional): Tenant to use in instnatiation request. + Defaults to None. + THAT PROPERTY WILL BE REQUIRED IN ONE OF THE FUTURE RELEASE. REFACTOR YOUR CODE + TO USE IT!. + vnf_instance_name (str, optional): VNF instantion name. + If no value is provided it's going to be + "Python_ONAP_SDK_vnf_instance_{str(uuid4())}". + Defaults to None. + vnf_parameters (Iterable[InstantiationParameter], optional): InstantiationParameter to + be passed as "userParams". Defaults to None. + so_vnf: (SoServiceVnf, optional): object with vnf instance parameters. Defaults to None. + a_la_carte (bool): instantiation type for vnf. Defaults to True. + + Raises: + StatusError: Service orchestration status is not "Active". + + Returns: + VnfInstantiation: VnfInstantiation request object + + """ + if not self.active: + raise StatusError('Service orchestration status must be "Active"') + + if a_la_carte: + return VnfInstantiation.instantiate_ala_carte( + self, + vnf, + line_of_business, + platform, + cloud_region=cloud_region, + tenant=tenant, + vnf_instance_name=vnf_instance_name, + vnf_parameters=vnf_parameters, + sdc_service=self.sdc_service + ) + + return VnfInstantiation.instantiate_macro( + self, + vnf, + line_of_business, + platform, + cloud_region=cloud_region, + tenant=tenant, + vnf_instance_name=vnf_instance_name, + vnf_parameters=vnf_parameters, + so_vnf=so_vnf, + sdc_service=self.sdc_service + ) + + def add_network(self, # pylint: disable=too-many-arguments + network: "Network", + line_of_business: "LineOfBusiness", + platform: "Platform", + cloud_region: "CloudRegion" = None, + tenant: "Tenant" = None, + network_instance_name: str = None, + subnets: Iterator["Subnet"] = None) -> "NetworkInstantiation": + """Add network into service instance. + + Instantiate vl. + + Args: + network (Network): Network from service configuration to instantiate + line_of_business (LineOfBusiness): LineOfBusiness to use in instantiation request + platform (Platform): Platform to use in instantiation request + cloud_region (CloudRegion, optional): Cloud region to use in instantiation request. + Defaults to None. + THAT PROPERTY WILL BE REQUIRED IN ONE OF THE FUTURE RELEASE. REFACTOR YOUR CODE + TO USE IT!. + tenant (Tenant, optional): Tenant to use in instnatiation request. + Defaults to None. + THAT PROPERTY WILL BE REQUIRED IN ONE OF THE FUTURE RELEASE. REFACTOR YOUR CODE + TO USE IT!. + network_instance_name (str, optional): Network instantion name. + If no value is provided it's going to be + "Python_ONAP_SDK_network_instance_{str(uuid4())}". + Defaults to None. + + Raises: + StatusError: Service orchestration status is not "Active" + + Returns: + NetworkInstantiation: NetworkInstantiation request object + + """ + if not self.active: + msg = f'Service orchestration status must be "Active"' + raise StatusError(msg) + + return NetworkInstantiation.instantiate_ala_carte( + self, + network, + line_of_business, + platform, + cloud_region=cloud_region, + tenant=tenant, + network_instance_name=network_instance_name, + subnets=subnets + ) + + def delete(self, a_la_carte: bool = True) -> "ServiceDeletionRequest": + """Create service deletion request. + + Send a request to delete service instance + + Args: + a_la_carte (boolean): deletion mode + + Returns: + ServiceDeletionRequest: Deletion request object + + """ + self._logger.debug("Delete %s service instance", self.instance_id) + return ServiceDeletionRequest.send_request(self, a_la_carte) diff --git a/src/onapsdk/aai/business/sp_partner.py b/src/onapsdk/aai/business/sp_partner.py new file mode 100644 index 0000000..05d6a05 --- /dev/null +++ b/src/onapsdk/aai/business/sp_partner.py @@ -0,0 +1,176 @@ +"""A&AI sp-partner 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. + +from typing import Iterator, Optional + +from onapsdk.utils.jinja import jinja_env + +from ..aai_element import AaiResource + + +class SpPartner(AaiResource): # pylint: disable=too-many-instance-attributes + """Sp partner class.""" + + def __init__(self, sp_partner_id: str, resource_version: str, url: str = None, # pylint: disable=too-many-arguments, too-many-locals + callsource: str = None, operational_status: str = None, + model_customization_id: str = None, model_invariant_id: str = None, + model_version_id: str = None) -> None: + """Sp partner object initialization. + + Args: + sp_partner_id (str): Uniquely identifies this sp-partner by id + resource_version (str): resource version + url (str, optional): Store the URL of this sp-partner. Defaults to None + callsource (str, optional): Store the callsource of this sp-partner. Defaults to None + operational_status (str, optional): Store the operational-status of this sp-partner. + Defaults to None + model_customization_id (str, optional): Store the model-customization-id + of this sp-partner. Defaults to None + model_invariant_id (str, optional): The ASDC model id for this sp-partner model. + Defaults to None + model_version_id (str, optional): The ASDC model version for this sp-partner model. + Defaults to None + + """ + super().__init__() + self.sp_partner_id: str = sp_partner_id + self.resource_version: str = resource_version + self.sp_partner_url: Optional[str] = url + self.callsource: Optional[str] = callsource + self.operational_status: Optional[str] = operational_status + self.model_customization_id: Optional[str] = model_customization_id + self.model_invariant_id: Optional[str] = model_invariant_id + self.model_version_id: Optional[str] = model_version_id + + def __repr__(self) -> str: + """Sp partner object representation. + + Returns: + str: SpPartner object representation + + """ + return f"SpPartner(sp_partner_id={self.sp_partner_id})" + + @property + def url(self) -> str: + """Sp partner's url. + + Returns: + str: Resource's url + + """ + return (f"{self.base_url}{self.api_version}/business/sp-partners/" + f"sp-partner/{self.sp_partner_id}") + + @classmethod + def get_all_url(cls) -> str: # pylint: disable=arguments-differ + """Return url to get all sp partners. + + Returns: + str: Url to get all sp partners + + """ + return f"{cls.base_url}{cls.api_version}/business/sp-partners" + + @classmethod + def get_all(cls) -> Iterator["SpPartner"]: + """Get all sp partners. + + Yields: + SpPartner: SpPartner object + + """ + url: str = cls.get_all_url() + for sp_partner in cls.send_message_json("GET", + "Get A&AI sp-partners", + url).get("sp-partner", []): + yield cls( + sp_partner["sp-partner-id"], + sp_partner["resource-version"], + sp_partner.get("url"), + sp_partner.get("callsource"), + sp_partner.get("operational-status"), + sp_partner.get("model-customization-id"), + sp_partner.get("model-invariant-id"), + sp_partner.get("model-version-id"), + ) + + @classmethod + def create(cls, sp_partner_id: str, url: str = "", callsource: str = "", # pylint: disable=too-many-arguments + operational_status: str = "", model_customization_id: str = "", + model_invariant_id: str = "", model_version_id: str = "") -> "SpPartner": + """Create sp partner A&AI resource. + + Args: + sp_partner_id (str): sp partner unique ID + url (str, optional): Store the URL of this sp-partner. Defaults to None + callsource (str, optional): Store the callsource of this sp-partner. Defaults to None + operational_status (str, optional): Store the operational-status of this sp-partner. + Defaults to None + model_customization_id (str, optional): Store the model-customization-id + of this sp-partner. Defaults to None + model_invariant_id (str, optional): The ASDC model id for this sp-partner model. + Defaults to None + model_version_id (str, optional): The ASDC model version for this sp-partner model. + Defaults to None + + Returns: + SpPartner: Created SpPartner object + + """ + cls.send_message( + "PUT", + "Declare A&AI sp partner", + (f"{cls.base_url}{cls.api_version}/business/sp-partners/" + f"sp-partner/{sp_partner_id}"), + data=jinja_env().get_template("aai_sp_partner_create.json.j2").render( + sp_partner_id=sp_partner_id, + url=url, + callsource=callsource, + operational_status=operational_status, + model_customization_id=model_customization_id, + model_invariant_id=model_invariant_id, + model_version_id=model_version_id + ) + ) + return cls.get_by_sp_partner_id(sp_partner_id) + + @classmethod + def get_by_sp_partner_id(cls, sp_partner_id: str) -> "SpPartner": + """Get sp partner by it's ID. + + Args: + sp_partner_id (str): sp partner object id + + Returns: + SpPartner: SpPartner object + + """ + response: dict = cls.send_message_json( + "GET", + "Get A&AI sp partner", + (f"{cls.base_url}{cls.api_version}/business/sp-partners/" + f"sp-partner/{sp_partner_id}") + ) + return cls( + response["sp-partner-id"], + response["resource-version"], + response.get("url"), + response.get("callsource"), + response.get("operational-status"), + response.get("model-customization-id"), + response.get("model-invariant-id"), + response.get("model-version-id") + ) diff --git a/src/onapsdk/aai/business/vf_module.py b/src/onapsdk/aai/business/vf_module.py new file mode 100644 index 0000000..ac91560 --- /dev/null +++ b/src/onapsdk/aai/business/vf_module.py @@ -0,0 +1,199 @@ +"""VF module instance.""" +# 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. + +from onapsdk.so.deletion import VfModuleDeletionRequest +from onapsdk.exceptions import ResourceNotFound + +from .instance import Instance + + +class VfModuleInstance(Instance): # pylint: disable=too-many-instance-attributes + """Vf module instance class.""" + + def __init__(self, # pylint: disable=too-many-arguments, too-many-locals + vnf_instance: "VnfInstance", + vf_module_id: str, + is_base_vf_module: bool, + automated_assignment: bool, + vf_module_name: str = None, + heat_stack_id: str = None, + resource_version: str = None, + model_invariant_id: str = None, + orchestration_status: str = None, + persona_model_version: str = None, + model_version_id: str = None, + model_customization_id: str = None, + widget_model_id: str = None, + widget_model_version: str = None, + contrail_service_instance_fqdn: str = None, + module_index: int = None, + selflink: str = None) -> None: + """Vf module initialization. + + Args: + vnf_instance (VnfInstance): VnfInstance + vf_module_id (str): Unique ID of vf-module + is_base_vf_module (bool): used to indicate whether or not this object is base vf module + automated_assignment (bool): ndicates whether vf-module assignment was done via + automation or manually + vf_module_name (str, optional): Name of vf-module. Defaults to None. + heat_stack_id (str, optional): Heat stack id corresponding to this instance. + Defaults to None. + orchestration_status (str, optional): orchestration status of this vf-module, + mastered by MSO. Defaults to None. + resource_version (str, optional): Used for optimistic concurrency. + Must be empty on create, valid on update and delete. Defaults to None. + model_invariant_id (str, optional): the ASDC model id for this resource or + service model. Defaults to None. + model_version_id (str, optional): the ASDC model version for this resource or + service model. Defaults to None. + persona_model_version (str, optional): the ASDC model version for this resource or + service model. Defaults to None. + model_customization_id (str, optional): captures the id of all the configuration + used to customize the resource for the service. Defaults to None. + widget_model_id (str, optional): the ASDC data dictionary widget model. + This maps directly to the A&AI widget. Defaults to None. + widget_model_version (str, optional): the ASDC data dictionary version of + the widget model. This maps directly to the A&AI version of the widget. + Defaults to None. + contrail_service_instance_fqdn (str, optional): the Contrail unique ID + for a service-instance. Defaults to None. + module_index (int, optional): the index will track the number of modules + of a given type that have been deployed in a VNF, starting with 0, + and always choosing the lowest available digit. Defaults to None. + selflink (str, optional): Path to the controller object. Defaults to None. + """ + super().__init__(resource_version=resource_version, model_version_id=model_version_id, + model_invariant_id=model_invariant_id) + self.vnf_instance: "VnfInstance" = vnf_instance + self.vf_module_id: str = vf_module_id + self.is_base_vf_module: bool = is_base_vf_module + self.automated_assignment: bool = automated_assignment + self.vf_module_name: str = vf_module_name + self.heat_stack_id: str = heat_stack_id + self.orchestration_status: str = orchestration_status + self.model_customization_id: str = model_customization_id + self.contrail_service_instance_fqdn: str = contrail_service_instance_fqdn + self.module_index: int = module_index + self.selflink: str = selflink + self.persona_model_version: str = persona_model_version + self.widget_model_id: str = widget_model_id + self.widget_model_version: str = widget_model_version + + self._vf_module: "VfModule" = None + + def __repr__(self) -> str: + """Object represetation. + + Returns: + str: Human readble VfModuleInstance representation + + """ + return (f"VfModuleInstance(vf_module_id={self.vf_module_id}, " + f"is_base_vf_module={self.is_base_vf_module}, " + f"automated_assignment={self.automated_assignment})") + + @classmethod + def get_all_url(cls, vnf_instance: "VnfInstance") -> str: # pylint: disable=arguments-differ + """Return url to get all vf modules for vnf instance. + + Args: + vnf_instance (VnfInstance): VNF instance object + + Returns: + str: Url to get all vf modules for vnf instance + + """ + return f"{vnf_instance.url}/vf-modules/" + + @property + def url(self) -> str: + """Resource url. + + Returns: + str: VfModuleInstance url + + """ + return f"{self.vnf_instance.url}/vf-modules/vf-module/{self.vf_module_id}" + + @property + def vf_module(self) -> "VfModule": + """Vf module associated with that vf module instance. + + Returns: + VfModule: VfModule object associated with vf module instance + + """ + if not self._vf_module: + for vf_module in self.vnf_instance.vnf.vf_modules: + if vf_module.model_version_id == self.model_version_id: + self._vf_module = vf_module + return self._vf_module + + msg = ( + f'Could not find VF modules for the VF Module instance' + f' with model version ID "{self.model_version_id}"' + ) + raise ResourceNotFound(msg) + return self._vf_module + + @classmethod + def create_from_api_response(cls, + api_response: dict, + vnf_instance: "VnfInstance") -> "VfModuleInstance": + """Create vf module instance object using HTTP API response dictionary. + + Args: + api_response (dict): HTTP API response content + vnf_instance (VnfInstance): VnfInstance associated with VfModuleInstance + + Returns: + VfModuleInstance: VfModuleInstance object + + """ + return cls( + vnf_instance=vnf_instance, + vf_module_id=api_response.get("vf-module-id"), + is_base_vf_module=api_response.get("is-base-vf-module"), + automated_assignment=api_response.get("automated-assignment"), + vf_module_name=api_response.get("vf-module-name"), + heat_stack_id=api_response.get("heat-stack-id"), + orchestration_status=api_response.get("orchestration-status"), + resource_version=api_response.get("resource-version"), + model_invariant_id=api_response.get("model-invariant-id"), + model_version_id=api_response.get("model-version-id"), + persona_model_version=api_response.get("persona-model-version"), + model_customization_id=api_response.get("model-customization-id"), + widget_model_id=api_response.get("widget-model-id"), + widget_model_version=api_response.get("widget-model-version"), + contrail_service_instance_fqdn=api_response.get("contrail-service-instance-fqdn"), + module_index=api_response.get("module-index"), + selflink=api_response.get("selflink") + ) + + def delete(self, a_la_carte: bool = True) -> "VfModuleDeletionRequest": + """Create deletion request. + + Send request to delete VF module instance + + Args: + a_la_carte (boolean): deletion mode + + Returns: + VfModuleDeletionRequest: Deletion request object + + """ + self._logger.debug("Delete %s VF module", self.vf_module_id) + return VfModuleDeletionRequest.send_request(self, a_la_carte) diff --git a/src/onapsdk/aai/business/vnf.py b/src/onapsdk/aai/business/vnf.py new file mode 100644 index 0000000..2045291 --- /dev/null +++ b/src/onapsdk/aai/business/vnf.py @@ -0,0 +1,536 @@ +"""Vnf instance 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. + +from typing import Iterable, Iterator + +from onapsdk.exceptions import ResourceNotFound, StatusError +from onapsdk.so.deletion import VnfDeletionRequest +from onapsdk.so.instantiation import VfModuleInstantiation, VnfInstantiation, SoService, \ + InstantiationParameter, VnfOperation +from onapsdk.configuration import settings + +from .instance import Instance +from .vf_module import VfModuleInstance + + +class VnfInstance(Instance): # pylint: disable=too-many-instance-attributes + """VNF Instance class.""" + + def __init__(self, # pylint: disable=too-many-arguments, too-many-locals + service_instance: "ServiceInstance", + vnf_id: str, + vnf_type: str, + in_maint: bool, + is_closed_loop_disabled: bool, + vnf_name: str = None, + service_id: str = None, + regional_resource_zone: str = None, + prov_status: str = None, + operational_status: str = None, + equipment_role: str = None, + orchestration_status: str = None, + vnf_package_name: str = None, + vnf_discriptor_name: str = None, + job_id: str = None, + heat_stack_id: str = None, + mso_catalog_key: str = None, + management_option: str = None, + ipv4_oam_address: str = None, + ipv4_loopback0_address: str = None, + nm_lan_v6_address: str = None, + management_v6_address: str = None, + vcpu: int = None, + vcpu_units: str = None, + vmemory: int = None, + vmemory_units: str = None, + vdisk: int = None, + vdisk_units: str = None, + nshd: int = None, + nvm: int = None, + nnet: int = None, + resource_version: str = None, + encrypted_access_flag: bool = None, + model_invariant_id: str = None, + model_version_id: str = None, + persona_model_version: str = None, + model_customization_id: str = None, + widget_model_id: str = None, + widget_model_version: str = None, + as_number: str = None, + regional_resource_subzone: str = None, + nf_type: str = None, + nf_function: str = None, + nf_role: str = None, + nf_naming_code: str = None, + selflink: str = None, + ipv4_oam_gateway_address: str = None, + ipv4_oam_gateway_address_prefix_length: int = None, + vlan_id_outer: int = None, + nm_profile_name: str = None) -> None: + """Vnf instance object initialization. + + Args: + vnf_id (str): Unique id of VNF. This is unique across the graph. + vnf_type (str): String capturing type of vnf, that was intended to identify + the ASDC resource. This field has been overloaded in service-specific ways and + clients should expect changes to occur in the future to this field as ECOMP + matures. + in_maint (bool): used to indicate whether or not this object is in maintenance mode + (maintenance mode = true). This field (in conjunction with prov-status) + is used to suppress alarms and vSCL on VNFs/VMs. + is_closed_loop_disabled (bool): used to indicate whether closed loop function is + enabled on this node + vnf_name (str, optional): Name of VNF. Defaults to None. + service_id (str, optional): Unique identifier of service, does not necessarily map to + ASDC service models. Defaults to None. + regional_resource_zone (str, optional): Regional way of organizing pservers, source of + truth should define values. Defaults to None. + prov_status (str, optional): Trigger for operational monitoring of this resource by + Service Assurance systems. Defaults to None. + operational_status (str, optional): Indicator for whether the resource is considered + operational. Valid values are in-service-path and out-of-service-path. + Defaults to None. + equipment_role (str, optional): Client should send valid enumerated value. + Defaults to None. + orchestration_status (str, optional): Orchestration status of this VNF, used by MSO. + Defaults to None. + vnf_package_name (str, optional): vnf package name. Defaults to None. + vnf_discriptor_name (str, optional): vnf discriptor name. Defaults to None. + job_id (str, optional): job id corresponding to vnf. Defaults to None. + heat_stack_id (str, optional): Heat stack id corresponding to this instance, + managed by MSO. Defaults to None. + mso_catalog_key (str, optional): Corresponds to the SDN-C catalog id used to + configure this VCE. Defaults to None. + management_option (str, optional): identifier of managed customer. Defaults to None. + ipv4_oam_address (str, optional): Address tail-f uses to configure generic-vnf, + also used for troubleshooting and is IP used for traps generated by generic-vnf. + Defaults to None. + ipv4_loopback0_address (str, optional): v4 Loopback0 address. Defaults to None. + nm_lan_v6_address (str, optional): v6 Loopback address. Defaults to None. + management_v6_address (str, optional): v6 management address. Defaults to None. + vcpu (int, optional): number of vcpus ordered for this instance of VNF, + used for VNFs with no vservers/flavors, to be used only by uCPE. Defaults to None. + vcpu_units (str, optional): units associated with vcpu, used for VNFs with no + vservers/flavors, to be used only by uCPE. Defaults to None. + vmemory (int, optional): number of GB of memory ordered for this instance of VNF, + used for VNFs with no vservers/flavors, to be used only by uCPE. Defaults to None. + vmemory_units (str, optional): units associated with vmemory, used for VNFs with + no vservers/flavors, to be used only by uCPE. Defaults to None. + vdisk (int, optional): number of vdisks ordered for this instance of VNF, + used for VNFs with no vservers/flavors, to be used only uCPE. Defaults to None. + vdisk_units (str, optional): units associated with vdisk, used for VNFs with + no vservers/flavors, to be used only by uCPE. Defaults to None. + nshd (int, optional): number of associated SHD in vnf. Defaults to None. + nvm (int, optional): number of vms in vnf. Defaults to None. + nnet (int, optional): number of network in vnf. Defaults to None. + resource_version (str, optional): Used for optimistic concurrency. + Must be empty on create, valid on update and delete. Defaults to None. + encrypted_access_flag (bool, optional): indicates whether generic-vnf access uses SSH. + Defaults to None. + model_invariant_id (str, optional): the ASDC model id for this resource or + service model. Defaults to None. + model_version_id (str, optional): the ASDC model version for this resource or + service model. Defaults to None. + persona_model_version (str, optional): the ASDC model version for this resource or + service model. Defaults to None. + model_customization_id (str, optional): captures the id of all the configuration used + to customize the resource for the service. Defaults to None. + widget_model_id (str, optional): the ASDC data dictionary widget model. This maps + directly to the A&AI widget. Defaults to None. + widget_model_version (str, optional): the ASDC data dictionary version of + the widget model.This maps directly to the A&AI version of the widget. + Defaults to None. + as_number (str, optional): as-number of the VNF. Defaults to None. + regional_resource_subzone (str, optional): represents sub zone of the rr plane. + Defaults to None. + nf_type (str, optional): Generic description of the type of NF. Defaults to None. + nf_function (str, optional): English description of Network function that + the specific VNF deployment is providing. Defaults to None. + nf_role (str, optional): role in the network that this model will be providing. + Defaults to None. + nf_naming_code (str, optional): string assigned to this model used for naming purposes. + Defaults to None. + selflink (str, optional): Path to the controller object. Defaults to None. + ipv4_oam_gateway_address (str, optional): Gateway address. Defaults to None. + ipv4_oam_gateway_address_prefix_length (int, optional): Prefix length for oam-address. + Defaults to None. + vlan_id_outer (int, optional): Temporary location for S-TAG to get to VCE. + Defaults to None. + nm_profile_name (str, optional): Network Management profile of this VNF. + Defaults to None. + + """ + super().__init__(resource_version=resource_version, + model_invariant_id=model_invariant_id, + model_version_id=model_version_id) + self.service_instance: "ServiceInstance" = service_instance + self.vnf_id: str = vnf_id + self.vnf_type: str = vnf_type + self.in_maint: bool = in_maint + self.is_closed_loop_disabled: bool = is_closed_loop_disabled + self.vnf_name: str = vnf_name + self.service_id: str = service_id + self.regional_resource_zone: str = regional_resource_zone + self.prov_status: str = prov_status + self.operational_status: str = operational_status + self.equipment_role: str = equipment_role + self.orchestration_status: str = orchestration_status + self.vnf_package_name: str = vnf_package_name + self.vnf_discriptor_name: str = vnf_discriptor_name + self.job_id: str = job_id + self.heat_stack_id: str = heat_stack_id + self.mso_catalog_key: str = mso_catalog_key + self.management_option: str = management_option + self.ipv4_oam_address: str = ipv4_oam_address + self.ipv4_loopback0_address: str = ipv4_loopback0_address + self.nm_lan_v6_address: str = nm_lan_v6_address + self.management_v6_address: str = management_v6_address + self.vcpu: int = vcpu + self.vcpu_units: str = vcpu_units + self.vmemory: int = vmemory + self.vmemory_units: str = vmemory_units + self.vdisk: int = vdisk + self.vdisk_units: str = vdisk_units + self.nshd: int = nshd + self.nvm: int = nvm + self.nnet: int = nnet + self.encrypted_access_flag: bool = encrypted_access_flag + self.model_customization_id: str = model_customization_id + self.as_number: str = as_number + self.regional_resource_subzone: str = regional_resource_subzone + self.nf_type: str = nf_type + self.nf_function: str = nf_function + self.nf_role: str = nf_role + self.nf_naming_code: str = nf_naming_code + self.selflink: str = selflink + self.ipv4_oam_gateway_address: str = ipv4_oam_gateway_address + self.ipv4_oam_gateway_address_prefix_length: int = ipv4_oam_gateway_address_prefix_length + self.vlan_id_outer: int = vlan_id_outer + self.nm_profile_name: str = nm_profile_name + self.persona_model_version: str = persona_model_version + self.widget_model_id: str = widget_model_id + self.widget_model_version: str = widget_model_version + + self._vnf: "Vnf" = None + + def __repr__(self) -> str: + """Vnf instance object representation. + + Returns: + str: Human readable vnf instance representation + + """ + return (f"VnfInstance(vnf_id={self.vnf_id}, vnf_type={self.vnf_type}, " + f"in_maint={self.in_maint}, " + f"is_closed_loop_disabled={self.is_closed_loop_disabled})") + + @classmethod + def get_all_url(cls) -> str: # pylint: disable=arguments-differ + """Return url to get all vnfs. + + Returns: + str: Url to get all vnfs + + """ + return f"{cls.base_url}{cls.api_version}/network/generic-vnfs/" + + @property + def url(self) -> str: + """Vnf instance url. + + Returns: + str: VnfInstance url + + """ + return f"{self.base_url}{self.api_version}/network/generic-vnfs/generic-vnf/{self.vnf_id}" + + @property + def vf_modules(self) -> Iterator[VfModuleInstance]: + """Vf modules associated with vnf instance. + + Yields: + VfModuleInstance: VfModuleInstance associated with VnfInstance + + """ + for vf_module in self.send_message_json("GET", + f"GET VNF {self.vnf_name} VF modules", + f"{self.url}/vf-modules").get("vf-module", []): + yield VfModuleInstance.create_from_api_response(vf_module, self) + + @property + def vnf(self) -> "Vnf": + """Vnf associated with that vnf instance. + + Raises: + ResourceNotFound: Could not find VNF for that VNF instance + + Returns: + Vnf: Vnf object associated with vnf instance + + """ + if not self._vnf: + for vnf in self.service_instance.sdc_service.vnfs: + if vnf.model_version_id == self.model_version_id: + self._vnf = vnf + return self._vnf + + msg = ( + f'Could not find VNF for the VNF instance' + f' with model version ID "{self.model_version_id}"' + ) + raise ResourceNotFound(msg) + return self._vnf + + @classmethod + def create_from_api_response(cls, api_response: dict, + service_instance: "ServiceInstance") -> "VnfInstance": + """Create vnf instance object using HTTP API response dictionary. + + Returns: + VnfInstance: VnfInstance object + + """ + return cls(service_instance=service_instance, + vnf_id=api_response.get("vnf-id"), + vnf_type=api_response.get("vnf-type"), + in_maint=api_response.get("in-maint"), + is_closed_loop_disabled=api_response.get("is-closed-loop-disabled"), + vnf_name=api_response.get("vnf-name"), + service_id=api_response.get("service-id"), + regional_resource_zone=api_response.get("regional-resource-zone"), + prov_status=api_response.get("prov-status"), + operational_status=api_response.get("operational-status"), + equipment_role=api_response.get("equipment-role"), + orchestration_status=api_response.get("orchestration-status"), + vnf_package_name=api_response.get("vnf-package-name"), + vnf_discriptor_name=api_response.get("vnf-discriptor-name"), + job_id=api_response.get("job-id"), + heat_stack_id=api_response.get("heat-stack-id"), + mso_catalog_key=api_response.get("mso-catalog-key"), + management_option=api_response.get("management-option"), + ipv4_oam_address=api_response.get("ipv4-oam-address"), + ipv4_loopback0_address=api_response.get("ipv4-loopback0-address"), + nm_lan_v6_address=api_response.get("nm-lan-v6-address"), + management_v6_address=api_response.get("management-v6-address"), + vcpu=api_response.get("vcpu"), + vcpu_units=api_response.get("vcpu-units"), + vmemory=api_response.get("vmemory"), + vmemory_units=api_response.get("vmemory-units"), + vdisk=api_response.get("vdisk"), + vdisk_units=api_response.get("vdisk-units"), + nshd=api_response.get("nshd"), + nvm=api_response.get("nvm"), + nnet=api_response.get("nnet"), + resource_version=api_response.get("resource-version"), + model_invariant_id=api_response.get("model-invariant-id"), + model_version_id=api_response.get("model-version-id"), + encrypted_access_flag=api_response.get("encrypted-access-flag"), + persona_model_version=api_response.get("persona-model-version"), + model_customization_id=api_response.get("model-customization-id"), + widget_model_id=api_response.get("widget-model-id"), + widget_model_version=api_response.get("widget-model-version"), + as_number=api_response.get("as-number"), + regional_resource_subzone=api_response.get("regional-resource-subzone"), + nf_type=api_response.get("nf-type"), + nf_function=api_response.get("nf-function"), + nf_role=api_response.get("nf-role"), + nf_naming_code=api_response.get("nf-naming-code"), + selflink=api_response.get("selflink"), + ipv4_oam_gateway_address=api_response.get("ipv4-oam-gateway-address"), + ipv4_oam_gateway_address_prefix_length=\ + api_response.get("ipv4-oam-gateway-address-prefix-length"), + vlan_id_outer=api_response.get("vlan-id-outer"), + nm_profile_name=api_response.get("nm-profile-name")) + + def add_vf_module(self, # pylint: disable=too-many-arguments + vf_module: "VfModule", + cloud_region: "CloudRegion" = None, + tenant: "Tenant" = None, + vf_module_instance_name: str = None, + vnf_parameters: Iterable["InstantiationParameter"] = None, + use_preload: bool = True + ) -> "VfModuleInstantiation": + """Instantiate vf module for that VNF instance. + + Args: + vf_module (VfModule): VfModule to instantiate + cloud_region (CloudRegion, optional): Cloud region to use in instantiation request. + Defaults to None. + THAT PROPERTY WILL BE REQUIRED IN ONE OF THE FUTURE RELEASE. REFACTOR YOUR CODE + TO USE IT!. + tenant (Tenant, optional): Tenant to use in instnatiation request. + Defaults to None. + THAT PROPERTY WILL BE REQUIRED IN ONE OF THE FUTURE RELEASE. REFACTOR YOUR CODE + TO USE IT!. + vf_module_instance_name (str, optional): VfModule instance name. Defaults to None. + vnf_parameters (Iterable[InstantiationParameter], optional): InstantiationParameter + to use for preloading or to be passed as "userParams". Defaults to None. + use_preload (bool, optional): Based on this flag InstantiationParameters are passed + in preload or as "userParam" in the request. Defaults to True + + Returns: + VfModuleInstantiation: VfModuleInstantiation request object + + """ + return VfModuleInstantiation.instantiate_ala_carte( + vf_module, + self, + cloud_region=cloud_region, + tenant=tenant, + vf_module_instance_name=vf_module_instance_name, + vnf_parameters=vnf_parameters, + use_preload=use_preload + ) + + def update(self, + vnf_parameters: Iterable["InstantiationParameter"] = None + ) -> VnfInstantiation: + """Update vnf instance. + + Args: + vnf_parameters (Iterable["InstantiationParameter"], Optional): list of instantiation + parameters for update operation. + Raises: + StatusError: Skip post instantiation configuration flag for VF to True. + It might cause problems with SO component. + + Returns: + VnfInstantiation: VnfInstantiation object. + + """ + skip_flag = next(p for p in self.vnf.properties + if p.name == 'skip_post_instantiation_configuration') + if not skip_flag.value or skip_flag.value != "false": + raise StatusError("Operation for the vnf is not supported! " + "Skip_post_instantiation_configuration flag for VF should be False") + + return self._execute_so_action(operation_type=VnfOperation.UPDATE, + vnf_parameters=vnf_parameters) + + def healthcheck(self) -> VnfInstantiation: + """Execute healthcheck operation for vnf instance. + + Returns: + VnfInstantiation: VnfInstantiation object. + + """ + return self._execute_so_action(operation_type=VnfOperation.HEALTHCHECK) + + def _execute_so_action(self, + operation_type: VnfOperation, + vnf_parameters: Iterable["InstantiationParameter"] = None + ) -> VnfInstantiation: + """Execute SO workflow for selected operation. + + Args: + operation_type (str): Name of the operation to execute. + vnf_parameters (Iterable["InstantiationParameter"], Optional): list of instantiation + parameters for update operation. + + Returns: + VnfInstantiation: VnfInstantiation object. + + """ + if not self.service_instance.active: + msg = f'Service orchestration status must be "Active"' + raise StatusError(msg) + + lob = settings.LOB + platform = settings.PLATFORM + + for relationship in self.relationships: + if relationship.related_to == "line-of-business": + lob = relationship.relationship_data.pop().get("relationship-value") + if relationship.related_to == "platform": + platform = relationship.relationship_data.pop().get("relationship-value") + + so_input = self._build_so_input(vnf_params=vnf_parameters) + + return VnfInstantiation.so_action( + vnf_instance=self, + operation_type=operation_type, + aai_service_instance=self.service_instance, + line_of_business=lob, + platform=platform, + sdc_service=self.service_instance.sdc_service, + so_service=so_input + ) + + def _build_so_input(self, vnf_params: Iterable[InstantiationParameter] = None) -> SoService: + """Prepare so_input with params retrieved from existing service instance. + + Args: + vnf_params (Iterable[InstantiationParameter], Optional): list of instantiation + parameters for update operation. + + Returns: + SoService: SoService object to store SO Service parameters used for macro instantiation. + + """ + so_vnfs = [] + so_pnfs = [] + + if not vnf_params: + vnf_params = [] + + for pnf in self.service_instance.pnfs: + _pnf = { + "model_name": pnf.pnf.model_name, + "instance_name": pnf.pnf_name + } + + so_pnfs.append(_pnf) + + for vnf in self.service_instance.vnf_instances: + _vnf = {"model_name": vnf.vnf.model_name, + "instance_name": vnf.vnf_name, + "parameters": {}} + if vnf.vnf_name == self.vnf_name: + for _param in vnf_params: + _vnf["parameters"][_param.name] = _param.value + + _vf_modules = [] + for vf_module in vnf.vf_modules: + _vf_module = { + "model_name": vf_module.vf_module.model_name.split('..')[1], + "instance_name": vf_module.vf_module_name, + "parameters": {} + } + + _vf_modules.append(_vf_module) + + _vnf["vf_modules"] = _vf_modules + so_vnfs.append(_vnf) + + return SoService.load(data={ + 'subscription_service_type': self.service_instance.service_subscription.service_type, + 'vnfs': so_vnfs, + 'pnfs': so_pnfs + }) + + def delete(self, a_la_carte: bool = True) -> "VnfDeletionRequest": + """Create VNF deletion request. + + Send request to delete VNF instance + + Args: + a_la_carte (boolean): deletion mode + + Returns: + VnfDeletionRequest: Deletion request + + """ + self._logger.debug("Delete %s VNF", self.vnf_id) + return VnfDeletionRequest.send_request(self, a_la_carte) |