diff options
Diffstat (limited to 'src')
32 files changed, 196 insertions, 203 deletions
diff --git a/src/onapsdk/aai/aai_element.py b/src/onapsdk/aai/aai_element.py index 2bfb565..613eb70 100644 --- a/src/onapsdk/aai/aai_element.py +++ b/src/onapsdk/aai/aai_element.py @@ -105,7 +105,7 @@ class AaiElement(OnapService): }) @classmethod - def get_guis(cls) -> GuiItem: + def get_guis(cls) -> GuiList: """Retrieve the status of the AAI GUIs. Only one GUI is referenced for AAI diff --git a/src/onapsdk/aai/business/customer.py b/src/onapsdk/aai/business/customer.py index 1305a12..6df3f7a 100644 --- a/src/onapsdk/aai/business/customer.py +++ b/src/onapsdk/aai/business/customer.py @@ -28,9 +28,9 @@ from .service import ServiceInstance 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 + cloud_owner: Optional[str] = None + cloud_region_id: Optional[str] = None + tenant_id: Optional[str] = None @dataclass diff --git a/src/onapsdk/aai/business/network.py b/src/onapsdk/aai/business/network.py index e36cf62..f5d5d42 100644 --- a/src/onapsdk/aai/business/network.py +++ b/src/onapsdk/aai/business/network.py @@ -21,7 +21,7 @@ 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 + def __init__(self, # NOSONAR # pylint: disable=too-many-arguments, too-many-locals service_instance: "ServiceInstance", network_id: str, is_bound_to_vpn: bool, @@ -49,7 +49,7 @@ class NetworkInstance(Instance): # pylint: disable=too-many-instance-attributes widget_model_version: str = None, selflink: str = None, operational_status: str = None, - is_trunked: bool = None): + is_trunked: bool = None) -> None: """Network instance object initialization. Args: diff --git a/src/onapsdk/aai/business/pnf.py b/src/onapsdk/aai/business/pnf.py index 9f5394f..d8ce746 100644 --- a/src/onapsdk/aai/business/pnf.py +++ b/src/onapsdk/aai/business/pnf.py @@ -25,7 +25,7 @@ if TYPE_CHECKING: class PnfInstance(Instance): # pylint: disable=too-many-instance-attributes """Pnf instance class.""" - def __init__(self, # pylint: disable=too-many-arguments, too-many-locals + def __init__(self, # NOSONAR # pylint: disable=too-many-arguments, too-many-locals service_instance: "ServiceInstance", pnf_name: str, in_maint: bool, diff --git a/src/onapsdk/aai/business/service.py b/src/onapsdk/aai/business/service.py index 12b5b4f..9c2e7f8 100644 --- a/src/onapsdk/aai/business/service.py +++ b/src/onapsdk/aai/business/service.py @@ -30,7 +30,9 @@ 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 + ACTIVE_STATUS_MESSAGE = 'Service orchestration status must be "Active"' + + def __init__(self, # NOSONAR # pylint: disable=too-many-arguments, too-many-locals service_subscription: "ServiceSubscription", instance_id: str, instance_name: str = None, @@ -150,7 +152,7 @@ class ServiceInstance(Instance): # pylint: disable=too-many-instance-attributes Iterator[ Union[NetworkInstance, VnfInstance]]: [description] """ - if not relationship_related_to_type in ["l3-network", "generic-vnf", "pnf"]: + if relationship_related_to_type not in ["l3-network", "generic-vnf", "pnf"]: msg = ( f'Invalid "relationship_related_to_type" value. ' f'Provided "{relationship_related_to_type}". ' @@ -167,7 +169,7 @@ class ServiceInstance(Instance): # pylint: disable=too-many-instance-attributes self) @classmethod - def create(cls, service_subscription: "ServiceSubscription", # pylint: disable=too-many-arguments, too-many-locals + def create(cls, service_subscription: "ServiceSubscription", # NOSONAR # pylint: disable=too-many-arguments, too-many-locals instance_id: str, instance_name: str = None, service_type: str = None, @@ -389,7 +391,7 @@ class ServiceInstance(Instance): # pylint: disable=too-many-instance-attributes """ if not self.active: - raise StatusError('Service orchestration status must be "Active"') + raise StatusError(self.ACTIVE_STATUS_MESSAGE) if a_la_carte: return VnfInstantiation.instantiate_ala_carte( @@ -447,7 +449,7 @@ class ServiceInstance(Instance): # pylint: disable=too-many-instance-attributes """ if not self.active: - raise StatusError('Service orchestration status must be "Active"') + raise StatusError(self.ACTIVE_STATUS_MESSAGE) return PnfInstantiation.instantiate_macro( self, @@ -497,7 +499,7 @@ class ServiceInstance(Instance): # pylint: disable=too-many-instance-attributes """ if not self.active: - raise StatusError('Service orchestration status must be "Active"') + raise StatusError(self.ACTIVE_STATUS_MESSAGE) return NetworkInstantiation.instantiate_ala_carte( self, diff --git a/src/onapsdk/aai/business/vf_module.py b/src/onapsdk/aai/business/vf_module.py index ac91560..817ba38 100644 --- a/src/onapsdk/aai/business/vf_module.py +++ b/src/onapsdk/aai/business/vf_module.py @@ -22,7 +22,7 @@ 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 + def __init__(self, # NOSONAR # pylint: disable=too-many-arguments, too-many-locals vnf_instance: "VnfInstance", vf_module_id: str, is_base_vf_module: bool, diff --git a/src/onapsdk/aai/business/vnf.py b/src/onapsdk/aai/business/vnf.py index 31762c9..48a4726 100644 --- a/src/onapsdk/aai/business/vnf.py +++ b/src/onapsdk/aai/business/vnf.py @@ -28,7 +28,7 @@ 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 + def __init__(self, # NOSONAR # pylint: disable=too-many-arguments, too-many-locals service_instance: "ServiceInstance", vnf_id: str, vnf_type: str, diff --git a/src/onapsdk/aai/cloud_infrastructure/cloud_region.py b/src/onapsdk/aai/cloud_infrastructure/cloud_region.py index 240b7a2..ec78d96 100644 --- a/src/onapsdk/aai/cloud_infrastructure/cloud_region.py +++ b/src/onapsdk/aai/cloud_infrastructure/cloud_region.py @@ -36,8 +36,8 @@ class AvailabilityZone: name: str hypervisor_type: str - operational_status: str = None - resource_version: str = None + operational_status: Optional[str] = None + resource_version: Optional[str] = None @dataclass @@ -49,22 +49,22 @@ class EsrSystemInfo: # pylint: disable=too-many-instance-attributes password: str system_type: str resource_version: str - system_name: str = None - esr_type: str = None - vendor: str = None - version: str = None - service_url: str = None - protocol: str = None - ssl_cacert: str = None + system_name: Optional[str] = None + esr_type: Optional[str] = None + vendor: Optional[str] = None + version: Optional[str] = None + service_url: Optional[str] = None + protocol: Optional[str] = None + ssl_cacert: Optional[str] = None ssl_insecure: Optional[bool] = None - ip_address: str = None - port: str = None - cloud_domain: str = None - default_tenant: str = None + ip_address: Optional[str] = None + port: Optional[str] = None + cloud_domain: Optional[str] = None + default_tenant: Optional[str] = None passive: Optional[bool] = None - remote_path: str = None - system_status: str = None - openstack_region_id: str = None + remote_path: Optional[str] = None + system_status: Optional[str] = None + openstack_region_id: Optional[str] = None class CloudRegion(AaiResource, AaiResourceLinkToComplexMixin, AaiResourceLinkToProjectMixin): # pylint: disable=too-many-instance-attributes @@ -73,7 +73,7 @@ class CloudRegion(AaiResource, AaiResourceLinkToComplexMixin, AaiResourceLinkToP Represents A&AI cloud region object. """ - def __init__(self, # pylint: disable=too-many-arguments + def __init__(self, # NOSONAR # pylint: disable=too-many-arguments cloud_owner: str, cloud_region_id: str, orchestration_disabled: bool, @@ -558,7 +558,7 @@ class CloudRegion(AaiResource, AaiResourceLinkToComplexMixin, AaiResourceLinkToP availability_zone_operational_status=availability_zone_operational_status) ) - def add_esr_system_info(self, # pylint: disable=too-many-arguments, too-many-locals + def add_esr_system_info(self, # NOSONAR # pylint: disable=too-many-arguments, too-many-locals esr_system_info_id: str, user_name: str, password: str, diff --git a/src/onapsdk/aai/cloud_infrastructure/complex.py b/src/onapsdk/aai/cloud_infrastructure/complex.py index 6b2b74f..3cfd3cb 100644 --- a/src/onapsdk/aai/cloud_infrastructure/complex.py +++ b/src/onapsdk/aai/cloud_infrastructure/complex.py @@ -27,7 +27,7 @@ class Complex(AaiResource, AaiResourceLinkToGeoRegionMixin): # pylint: disable= Collection of physical locations that can house cloud-regions. """ - def __init__(self, # pylint: disable=too-many-locals, too-many-arguments + def __init__(self, # NOSONAR # pylint: disable=too-many-locals, too-many-arguments physical_location_id: str, *, name: str = "", @@ -124,7 +124,7 @@ class Complex(AaiResource, AaiResourceLinkToGeoRegionMixin): # pylint: disable= f"{self.physical_location_id}") @classmethod - def create(cls, # pylint: disable=too-many-locals, too-many-arguments + def create(cls, # NOSONAR # pylint: disable=too-many-locals, too-many-arguments physical_location_id: str, *, name: str = "", @@ -189,7 +189,7 @@ class Complex(AaiResource, AaiResourceLinkToGeoRegionMixin): # pylint: disable= return complex_object @classmethod - def update(cls, # pylint: disable=too-many-locals, too-many-arguments + def update(cls, # NOSONAR # pylint: disable=too-many-locals, too-many-arguments physical_location_id: str, *, name: str = "", diff --git a/src/onapsdk/aai/network/site_resource.py b/src/onapsdk/aai/network/site_resource.py index 6804e65..4e1780a 100644 --- a/src/onapsdk/aai/network/site_resource.py +++ b/src/onapsdk/aai/network/site_resource.py @@ -25,7 +25,7 @@ from ..mixins.link_to_project import AaiResourceLinkToProjectMixin class SiteResource(AaiResource, AaiResourceLinkToComplexMixin, AaiResourceLinkToProjectMixin): # pylint: disable=too-many-instance-attributes """Site resource class.""" - def __init__(self, # pylint: disable=too-many-locals, too-many-arguments + def __init__(self, # NOSONAR # pylint: disable=too-many-locals, too-many-arguments site_resource_id: str, *, site_resource_name: str = "", @@ -175,7 +175,7 @@ class SiteResource(AaiResource, AaiResourceLinkToComplexMixin, AaiResourceLinkTo resource_version=site_resource_data.get("resource-version", "")) @classmethod - def create(cls, # pylint: disable=too-many-arguments + def create(cls, # NOSONAR # pylint: disable=too-many-arguments site_resource_id: str, site_resource_name: Optional[str] = None, description: Optional[str] = None, diff --git a/src/onapsdk/cds/blueprint.py b/src/onapsdk/cds/blueprint.py index 2d6aa37..2916977 100644 --- a/src/onapsdk/cds/blueprint.py +++ b/src/onapsdk/cds/blueprint.py @@ -241,9 +241,9 @@ class Workflow(CdsElement): self.name: str = cba_workflow_name self.workflow_data: dict = cba_workflow_data self.blueprint: "Blueprint" = blueprint - self._steps: List[self.WorkflowStep] = None - self._inputs: List[self.WorkflowInput] = None - self._outputs: List[self.WorkflowOutput] = None + self._steps: Optional[List[self.WorkflowStep]] = None + self._inputs: Optional[List[self.WorkflowInput]] = None + self._outputs: Optional[List[self.WorkflowOutput]] = None def __repr__(self) -> str: """Representation of object. @@ -425,7 +425,7 @@ class ResolvedTemplate(CdsElement): str: Retrieve resolved template url """ - params_dict: Dict[str, str] = urlencode(dict(filter(lambda item: item[1] is not None, { + params_dict: str = urlencode(dict(filter(lambda item: item[1] is not None, { "bpName": self.blueprint.metadata.template_name, "bpVersion": self.blueprint.metadata.template_version, "artifactName": self.artifact_name, @@ -534,9 +534,9 @@ class Blueprint(CdsElement): """ super().__init__() self.cba_file_bytes: bytes = cba_file_bytes - self._cba_metadata: CbaMetadata = None - self._cba_mappings: MappingSet = None - self._cba_workflows: List[Workflow] = None + self._cba_metadata: Optional[CbaMetadata] = None + self._cba_mappings: Optional[MappingSet] = None + self._cba_workflows: Optional[List[Workflow]] = None @property def url(self) -> str: diff --git a/src/onapsdk/cds/cds_element.py b/src/onapsdk/cds/cds_element.py index 7a4b9c0..973e559 100644 --- a/src/onapsdk/cds/cds_element.py +++ b/src/onapsdk/cds/cds_element.py @@ -30,7 +30,7 @@ class CdsElement(OnapService, ABC): auth: tuple = settings.CDS_AUTH @classmethod - def get_guis(cls) -> GuiItem: + def get_guis(cls) -> GuiList: """Retrieve the status of the CDS GUIs. Only one GUI is referenced for CDS: CDS UI diff --git a/src/onapsdk/configuration/global_settings.py b/src/onapsdk/configuration/global_settings.py index 59a9242..1412203 100644 --- a/src/onapsdk/configuration/global_settings.py +++ b/src/onapsdk/configuration/global_settings.py @@ -25,20 +25,20 @@ AAI_URL = "https://aai.api.sparky.simpledemo.onap.org:30233" AAI_API_VERSION = "v27" AAI_AUTH = "Basic QUFJOkFBSQ==" AAI_BULK_CHUNK = 30 -CDS_URL = "http://portal.api.simpledemo.onap.org:30449" +CDS_URL = "http://portal.api.simpledemo.onap.org:30449" # NOSONAR CDS_AUTH = ("ccsdkapps", "ccsdkapps") -CPS_URL = "http://portal.api.simpledemo.onap.org:8080" +CPS_URL = "http://portal.api.simpledemo.onap.org:8080" # NOSONAR CPS_AUTH = ("cpsuser", "cpsr0cks!") CPS_VERSION = "v2" MSB_URL = "https://msb.api.simpledemo.onap.org:30283" -K8SPLUGIN_URL = "http://k8splugin.api.simpledemo.onap.org:30455" +K8SPLUGIN_URL = "http://k8splugin.api.simpledemo.onap.org:30455" # NOSONAR SDC_BE_URL = "https://sdc.api.be.simpledemo.onap.org:30204" SDC_FE_URL = "https://sdc.api.fe.simpledemo.onap.org:30207" SDC_AUTH = "Basic YWFpOktwOGJKNFNYc3pNMFdYbGhhazNlSGxjc2UyZ0F3ODR2YW9HR21KdlV5MlU=" # pylint: disable=line-too-long SDNC_URL = "https://sdnc.api.simpledemo.onap.org:30267" SDNC_AUTH = "Basic YWRtaW46S3A4Yko0U1hzek0wV1hsaGFrM2VIbGNzZTJnQXc4NHZhb0dHbUp2VXkyVQ==" # pylint: disable=line-too-long -SO_CATALOG_DB_ADAPTER_URL = "http://so-catalog-db-adapter:8082" -SO_URL = "http://so.api.simpledemo.onap.org:30277" +SO_CATALOG_DB_ADAPTER_URL = "http://so-catalog-db-adapter:8082" # NOSONAR +SO_URL = "http://so.api.simpledemo.onap.org:30277" # NOSONAR SO_API_VERSION = "v7" SO_AUTH = "Basic SW5mcmFQb3J0YWxDbGllbnQ6cGFzc3dvcmQxJA==" SO_CAT_DB_AUTH = "Basic YnBlbDpwYXNzd29yZDEk" @@ -46,8 +46,8 @@ VID_URL = "https://vid.api.simpledemo.onap.org:30200" VID_API_VERSION = "/vid" CLAMP_URL = "https://clamp.api.simpledemo.onap.org:30258" CLAMP_AUTH = "Basic ZGVtb0BwZW9wbGUub3NhYWYub3JnOmRlbW8xMjM0NTYh" -VES_URL = "http://ves.api.simpledemo.onap.org:30417" -DMAAP_URL = "http://dmaap.api.simpledemo.onap.org:3904" +VES_URL = "http://ves.api.simpledemo.onap.org:30417" # NOSONAR +DMAAP_URL = "http://dmaap.api.simpledemo.onap.org:3904" # NOSONAR NBI_URL = "https://nbi.api.simpledemo.onap.org:30274" NBI_API_VERSION = "/nbi/api/v4" DCAEMOD_URL = "" diff --git a/src/onapsdk/k8s/definition.py b/src/onapsdk/k8s/definition.py index 305bca2..5e7eecf 100644 --- a/src/onapsdk/k8s/definition.py +++ b/src/onapsdk/k8s/definition.py @@ -72,7 +72,7 @@ class Profile(DefinitionBase): kubernetes_version: str, labels: dict = None, release_name: str = None, - extra_resource_types: list = None) -> None: + extra_resource_types: dict = None) -> None: """Profile object initialization. Args: @@ -83,7 +83,7 @@ class Profile(DefinitionBase): namespace (str): Namespace that service is created in kubernetes_version (str): Required Kubernetes version labels (dict): Extra Labels for k8s resources - extra_resource_types (list): Extra k8s resources types (GVK) for status monitoring + extra_resource_types (dict): Extra k8s resources types (GVK) for status monitoring """ super().__init__(rb_name, rb_version) self.profile_name: str = profile_name diff --git a/src/onapsdk/k8s/instance.py b/src/onapsdk/k8s/instance.py index 64726b9..f08733d 100644 --- a/src/onapsdk/k8s/instance.py +++ b/src/onapsdk/k8s/instance.py @@ -208,7 +208,7 @@ class Configuration(InstanceBase): """ config: dict = self.send_message_json( "GET", - "Get Configuration", + "Get Configuration by version", f"{self.url}/version/{config_version}" ) return self.__class__( @@ -253,7 +253,7 @@ class Configuration(InstanceBase): """ config: dict = self.send_message_json( "GET", - "Get Configuration", + "Get Configuration by tag", f"{self.url}/tag/{config_tag}" ) return self.__class__( @@ -526,7 +526,7 @@ class Instance(InstanceBase, QueryResourceStatusMixin): """ for config in self.send_message_json("GET", - "Get configurations", + "Get all configurations", f"{self.url}/config"): yield self.config_class( self.instance_id, @@ -552,7 +552,7 @@ class Instance(InstanceBase, QueryResourceStatusMixin): config: dict = self.send_message_json( "GET", - "Get Configuration", + "Get Configuration by name", url ) return self.config_class( diff --git a/src/onapsdk/msb/esr.py b/src/onapsdk/msb/esr.py index b29cfa4..4157f37 100644 --- a/src/onapsdk/msb/esr.py +++ b/src/onapsdk/msb/esr.py @@ -22,7 +22,7 @@ class ESR(MSB): base_url = f"{MSB.base_url}/api/aai-esr-server/v1/vims" @classmethod - def register_vim(cls, # pylint: disable=too-many-arguments + def register_vim(cls, # NOSONAR # pylint: disable=too-many-arguments cloud_owner: str, cloud_region_id: str, cloud_type: str, diff --git a/src/onapsdk/nbi/nbi.py b/src/onapsdk/nbi/nbi.py index 7afc718..ffcd62c 100644 --- a/src/onapsdk/nbi/nbi.py +++ b/src/onapsdk/nbi/nbi.py @@ -14,7 +14,7 @@ # limitations under the License. from abc import ABC from enum import Enum -from typing import Iterator +from typing import Iterator, Optional from uuid import uuid4 from onapsdk.aai.business.customer import Customer @@ -216,7 +216,7 @@ class Service(Nbi): service.get("href")) @property - def customer(self) -> Customer: + def customer(self) -> Optional[Customer]: """Service order Customer object. Returns: @@ -228,7 +228,7 @@ class Service(Nbi): return Customer.get_by_global_customer_id(self._customer_id) @property - def service_specification(self) -> ServiceSpecification: + def service_specification(self) -> Optional[ServiceSpecification]: """Service specification. Returns: @@ -322,7 +322,7 @@ class ServiceOrder(Nbi, WaitForFinishMixin): # pylint: disable=too-many-instanc f"service_instance_name={self.service_instance_name}, state={self.state})") @property - def customer(self) -> Customer: + def customer(self) -> Optional[Customer]: """Get customer object used in service order. Returns: @@ -337,7 +337,7 @@ class ServiceOrder(Nbi, WaitForFinishMixin): # pylint: disable=too-many-instanc return self._customer @property - def service_specification(self) -> ServiceSpecification: + def service_specification(self) -> Optional[ServiceSpecification]: """Service order service specification used in order item. Returns: diff --git a/src/onapsdk/onap_service.py b/src/onapsdk/onap_service.py index 7b1dc17..06c358c 100644 --- a/src/onapsdk/onap_service.py +++ b/src/onapsdk/onap_service.py @@ -32,6 +32,8 @@ from onapsdk.exceptions import (APIError, ConnectionFailed, InvalidResponse, urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) +APPLICATION_JSON: str = "application/json" + class OnapService(ABC): """ @@ -74,13 +76,13 @@ class OnapService(ABC): yield ph_call() _logger: logging.Logger = logging.getLogger(__qualname__) - server: str = None + server: Optional[str] = None headers: Dict[str, str] = { - "Content-Type": "application/json", - "Accept": "application/json", + "Content-Type": APPLICATION_JSON, + "Accept": APPLICATION_JSON, } patch_headers: Dict[str, str] = headers.copy() - proxy: Dict[str, str] = None + proxy: Optional[Dict[str, str]] = None permanent_headers: PermanentHeadersCollection = PermanentHeadersCollection() def __init_subclass__(cls): @@ -95,7 +97,7 @@ class OnapService(ABC): """Initialize the service.""" @classmethod - def send_message(cls, method: str, action: str, url: str, # pylint: disable=too-many-locals + def send_message(cls, method: str, action: str, url: str, # pylint: disable=too-many-locals # NOSONAR **kwargs) -> Union[requests.Response, None]: """ Send a message to an ONAP service. @@ -161,7 +163,7 @@ class OnapService(ABC): cls.server, action, response.text if (response is not None and response.headers.get("Content-Type", "") in \ - ["application/json", "text/plain"]) else "n/a") + [APPLICATION_JSON, "text/plain"]) else "n/a") response.raise_for_status() return response @@ -285,7 +287,7 @@ class OnapService(ABC): backoff_factor=backoff_factor, ) adapter = HTTPAdapter(max_retries=retry) - session.mount('http://', adapter) + session.mount('http://', adapter) # NOSONAR session.mount('https://', adapter) return session diff --git a/src/onapsdk/sdc/__init__.py b/src/onapsdk/sdc/__init__.py index f359344..cac1356 100644 --- a/src/onapsdk/sdc/__init__.py +++ b/src/onapsdk/sdc/__init__.py @@ -232,7 +232,7 @@ class SDC(OnapService, ABC): return True @classmethod - def get_guis(cls) -> GuiItem: + def get_guis(cls) -> GuiList: """Retrieve the status of the SDC GUIs. Only one GUI is referenced for SDC @@ -258,9 +258,9 @@ class SdcOnboardable(SDC, ABC): def __init__(self, name: str = None) -> None: """Initialize the object.""" super().__init__(name) - self._identifier: str = None - self._status: str = None - self._version: str = None + self._identifier: Optional[str] = None + self._status: Optional[str] = None + self._version: Optional[str] = None @property def identifier(self) -> str: diff --git a/src/onapsdk/sdc/category_management.py b/src/onapsdk/sdc/category_management.py index c2d63b4..c78a909 100644 --- a/src/onapsdk/sdc/category_management.py +++ b/src/onapsdk/sdc/category_management.py @@ -15,7 +15,7 @@ import json from abc import ABC, abstractmethod -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional from onapsdk.configuration import settings from onapsdk.exceptions import ResourceNotFound @@ -42,14 +42,14 @@ class BaseCategory(SDC, ABC): # pylint: disable=too-many-instance-attributes """ super().__init__(name) - self.normalized_name: str = None - self.unique_id: str = None - self.icons: List[str] = None - self.subcategories: List[Dict[str, str]] = None - self.version: str = None - self.owner_id: str = None - self.empty: bool = None - self.type: str = None + self.normalized_name: Optional[str] = None + self.unique_id: Optional[str] = None + self.icons: Optional[List[str]] = None + self.subcategories: Optional[List[Dict[str, str]]] = None + self.version: Optional[str] = None + self.owner_id: Optional[str] = None + self.empty: Optional[str] = None + self.type: Optional[str] = None @classmethod def _get_all_url(cls) -> str: @@ -235,8 +235,9 @@ class ResourceCategory(BaseCategory): category_obj: "ResourceCategory" = super().get(name=name) if not subcategory: return category_obj - filtered_subcategories: Dict[str, str] = list(filter(lambda x: x["name"] == subcategory, - category_obj.subcategories)) + filtered_subcategories: List[Dict[str, str]] = list( + filter(lambda x: x["name"] == subcategory, + category_obj.subcategories)) if not filtered_subcategories: raise ResourceNotFound(f"Subcategory {subcategory} does not exist.") category_obj.subcategories = filtered_subcategories diff --git a/src/onapsdk/sdc/properties.py b/src/onapsdk/sdc/properties.py index 519772c..5551efc 100644 --- a/src/onapsdk/sdc/properties.py +++ b/src/onapsdk/sdc/properties.py @@ -117,7 +117,7 @@ class Property: # pylint: disable=too-many-instance-attributes, too-few-public- return self.name == obj.name and self.property_type == obj.property_type @property - def input(self) -> Input: + def input(self) -> Optional[Input]: """Property input. Returns property Input object. diff --git a/src/onapsdk/sdc/sdc_resource.py b/src/onapsdk/sdc/sdc_resource.py index ea25b31..8d693ab 100644 --- a/src/onapsdk/sdc/sdc_resource.py +++ b/src/onapsdk/sdc/sdc_resource.py @@ -14,7 +14,7 @@ # limitations under the License. import logging from abc import ABC -from typing import Any, Dict, Iterator, List, Union +from typing import Any, Dict, Iterator, List, Optional, Union import base64 import time @@ -47,8 +47,8 @@ class SdcResource(SdcOnboardable, ABC): # pylint: disable=too-many-instance-att """Initialize the object.""" super().__init__(name) self.version_filter: str = version - self._unique_uuid: str = None - self._unique_identifier: str = None + self._unique_uuid: Optional[str] = None + self._unique_identifier: Optional[str] = None self._resource_type: str = "resources" self._properties_to_add: List[Property] = properties or [] self._inputs_to_add: Union[Property, NestedInput] = inputs or [] @@ -106,7 +106,7 @@ class SdcResource(SdcOnboardable, ABC): # pylint: disable=too-many-instance-att """Load Object information from SDC.""" self.exists() - def deep_load(self) -> None: + def deep_load(self) -> None: # NOSONAR """Deep load Object informations from SDC.""" url = ( f"{self.base_front_url}/sdc1/feProxy/rest/v1/" @@ -334,7 +334,7 @@ class SdcResource(SdcOnboardable, ABC): # pylint: disable=too-many-instance-att # pylint: disable=too-many-return-statements @staticmethod def _parse_sdc_status(sdc_status: str, distribution_state: str, - logger: logging.Logger) -> str: + logger: logging.Logger) -> Optional[str]: """ Parse SDC status in order to normalize it. diff --git a/src/onapsdk/sdc/service.py b/src/onapsdk/sdc/service.py index 7f195a5..0457215 100644 --- a/src/onapsdk/sdc/service.py +++ b/src/onapsdk/sdc/service.py @@ -170,15 +170,15 @@ class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too self.category_name = sdc_values["category"] self.resources = resources or [] self._instantiation_type: Optional[ServiceInstantiationType] = instantiation_type - self._distribution_id: str = None - self._distributed: bool = False + self._distribution_id: Optional[str] = None + self._distributed: Optional[bool] = False self._resource_type: str = "services" - self._tosca_model: bytes = None - self._tosca_template: str = None - self._vnfs: list = None - self._pnfs: list = None - self._networks: list = None - self._vf_modules: list = None + self._tosca_model: Optional[bytes] = None + self._tosca_template: Optional[str] = None + self._vnfs: Optional[list] = None + self._pnfs: Optional[list] = None + self._networks: Optional[list] = None + self._vf_modules: Optional[list] = None @classmethod def get_by_unique_uuid(cls, unique_uuid: str) -> "Service": @@ -332,33 +332,32 @@ class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too model_instance_name=self.name, component=component ) - if node_template_type is Vnf: - if component.group_instances: - for vf_module in component.group_instances: - if not any([property_def["name"] == "vf_module_label"] and \ - property_def["value"] == "base_template_dummy_ignore" for \ - property_def in vf_module["properties"]): - node_template.vf_modules.append(VfModule( - name=vf_module["name"], - group_type=vf_module["type"], - model_name=vf_module["groupName"], - model_version_id=vf_module["groupUUID"], - model_invariant_uuid=vf_module["invariantUUID"], - model_version=vf_module["version"], - model_customization_id=vf_module["customizationUUID"], - properties=( - Property( - name=property_def["name"], - property_type=property_def["type"], - description=property_def["description"], - value=property_def["value"] - ) for property_def in vf_module["properties"] \ - if property_def["value"] and not ( - property_def["name"] == "vf_module_label" and \ - property_def["value"] == "base_template_dummy_ignore" - ) - ) - )) + if node_template_type is Vnf and component.group_instances: + for vf_module in component.group_instances: + if not any([property_def["name"] == "vf_module_label"] and \ + property_def["value"] == "base_template_dummy_ignore" for \ + property_def in vf_module["properties"]): + node_template.vf_modules.append(VfModule( + name=vf_module["name"], + group_type=vf_module["type"], + model_name=vf_module["groupName"], + model_version_id=vf_module["groupUUID"], + model_invariant_uuid=vf_module["invariantUUID"], + model_version=vf_module["version"], + model_customization_id=vf_module["customizationUUID"], + properties=( + Property( + name=property_def["name"], + property_type=property_def["type"], + description=property_def["description"], + value=property_def["value"] + ) for property_def in vf_module["properties"] \ + if property_def["value"] and not ( + property_def["name"] == "vf_module_label" and \ + property_def["value"] == "base_template_dummy_ignore" + ) + ) + )) return node_template def __has_component_type(self, origin_type: str) -> bool: @@ -629,7 +628,7 @@ class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too except BadZipFile as exc: self._logger.exception(exc) - def _check_distributed(self) -> bool: + def _check_distributed(self) -> None: """Check if service is distributed and update status accordingly.""" url = f"{self._base_create_url()}/services/distribution/{self.distribution_id}" headers = headers_sdc_creator(SdcResource.headers) @@ -656,7 +655,7 @@ class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too self._distributed = True def _update_components_status(self, status: Dict[str, bool], - result: Response) -> Dict[str, bool]: + result: Dict[str, Any]) -> Dict[str, bool]: """Update components distribution status.""" distrib_list = result['distributionStatusList'] self._logger.debug("[SDC][Get Distribution] distrib_list = %s", diff --git a/src/onapsdk/sdc/vsp.py b/src/onapsdk/sdc/vsp.py index 07b8d1e..a600252 100644 --- a/src/onapsdk/sdc/vsp.py +++ b/src/onapsdk/sdc/vsp.py @@ -18,7 +18,7 @@ from typing import BinaryIO from typing import Callable from typing import Dict -from onapsdk.exceptions import APIError, ParameterError +from onapsdk.exceptions import ParameterError from onapsdk.sdc.sdc_element import SdcElement from onapsdk.sdc.vendor import Vendor import onapsdk.constants as const @@ -42,8 +42,8 @@ class Vsp(SdcElement): # pylint: disable=too-many-instance-attributes VSP_PATH = "vendor-software-products" headers = headers_sdc_creator(SdcElement.headers) - def __init__(self, name: str = None, package: BinaryIO = None, - vendor: Vendor = None): + def __init__(self, name: Optional[str] = None, package: Optional[BinaryIO] = None, + vendor: Optional[Vendor] = None): """ Initialize vsp object. @@ -52,10 +52,10 @@ class Vsp(SdcElement): # pylint: disable=too-many-instance-attributes """ super().__init__() - self._csar_uuid: str = None - self._vendor: Vendor = vendor or None + self._csar_uuid: Optional[str] = None + self._vendor: Optional[Vendor] = vendor self.name: str = name or "ONAP-test-VSP" - self.package = package or None + self.package: Optional[BinaryIO] = package @property def status(self): @@ -192,26 +192,11 @@ class Vsp(SdcElement): # pylint: disable=too-many-instance-attributes headers.pop("Content-Type") headers["Accept-Encoding"] = "gzip, deflate" data = {'upload': package_to_upload} - upload_result = self.send_message('POST', - 'upload ZIP for Vsp', - url, - headers=headers, - files=data) - if upload_result: - # TODO https://jira.onap.org/browse/SDC-3505 pylint: disable=W0511 - response_json = json.loads(upload_result.text) - if response_json["status"] != "Success": - self._logger.error( - "an error occured during file upload for Vsp %s", - self.name) - raise APIError(response_json) - # End TODO https://jira.onap.org/browse/SDC-3505 - self._logger.info("Files for Vsp %s have been uploaded", - self.name) - else: - self._logger.error( - "an error occured during file upload for Vsp %s", - self.name) + self.send_message('POST', + 'upload ZIP for Vsp', + url, + headers=headers, + files=data) def _validate_action(self): """Do validate for real.""" diff --git a/src/onapsdk/sdnc/sdnc_element.py b/src/onapsdk/sdnc/sdnc_element.py index 84f56d9..1644af7 100644 --- a/src/onapsdk/sdnc/sdnc_element.py +++ b/src/onapsdk/sdnc/sdnc_element.py @@ -23,7 +23,7 @@ class SdncElement(OnapService): base_url = settings.SDNC_URL @classmethod - def get_guis(cls) -> GuiItem: + def get_guis(cls) -> GuiList: """Retrieve the status of the SDNC GUIs. There are 2 GUIS diff --git a/src/onapsdk/sdnc/topology.py b/src/onapsdk/sdnc/topology.py index 2b79ec9..8cc3e5e 100644 --- a/src/onapsdk/sdnc/topology.py +++ b/src/onapsdk/sdnc/topology.py @@ -20,6 +20,12 @@ from onapsdk.utils.jinja import jinja_env from .sdnc_element import SdncElement +NETCONF_NODE_TOPOLOGY_HOST: str = "netconf-node-topology:host" +NETCONF_NODE_TOPOLOGY_PORT: str = "netconf-node-topology:port" +NETCONF_NODE_TOPOLOGY_USERNAME: str = "netconf-node-topology:username" +NETCONF_NODE_TOPOLOGY_PASSWORD: str = "netconf-node-topology:password" + + class Node(SdncElement): """SDNC Node.""" @@ -76,10 +82,10 @@ class Node(SdncElement): node_json_template = { "node": { "node-id": "", - "netconf-node-topology:host": "", - "netconf-node-topology:port": 0, - "netconf-node-topology:username": "", - "netconf-node-topology:password": "" + NETCONF_NODE_TOPOLOGY_HOST: "", + NETCONF_NODE_TOPOLOGY_PORT: 0, + NETCONF_NODE_TOPOLOGY_USERNAME: "", + NETCONF_NODE_TOPOLOGY_PASSWORD: "" } } self.send_message( @@ -109,10 +115,10 @@ class Node(SdncElement): node_json_template = { "node": { "node-id": "", - "netconf-node-topology:host": "", - "netconf-node-topology:port": 0, - "netconf-node-topology:username": "", - "netconf-node-topology:password": "" + NETCONF_NODE_TOPOLOGY_HOST: "", + NETCONF_NODE_TOPOLOGY_PORT: 0, + NETCONF_NODE_TOPOLOGY_USERNAME: "", + NETCONF_NODE_TOPOLOGY_PASSWORD: "" } } self.send_message( @@ -234,10 +240,10 @@ class Topology(SdncElement): try: node = node_object["network-topology:node"][0] return Node(node_id=node["node-id"], - host=node["netconf-node-topology:host"], - port=node["netconf-node-topology:port"], - username=node["netconf-node-topology:username"], - password=node["netconf-node-topology:password"], + host=node[NETCONF_NODE_TOPOLOGY_HOST], + port=node[NETCONF_NODE_TOPOLOGY_PORT], + username=node[NETCONF_NODE_TOPOLOGY_USERNAME], + password=node[NETCONF_NODE_TOPOLOGY_PASSWORD], topology_id=self.topology_id) except KeyError: self._logger.error("Error. Node creation skipped.") @@ -259,10 +265,10 @@ class Topology(SdncElement): for node in nodes: try: yield Node(node_id=node["node-id"], - host=node["netconf-node-topology:host"], - port=node["netconf-node-topology:port"], - username=node["netconf-node-topology:username"], - password=node["netconf-node-topology:password"], + host=node[NETCONF_NODE_TOPOLOGY_HOST], + port=node[NETCONF_NODE_TOPOLOGY_PORT], + username=node[NETCONF_NODE_TOPOLOGY_USERNAME], + password=node[NETCONF_NODE_TOPOLOGY_PASSWORD], topology_id=self.topology_id) except KeyError: self._logger.error("Error. Node creation skipped. KeyError") diff --git a/src/onapsdk/so/instantiation.py b/src/onapsdk/so/instantiation.py index 2ca25c2..084a112 100644 --- a/src/onapsdk/so/instantiation.py +++ b/src/onapsdk/so/instantiation.py @@ -136,8 +136,8 @@ class VnfParameters: """ name: str - vnf_parameters: Iterable["InstantiationParameter"] = None - vfmodule_parameters: Iterable["VfmoduleParameters"] = None + vnf_parameters: Optional[Iterable["InstantiationParameter"]] = None + vfmodule_parameters: Optional[Iterable["VfmoduleParameters"]] = None @dataclass class VfmoduleParameters: @@ -147,7 +147,7 @@ class VfmoduleParameters: """ name: str - vfmodule_parameters: Iterable["InstantiationParameter"] = None + vfmodule_parameters: Optional[Iterable["InstantiationParameter"]] = None @dataclass @@ -194,7 +194,7 @@ class Subnet: # pylint: disable=too-many-instance-attributes name: str start_address: str gateway_address: str - role: str = None + role: Optional[str] = None cidr_mask: str = "24" ip_version: str = "4" dhcp_enabled: bool = False @@ -384,7 +384,7 @@ class VnfInstantiation(NodeTemplateInstantiation): # pylint: disable=too-many-a self.vnf = vnf @classmethod - def create_from_request_response(cls, request_response: dict) -> "VnfInstantiation": + def create_from_request_response(cls, request_response: dict) -> "VnfInstantiation": # NOSONAR """Create VNF instantiation object based on request details. Raises: @@ -399,7 +399,7 @@ class VnfInstantiation(NodeTemplateInstantiation): # pylint: disable=too-many-a """ if request_response.get("request", {}).get("requestScope") == "vnf" and \ request_response.get("request", {}).get("requestType") == "createInstance": - service: SdcService = None + service: Optional[SdcService] = None for related_instance in request_response.get("request", {}).get("requestDetails", {})\ .get("relatedInstanceList", []): if related_instance.get("relatedInstance", {}).get("modelInfo", {})\ @@ -408,7 +408,7 @@ class VnfInstantiation(NodeTemplateInstantiation): # pylint: disable=too-many-a .get("modelInfo", {}).get("modelName")) if not service: raise ResourceNotFound("No related service in Vnf instance details response") - vnf: Vnf = None + vnf: Optional[Vnf] = None for service_vnf in service.vnfs: if service_vnf.name == request_response.get("request", {})\ .get("requestDetails", {}).get("modelInfo", {}).get("modelCustomizationName"): @@ -725,7 +725,7 @@ class PnfInstantiation(NodeTemplateInstantiation): # pylint: disable=too-many-a """ if request_response.get("request", {}).get("requestScope") == "pnf" and \ request_response.get("request", {}).get("requestType") == "createInstance": - service: SdcService = None + service: Optional[SdcService] = None for related_instance in request_response.get("request", {}).get("requestDetails", {})\ .get("relatedInstanceList", []): if related_instance.get("relatedInstance", {}).get("modelInfo", {})\ @@ -734,7 +734,7 @@ class PnfInstantiation(NodeTemplateInstantiation): # pylint: disable=too-many-a .get("modelInfo", {}).get("modelName")) if not service: raise ResourceNotFound("No related service in Pnf instance details response") - pnf: Pnf = None + pnf: Optional[Pnf] = None for service_pnf in service.pnfs: if service_pnf.name == request_response.get("request", {})\ .get("requestDetails", {}).get("modelInfo", {}).get("modelCustomizationName"): @@ -969,7 +969,7 @@ class ServiceInstantiation(Instantiation): # pylint: disable=too-many-ancestors # pylint: disable=too-many-arguments, too-many-locals @classmethod - def instantiate_macro(cls, + def instantiate_macro(cls, # NOSONAR sdc_service: "SdcService", customer: "Customer", owning_entity: OwningEntity, diff --git a/src/onapsdk/so/so_db_adapter.py b/src/onapsdk/so/so_db_adapter.py index 31a93fa..57c5f33 100644 --- a/src/onapsdk/so/so_db_adapter.py +++ b/src/onapsdk/so/so_db_adapter.py @@ -28,7 +28,7 @@ class IdentityService: # pylint: disable=too-many-instance-attributes """Class to store identity service details.""" identity_id: str - url: str = "http://1.2.3.4:5000/v2.0" + url: str = "http://1.2.3.4:5000/v2.0" # NOSONAR mso_id: str = "onapsdk_user" mso_pass: str = "mso_pass_onapsdk" project_domain_name: str = "NULL" diff --git a/src/onapsdk/so/so_element.py b/src/onapsdk/so/so_element.py index 5c00654..e975cd9 100644 --- a/src/onapsdk/so/so_element.py +++ b/src/onapsdk/so/so_element.py @@ -16,7 +16,7 @@ import json from abc import ABC from dataclasses import dataclass from enum import Enum -from typing import Dict +from typing import Dict, Optional from onapsdk.configuration import settings from onapsdk.sdc.service import Service @@ -32,11 +32,11 @@ from onapsdk.utils.gui import GuiItem, GuiList class SoElement(OnapService): """Mother Class of all SO elements.""" - name: str = None + name: Optional[str] = None _server: str = "SO" base_url = settings.SO_URL api_version = settings.SO_API_VERSION - _status: str = None + _status: Optional[str] = None @property def headers(self): @@ -111,7 +111,7 @@ class SoElement(OnapService): f"{cls.api_version}/serviceInstances") @classmethod - def get_guis(cls) -> GuiItem: + def get_guis(cls) -> GuiList: """Retrieve the status of the SO GUIs. Only one GUI is referenced for SO: SO monitor diff --git a/src/onapsdk/utils/gui.py b/src/onapsdk/utils/gui.py index 421e966..b5a71cb 100644 --- a/src/onapsdk/utils/gui.py +++ b/src/onapsdk/utils/gui.py @@ -26,10 +26,10 @@ class GuiItem: class GuiList: """Class to list all the GUIs.""" - guilist: List[GuiItem] + guis: List[GuiItem] def add(self, element): """Add a GUi to GUI list.""" if not isinstance(element, GuiItem): raise AttributeError - self.guilist.append(element) + self.guis.append(element) diff --git a/src/onapsdk/utils/headers_creator.py b/src/onapsdk/utils/headers_creator.py index 75fc457..7df3ffb 100644 --- a/src/onapsdk/utils/headers_creator.py +++ b/src/onapsdk/utils/headers_creator.py @@ -223,7 +223,7 @@ def headers_sdc_artifact_upload(base_header: Dict[str, str], data: str): headers["Accept"] = "application/json, text/plain, */*" headers["Accept-Encoding"] = "gzip, deflate, br" headers["Content-Type"] = "application/json; charset=UTF-8" - md5_content = hashlib.new('md5', data.encode('UTF-8'), # nosec + md5_content = hashlib.new('md5', data.encode('UTF-8'), # nosec # NOSONAR usedforsecurity=False).hexdigest() # nosec content = base64.b64encode(md5_content.encode('ascii')).decode('UTF-8') headers["Content-MD5"] = content diff --git a/src/onapsdk/vid/vid.py b/src/onapsdk/vid/vid.py index 31cb92e..8abef6b 100644 --- a/src/onapsdk/vid/vid.py +++ b/src/onapsdk/vid/vid.py @@ -20,6 +20,10 @@ from onapsdk.onap_service import OnapService from onapsdk.utils.jinja import jinja_env +WARN_MSG = ("VID is deprecated and shouldn't be used! " + "It's not a part of the ONAP release since Istanbul.") + + class Vid(OnapService, ABC): """VID base class.""" @@ -32,8 +36,7 @@ class Vid(OnapService, ABC): Args: name (str): Resource name """ - warn("VID is deprecated and shouldn't be used! " - "It's not a part of the ONAP release since Istanbul.") + warn(WARN_MSG) super().__init__() self.name: str = name @@ -57,8 +60,7 @@ class Vid(OnapService, ABC): Vid: Created VID resource """ - warn("VID is deprecated and shouldn't be used! " - "It's not a part of the ONAP release since Istanbul.") + warn(WARN_MSG) cls.send_message( "POST", f"Declare VID resource with {name} name", @@ -81,8 +83,7 @@ class OwningEntity(Vid): str: Url used for ownint entity creation """ - warn("VID is deprecated and shouldn't be used! " - "It's not a part of the ONAP release since Istanbul.") + warn(WARN_MSG) return f"{cls.base_url}{cls.api_version}/maintenance/category_parameter/owningEntity" @@ -97,8 +98,7 @@ class Project(Vid): str: Url used for project creation """ - warn("VID is deprecated and shouldn't be used! " - "It's not a part of the ONAP release since Istanbul.") + warn(WARN_MSG) return f"{cls.base_url}{cls.api_version}/maintenance/category_parameter/project" @@ -113,8 +113,7 @@ class LineOfBusiness(Vid): str: Url used for line of business creation """ - warn("VID is deprecated and shouldn't be used! " - "It's not a part of the ONAP release since Istanbul.") + warn(WARN_MSG) return f"{cls.base_url}{cls.api_version}/maintenance/category_parameter/lineOfBusiness" @@ -129,6 +128,5 @@ class Platform(Vid): str: Url used for platform creation """ - warn("VID is deprecated and shouldn't be used! " - "It's not a part of the ONAP release since Istanbul.") + warn(WARN_MSG) return f"{cls.base_url}{cls.api_version}/maintenance/category_parameter/platform" |