aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
authorMichal Jagiello <michal.jagiello@t-mobile.pl>2023-10-27 13:26:12 +0200
committerMichal Jagiello <michal.jagiello@t-mobile.pl>2023-10-27 13:26:12 +0200
commit4bdb9d7a9e6b0087332a61125d544a95e695cb8b (patch)
tree6e511ab58181713c3aa10d465110def7197e9a54 /src
parentf2b1b5ed7fb348eb7c7f065840afaffc8cbec557 (diff)
Use latest pylint and make fixes for it
Do not be sticked into one version of pylint, but use the latest Issue-ID: INT-404 Signed-off-by: Michal Jagiello <michal.jagiello@t-mobile.pl> Change-Id: Ib2c0a2ebe680968ae9d69958a564733a76990367
Diffstat (limited to 'src')
-rw-r--r--src/onapsdk/aai/business/customer.py10
-rw-r--r--src/onapsdk/aai/business/line_of_business.py2
-rw-r--r--src/onapsdk/aai/business/owning_entity.py2
-rw-r--r--src/onapsdk/aai/business/platform.py2
-rw-r--r--src/onapsdk/aai/business/service.py3
-rw-r--r--src/onapsdk/aai/business/vnf.py3
-rw-r--r--src/onapsdk/aai/cloud_infrastructure/cloud_region.py12
-rw-r--r--src/onapsdk/aai/cloud_infrastructure/complex.py8
-rw-r--r--src/onapsdk/aai/cloud_infrastructure/geo_region.py2
-rw-r--r--src/onapsdk/aai/network/site_resource.py2
-rw-r--r--src/onapsdk/aai/service_design_and_creation.py2
-rw-r--r--src/onapsdk/cds/blueprint.py4
-rw-r--r--src/onapsdk/cds/blueprint_model.py9
-rw-r--r--src/onapsdk/cds/data_dictionary.py8
-rw-r--r--src/onapsdk/clamp/loop_instance.py2
-rw-r--r--src/onapsdk/configuration/global_settings.py2
-rw-r--r--src/onapsdk/configuration/loader.py4
-rw-r--r--src/onapsdk/k8s/definition.py4
-rw-r--r--src/onapsdk/k8s/instance.py2
-rw-r--r--src/onapsdk/sdc/__init__.py23
-rw-r--r--src/onapsdk/sdc/properties.py4
-rw-r--r--src/onapsdk/sdc/sdc_element.py15
-rw-r--r--src/onapsdk/sdc/sdc_resource.py40
-rw-r--r--src/onapsdk/sdc/service.py38
-rw-r--r--src/onapsdk/sdc/vsp.py17
-rw-r--r--src/onapsdk/sdnc/topology.py9
-rw-r--r--src/onapsdk/so/so_element.py5
-rw-r--r--src/onapsdk/utils/__init__.py2
-rw-r--r--src/onapsdk/utils/tosca_file_handler.py2
29 files changed, 110 insertions, 128 deletions
diff --git a/src/onapsdk/aai/business/customer.py b/src/onapsdk/aai/business/customer.py
index 6799fac..1305a12 100644
--- a/src/onapsdk/aai/business/customer.py
+++ b/src/onapsdk/aai/business/customer.py
@@ -207,9 +207,9 @@ class ServiceSubscription(AaiResource):
"""
try:
return next(self.cloud_regions)
- except StopIteration:
+ except StopIteration as exc:
msg = f"No cloud region for service subscription '{self.name}'"
- raise ParameterError(msg)
+ raise ParameterError(msg) from exc
@property
def tenant(self) -> "Tenant":
@@ -226,9 +226,9 @@ class ServiceSubscription(AaiResource):
"""
try:
return next(self.tenants)
- except StopIteration:
+ except StopIteration as exc:
msg = f"No tenants for service subscription '{self.name}'"
- raise ParameterError(msg)
+ raise ParameterError(msg) from exc
@property
def _cloud_regions_tenants_data(self) -> Iterator["ServiceSubscriptionCloudRegionTenantData"]:
@@ -441,7 +441,7 @@ class Customer(AaiResource):
"subscriber-type": subscriber_type,
}
)
- url: str = (f"{cls.get_all_url()}?{urlencode(filter_parameters)}")
+ 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"],
diff --git a/src/onapsdk/aai/business/line_of_business.py b/src/onapsdk/aai/business/line_of_business.py
index a1b807d..0eb886f 100644
--- a/src/onapsdk/aai/business/line_of_business.py
+++ b/src/onapsdk/aai/business/line_of_business.py
@@ -131,6 +131,6 @@ class LineOfBusiness(AaiResource, AaiResourceLinkToTenantMixin):
"""
self.send_message(
"DELETE",
- f"Delete line of business",
+ "Delete line of business",
f"{self.url}?resource-version={self.resource_version}"
)
diff --git a/src/onapsdk/aai/business/owning_entity.py b/src/onapsdk/aai/business/owning_entity.py
index 1ef0a39..b01ccc4 100644
--- a/src/onapsdk/aai/business/owning_entity.py
+++ b/src/onapsdk/aai/business/owning_entity.py
@@ -187,6 +187,6 @@ class OwningEntity(AaiResource, AaiResourceLinkToTenantMixin):
"""
self.send_message(
"DELETE",
- f"Delete owning entity",
+ "Delete owning entity",
f"{self.url}?resource-version={self.resource_version}"
)
diff --git a/src/onapsdk/aai/business/platform.py b/src/onapsdk/aai/business/platform.py
index f652e32..7a6be76 100644
--- a/src/onapsdk/aai/business/platform.py
+++ b/src/onapsdk/aai/business/platform.py
@@ -126,6 +126,6 @@ class Platform(AaiResource):
"""Delete platform."""
self.send_message(
"DELETE",
- f"Delete platform",
+ "Delete platform",
f"{self.url}?resource-version={self.resource_version}"
)
diff --git a/src/onapsdk/aai/business/service.py b/src/onapsdk/aai/business/service.py
index 4351a26..12b5b4f 100644
--- a/src/onapsdk/aai/business/service.py
+++ b/src/onapsdk/aai/business/service.py
@@ -497,8 +497,7 @@ class ServiceInstance(Instance): # pylint: disable=too-many-instance-attributes
"""
if not self.active:
- msg = f'Service orchestration status must be "Active"'
- raise StatusError(msg)
+ raise StatusError('Service orchestration status must be "Active"')
return NetworkInstantiation.instantiate_ala_carte(
self,
diff --git a/src/onapsdk/aai/business/vnf.py b/src/onapsdk/aai/business/vnf.py
index 39b1be0..31762c9 100644
--- a/src/onapsdk/aai/business/vnf.py
+++ b/src/onapsdk/aai/business/vnf.py
@@ -453,8 +453,7 @@ class VnfInstance(Instance): # pylint: disable=too-many-instance-attributes
"""
if not self.service_instance.active:
- msg = f'Service orchestration status must be "Active"'
- raise StatusError(msg)
+ raise StatusError('Service orchestration status must be "Active"')
lob = settings.LOB
platform = settings.PLATFORM
diff --git a/src/onapsdk/aai/cloud_infrastructure/cloud_region.py b/src/onapsdk/aai/cloud_infrastructure/cloud_region.py
index 5f01548..240b7a2 100644
--- a/src/onapsdk/aai/cloud_infrastructure/cloud_region.py
+++ b/src/onapsdk/aai/cloud_infrastructure/cloud_region.py
@@ -73,7 +73,7 @@ class CloudRegion(AaiResource, AaiResourceLinkToComplexMixin, AaiResourceLinkToP
Represents A&AI cloud region object.
"""
- def __init__(self,
+ def __init__(self, # pylint: disable=too-many-arguments
cloud_owner: str,
cloud_region_id: str,
orchestration_disabled: bool,
@@ -177,7 +177,7 @@ class CloudRegion(AaiResource, AaiResourceLinkToComplexMixin, AaiResourceLinkToP
"owner-defined-type": owner_defined_type,
}
)
- url: str = (f"{cls.get_all_url()}?{urlencode(filter_parameters)}")
+ url: str = f"{cls.get_all_url()}?{urlencode(filter_parameters)}"
response_json: Dict[str, List[Dict[str, Any]]] = cls.send_message_json(
"GET", "get cloud regions", url
)
@@ -215,15 +215,15 @@ class CloudRegion(AaiResource, AaiResourceLinkToComplexMixin, AaiResourceLinkToP
"""
try:
return next(cls.get_all(cloud_owner=cloud_owner, cloud_region_id=cloud_region_id))
- except StopIteration:
+ except StopIteration as exc:
msg = (
f'CloudRegion with {cloud_owner}, '
f'{cloud_region_id} cloud-id not found. '
)
- raise ResourceNotFound(msg)
+ raise ResourceNotFound(msg) from exc
@classmethod
- def create(cls, # pylint: disable=too-many-locals
+ def create(cls, # pylint: disable=too-many-locals, too-many-arguments
cloud_owner: str,
cloud_region_id: str,
orchestration_disabled: bool,
@@ -276,7 +276,7 @@ class CloudRegion(AaiResource, AaiResourceLinkToComplexMixin, AaiResourceLinkToP
return cloud_region
@classmethod
- def update(cls, # pylint: disable=too-many-locals
+ def update(cls, # pylint: disable=too-many-locals, too-many-arguments
cloud_owner: str,
cloud_region_id: str,
orchestration_disabled: bool,
diff --git a/src/onapsdk/aai/cloud_infrastructure/complex.py b/src/onapsdk/aai/cloud_infrastructure/complex.py
index 1a1684a..6b2b74f 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
+ def __init__(self, # 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
+ def create(cls, # 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
+ def update(cls, # pylint: disable=too-many-locals, too-many-arguments
physical_location_id: str,
*,
name: str = "",
@@ -294,7 +294,7 @@ class Complex(AaiResource, AaiResourceLinkToGeoRegionMixin): # pylint: disable=
"identity-url": identity_url,
}
)
- url: str = (f"{cls.get_all_url()}?{urlencode(filter_parameters)}")
+ url: str = f"{cls.get_all_url()}?{urlencode(filter_parameters)}"
for complex_json in cls.send_message_json("GET",
"get cloud regions",
url).get("complex", []):
diff --git a/src/onapsdk/aai/cloud_infrastructure/geo_region.py b/src/onapsdk/aai/cloud_infrastructure/geo_region.py
index 32ff820..7d17130 100644
--- a/src/onapsdk/aai/cloud_infrastructure/geo_region.py
+++ b/src/onapsdk/aai/cloud_infrastructure/geo_region.py
@@ -23,7 +23,7 @@ from ..aai_element import AaiResource
class GeoRegion(AaiResource): # pylint: disable=too-many-instance-attributes
"""Geo region class."""
- def __init__(self,
+ def __init__(self, # pylint: disable=too-many-arguments
geo_region_id: str,
*,
geo_region_name: str = "",
diff --git a/src/onapsdk/aai/network/site_resource.py b/src/onapsdk/aai/network/site_resource.py
index e6916f7..6804e65 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
+ def __init__(self, # pylint: disable=too-many-locals, too-many-arguments
site_resource_id: str,
*,
site_resource_name: str = "",
diff --git a/src/onapsdk/aai/service_design_and_creation.py b/src/onapsdk/aai/service_design_and_creation.py
index 5bf2c6f..f4c0f53 100644
--- a/src/onapsdk/aai/service_design_and_creation.py
+++ b/src/onapsdk/aai/service_design_and_creation.py
@@ -86,7 +86,7 @@ class Service(AaiResource):
filter_parameters: dict = cls.filter_none_key_values(
{"service-id": service_id, "service-description": service_description}
)
- url: str = (f"{cls.get_all_url()}?{urlencode(filter_parameters)}")
+ url: str = f"{cls.get_all_url()}?{urlencode(filter_parameters)}"
for service in cls.send_message_json("GET", "get subscriptions", url).get("service", []):
yield Service(
service_id=service["service-id"],
diff --git a/src/onapsdk/cds/blueprint.py b/src/onapsdk/cds/blueprint.py
index 1286375..2d6aa37 100644
--- a/src/onapsdk/cds/blueprint.py
+++ b/src/onapsdk/cds/blueprint.py
@@ -772,8 +772,8 @@ class Blueprint(CdsElement):
"""
try:
return next(filter(lambda workflow: workflow.name == workflow_name, self.workflows))
- except StopIteration:
- raise ParameterError("Workflow with given name does not exist")
+ except StopIteration as exc:
+ raise ParameterError("Workflow with given name does not exist") from exc
def get_resolved_template(self, # pylint: disable=too-many-arguments
artifact_name: str,
diff --git a/src/onapsdk/cds/blueprint_model.py b/src/onapsdk/cds/blueprint_model.py
index 7976001..9c4c8a5 100644
--- a/src/onapsdk/cds/blueprint_model.py
+++ b/src/onapsdk/cds/blueprint_model.py
@@ -110,9 +110,9 @@ class BlueprintModel(CdsElement): # pylint: disable=too-many-instance-attribute
tags=blueprint_model["blueprintModel"]['tags']
)
- except ResourceNotFound:
+ except ResourceNotFound as exc:
raise ResourceNotFound(f"BlueprintModel blueprint_model_id='{blueprint_model_id}"
- f" not found")
+ f" not found") from exc
@classmethod
def get_by_name_and_version(cls, blueprint_name: str,
@@ -151,9 +151,10 @@ class BlueprintModel(CdsElement): # pylint: disable=too-many-instance-attribute
tags=blueprint_model["blueprintModel"]['tags']
)
- except ResourceNotFound:
+ except ResourceNotFound as exc:
raise ResourceNotFound(f"BlueprintModel blueprint_name='{blueprint_name}"
- f" and blueprint_version='{blueprint_version}' not found")
+ f" and blueprint_version='{blueprint_version}' "
+ "not found") from exc
@classmethod
def get_all(cls) -> Iterator["BlueprintModel"]:
diff --git a/src/onapsdk/cds/data_dictionary.py b/src/onapsdk/cds/data_dictionary.py
index 27cfc29..380cec8 100644
--- a/src/onapsdk/cds/data_dictionary.py
+++ b/src/onapsdk/cds/data_dictionary.py
@@ -180,8 +180,8 @@ class DataDictionary(CdsElement):
"updatedBy": self.data_dictionary_json["updated-by"],
"definition": self.data_dictionary_json
}
- except KeyError:
- raise ValidationError("Raw data dictionary JSON has invalid schema")
+ except KeyError as exc:
+ raise ValidationError("Raw data dictionary JSON has invalid schema") from exc
class DataDictionarySet:
@@ -236,7 +236,7 @@ class DataDictionarySet:
Args:
dd_file_path (str): Data dictinary file path.
"""
- with open(dd_file_path, "w") as dd_file:
+ with open(dd_file_path, "w", encoding="utf-8") as dd_file:
dd_file.write(json.dumps([dd.data_dictionary_json for dd in self.dd_set], indent=4))
@classmethod
@@ -259,7 +259,7 @@ class DataDictionarySet:
dd_set: DataDictionarySet = DataDictionarySet()
try:
- with open(dd_file_path, "r") as dd_file: # type file
+ with open(dd_file_path, "r", encoding="utf-8") as dd_file: # type file
dd_json: dict = json.loads(dd_file.read())
for data_dictionary in dd_json: # type DataDictionary
dd_set.add(DataDictionary(data_dictionary, fix_schema=fix_schema))
diff --git a/src/onapsdk/clamp/loop_instance.py b/src/onapsdk/clamp/loop_instance.py
index a72f9d1..a010796 100644
--- a/src/onapsdk/clamp/loop_instance.py
+++ b/src/onapsdk/clamp/loop_instance.py
@@ -343,7 +343,7 @@ class LoopInstance(Clamp):
def delete(self) -> None:
"""Delete the loop instance."""
self._logger.debug("Delete %s loop instance", self.name)
- url = "{}/loop/delete/{}".format(self.base_url(), self.name)
+ url = f"{self.base_url()}/loop/delete/{self.name}"
self.send_message('PUT',
'Delete loop instance',
url)
diff --git a/src/onapsdk/configuration/global_settings.py b/src/onapsdk/configuration/global_settings.py
index 063e462..59a9242 100644
--- a/src/onapsdk/configuration/global_settings.py
+++ b/src/onapsdk/configuration/global_settings.py
@@ -1,4 +1,4 @@
-"""Global settings module.""" # pylint: disable=bad-whitespace
+"""Global settings module."""
# Copyright 2022 Orange, Deutsche Telekom AG
#
# Licensed under the Apache License, Version 2.0 (the "License");
diff --git a/src/onapsdk/configuration/loader.py b/src/onapsdk/configuration/loader.py
index a9aae6f..ae38588 100644
--- a/src/onapsdk/configuration/loader.py
+++ b/src/onapsdk/configuration/loader.py
@@ -54,9 +54,9 @@ class SettingsLoader:
# Load values from custom settings
try:
module = importlib.import_module(settings_env_value)
- except ModuleNotFoundError:
+ except ModuleNotFoundError as exc:
msg = "Can't import custom settings. Is it under PYTHONPATH?"
- raise ModuleError(msg)
+ raise ModuleError(msg) from exc
self.filter_and_set(module)
def __getattribute__(self, name: str) -> Any:
diff --git a/src/onapsdk/k8s/definition.py b/src/onapsdk/k8s/definition.py
index 4d86afc..305bca2 100644
--- a/src/onapsdk/k8s/definition.py
+++ b/src/onapsdk/k8s/definition.py
@@ -94,10 +94,10 @@ class Profile(DefinitionBase):
self.kubernetes_version: str = kubernetes_version
self.labels: dict = labels
if self.labels is None:
- self.labels = dict()
+ self.labels = {}
self.extra_resource_types: dict = extra_resource_types
if self.extra_resource_types is None:
- self.extra_resource_types = dict()
+ self.extra_resource_types = {}
@property
def url(self) -> str:
diff --git a/src/onapsdk/k8s/instance.py b/src/onapsdk/k8s/instance.py
index 4b93258..64726b9 100644
--- a/src/onapsdk/k8s/instance.py
+++ b/src/onapsdk/k8s/instance.py
@@ -139,7 +139,7 @@ class Configuration(InstanceBase):
"""
url: str = f"{self.url}/rollback"
- params = dict()
+ params = {}
if config_version is not None:
params["config-version"] = config_version
if config_tag is not None:
diff --git a/src/onapsdk/sdc/__init__.py b/src/onapsdk/sdc/__init__.py
index c96c77f..f359344 100644
--- a/src/onapsdk/sdc/__init__.py
+++ b/src/onapsdk/sdc/__init__.py
@@ -171,7 +171,7 @@ class SDC(OnapService, ABC):
try:
result = \
- cls.send_message_json('GET', "get {}s".format(cls.__name__),
+ cls.send_message_json('GET', f"get {cls.__name__}s",
url, **kwargs)
for obj_info in cls._get_objects_list(result):
@@ -322,13 +322,12 @@ class SdcOnboardable(SDC, ABC):
self._logger.info("attempting to create %s %s in SDC",
type(self).__name__, self.name)
if not self.exists():
- url = "{}/{}".format(self._base_create_url(), self._sdc_path())
+ url = f"{self._base_create_url()}/{self._sdc_path()}"
template = jinja_env().get_template(template_name)
data = template.render(**kwargs)
try:
create_result = self.send_message_json('POST',
- "create {}".format(
- type(self).__name__),
+ f"create {type(self).__name__}",
url,
data=data)
except RequestError as exc:
@@ -336,13 +335,12 @@ class SdcOnboardable(SDC, ABC):
"an error occured during creation of %s %s in SDC",
type(self).__name__, self.name)
raise exc
- else:
- self._logger.info("%s %s is created in SDC",
- type(self).__name__, self.name)
- self._status = const.DRAFT
- self.identifier = self._get_identifier_from_sdc(create_result)
- self._version = self._get_version_from_sdc(create_result)
- self.update_informations_from_sdc_creation(create_result)
+ self._logger.info("%s %s is created in SDC",
+ type(self).__name__, self.name)
+ self._status = const.DRAFT
+ self.identifier = self._get_identifier_from_sdc(create_result)
+ self._version = self._get_version_from_sdc(create_result)
+ self.update_informations_from_sdc_creation(create_result)
else:
self._logger.warning("%s %s is already created in SDC",
@@ -371,8 +369,7 @@ class SdcOnboardable(SDC, ABC):
data = template.render(action=action, const=const)
return self.send_message(self.ACTION_METHOD,
- "{} {}".format(action,
- type(self).__name__),
+ f"{action} {type(self).__name__}",
url,
data=data,
**kwargs)
diff --git a/src/onapsdk/sdc/properties.py b/src/onapsdk/sdc/properties.py
index f7e07a0..519772c 100644
--- a/src/onapsdk/sdc/properties.py
+++ b/src/onapsdk/sdc/properties.py
@@ -141,8 +141,8 @@ class Property: # pylint: disable=too-many-instance-attributes, too-few-public-
try:
return next(filter(lambda x: x.unique_id == self.get_input_values[0].get("inputId"),
self.sdc_resource.inputs))
- except StopIteration:
- raise ParameterError("Property input does not exist")
+ except StopIteration as exc:
+ raise ParameterError("Property input does not exist") from exc
@property
def value(self) -> Any:
diff --git a/src/onapsdk/sdc/sdc_element.py b/src/onapsdk/sdc/sdc_element.py
index 8fd8da1..70abadd 100644
--- a/src/onapsdk/sdc/sdc_element.py
+++ b/src/onapsdk/sdc/sdc_element.py
@@ -40,8 +40,7 @@ class SdcElement(SdcOnboardable, ABC):
"""
if self.created():
- url = "{}/items/{}/versions".format(self._base_url(),
- self.identifier)
+ url = f"{self._base_url()}/items/{self.identifier}/versions"
results: Dict[str, Any] = self.send_message_json('GET', 'get item', url)
if results["listCount"] > 1:
items: List[Dict[str, Any]] = results["results"]
@@ -89,7 +88,7 @@ class SdcElement(SdcOnboardable, ABC):
str: the base url
"""
- return "{}/sdc1/feProxy/onboarding-api/v1.0".format(cls.base_front_url)
+ return f"{cls.base_front_url}/sdc1/feProxy/onboarding-api/v1.0"
@classmethod
def _base_create_url(cls) -> str:
@@ -100,7 +99,7 @@ class SdcElement(SdcOnboardable, ABC):
str: the base url
"""
- return "{}/sdc1/feProxy/onboarding-api/v1.0".format(cls.base_front_url)
+ return f"{cls.base_front_url}/sdc1/feProxy/onboarding-api/v1.0"
def _generate_action_subpath(self, action: str) -> str:
"""
@@ -127,7 +126,7 @@ class SdcElement(SdcOnboardable, ABC):
str: the end of the path
"""
- return "{}/versions/{}".format(self.identifier, self.version)
+ return f"{self.identifier}/versions/{self.version}"
def _action_url(self, base: str, subpath: str, version_path: str,
action_type: str = None) -> str:
@@ -145,8 +144,8 @@ class SdcElement(SdcOnboardable, ABC):
"""
if action_type == const.ARCHIVE:
- version_path = version_path.split("/")[0]
- return "{}/{}/{}/actions".format(base, subpath, version_path)
+ version_path = version_path.split("/", maxsplit=1)[0]
+ return f"{base}/{subpath}/{version_path}/actions"
@classmethod
def _get_objects_list(cls, result: List[Dict[str, Any]]
@@ -172,7 +171,7 @@ class SdcElement(SdcOnboardable, ABC):
str: the url
"""
- return "{}/{}".format(cls._base_url(), cls._sdc_path())
+ return f"{cls._base_url()}/{cls._sdc_path()}"
@property
def delete_url(self) -> str:
diff --git a/src/onapsdk/sdc/sdc_resource.py b/src/onapsdk/sdc/sdc_resource.py
index a92070c..ea25b31 100644
--- a/src/onapsdk/sdc/sdc_resource.py
+++ b/src/onapsdk/sdc/sdc_resource.py
@@ -117,8 +117,7 @@ class SdcResource(SdcOnboardable, ABC): # pylint: disable=too-many-instance-att
headers = headers_sdc_tester(SdcResource.headers)
response = self.send_message_json("GET",
- "Deep Load {}".format(
- type(self).__name__),
+ f"Deep Load {type(self).__name__}",
url,
headers=headers)
@@ -187,8 +186,7 @@ class SdcResource(SdcOnboardable, ABC): # pylint: disable=too-many-instance-att
"""
if not action_type:
action_type = "lifecycleState"
- return "{}/{}/{}/{}/{}".format(base, self._resource_type, version_path,
- action_type, subpath)
+ return f"{base}/{self._resource_type}/{version_path}/{action_type}/{subpath}"
@classmethod
def _base_create_url(cls) -> str:
@@ -199,7 +197,7 @@ class SdcResource(SdcOnboardable, ABC): # pylint: disable=too-many-instance-att
str: the base url
"""
- return "{}/sdc1/feProxy/rest/v1/catalog".format(cls.base_front_url)
+ return f"{cls.base_front_url}/sdc1/feProxy/rest/v1/catalog"
@classmethod
def _base_url(cls) -> str:
@@ -210,7 +208,7 @@ class SdcResource(SdcOnboardable, ABC): # pylint: disable=too-many-instance-att
str: the base url
"""
- return "{}/sdc/v1/catalog".format(cls.base_back_url)
+ return f"{cls.base_back_url}/sdc/v1/catalog"
@classmethod
def _get_all_url(cls) -> str:
@@ -221,8 +219,7 @@ class SdcResource(SdcOnboardable, ABC): # pylint: disable=too-many-instance-att
str: the url
"""
- return "{}/{}?resourceType={}".format(cls._base_url(), cls._sdc_path(),
- cls.__name__.upper())
+ return f"{cls._base_url()}/{cls._sdc_path()}?resourceType={cls.__name__.upper()}"
@classmethod
def _get_objects_list(cls, result: List[Dict[str, Any]]
@@ -371,12 +368,20 @@ class SdcResource(SdcOnboardable, ABC): # pylint: disable=too-many-instance-att
"""Really submit the SDC Vf in order to enable it."""
raise NotImplementedError("SDC is an abstract class")
+ def _create_onboard_step(self) -> None:
+ self.create()
+ time.sleep(self._time_wait)
+ self.onboard()
+
+ def _certify_onboard_step(self) -> None:
+ self.certify()
+ time.sleep(self._time_wait)
+ self.onboard()
+
def onboard(self) -> None:
"""Onboard resource in SDC."""
if not self.status:
- self.create()
- time.sleep(self._time_wait)
- self.onboard()
+ self._create_onboard_step()
elif self.status == const.DRAFT:
for property_to_add in self._properties_to_add:
self.add_property(property_to_add)
@@ -386,10 +391,7 @@ class SdcResource(SdcOnboardable, ABC): # pylint: disable=too-many-instance-att
time.sleep(self._time_wait)
self.onboard()
elif self.status == const.CHECKED_IN:
- # Checked in status check added
- self.certify()
- time.sleep(self._time_wait)
- self.onboard()
+ self._certify_onboard_step()
elif self.status == const.CERTIFIED:
self.load()
@@ -594,7 +596,8 @@ class SdcResource(SdcOnboardable, ABC): # pylint: disable=too-many-instance-att
StatusError: Resource has not DRAFT status
"""
- data = open(artifact, 'rb').read()
+ with open(artifact, 'rb') as artifact_file:
+ data = artifact_file.read()
artifact_string = base64.b64encode(data).decode('utf-8')
if self.status != const.DRAFT:
msg = "Can't add artifact to resource which is not in DRAFT status"
@@ -931,9 +934,8 @@ class SdcResource(SdcOnboardable, ABC): # pylint: disable=too-many-instance-att
"""
if self.status == const.DRAFT:
- url = "{}/{}/{}/resourceInstance".format(self._base_create_url(),
- self._sdc_path(),
- self.unique_identifier)
+ url = (f"{self._base_create_url()}/{self._sdc_path()}/"
+ f"{self.unique_identifier}/resourceInstance")
template = jinja_env().get_template(
"add_resource_to_service.json.j2")
diff --git a/src/onapsdk/sdc/service.py b/src/onapsdk/sdc/service.py
index 507b068..7f195a5 100644
--- a/src/onapsdk/sdc/service.py
+++ b/src/onapsdk/sdc/service.py
@@ -229,10 +229,7 @@ class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too
# first Lines are equivalent for all onboard functions but it's more
# readable
if not self.status:
- # equivalent step as in onboard-function in sdc_resource
- self.create()
- time.sleep(self._time_wait)
- self.onboard()
+ self._create_onboard_step()
elif self.status == const.DRAFT:
if not any([self.resources, self._properties_to_add]):
raise ParameterError("No resources nor properties were given")
@@ -241,10 +238,8 @@ class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too
time.sleep(self._time_wait)
self.onboard()
elif self.status == const.CHECKED_IN:
- self.certify()
- time.sleep(self._time_wait)
- self.onboard()
- elif self.status == const.CERTIFIED:
+ self._certify_onboard_step()
+ elif self.status == const.CERTIFIED: # pylint: disable=duplicate-code
self.distribute()
self.onboard()
elif self.status == const.DISTRIBUTED:
@@ -298,13 +293,12 @@ class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too
"""
if not self._tosca_model:
- url = "{}/services/{}/toscaModel".format(self._base_url(),
- self.identifier)
+ url = f"{self._base_url()}/services/{self.identifier}/toscaModel"
headers = self.headers.copy()
headers["Accept"] = "application/octet-stream"
self._tosca_model = self.send_message(
"GET",
- "Download Tosca Model for {}".format(self.name),
+ f"Download Tosca Model for {self.name}",
url,
headers=headers).content
return self._tosca_model
@@ -612,13 +606,11 @@ class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too
def get_tosca(self, paths) -> None:
"""Get Service tosca files and save it."""
- url = "{}/services/{}/toscaModel".format(self._base_url(),
- self.identifier)
+ url = f"{self._base_url()}/services/{self.identifier}/toscaModel"
headers = self.headers.copy()
headers["Accept"] = "application/octet-stream"
result = self.send_message("GET",
- "Download Tosca Model for {}".format(
- self.name),
+ f"Download Tosca Model for {self.name}",
url,
headers=headers)
if result:
@@ -626,7 +618,7 @@ class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too
def _create_tosca_file(self, paths, result: Response) -> None:
"""Create Service Tosca files from HTTP response."""
- csar_filename = "service-{}-csar.csar".format(self.name)
+ csar_filename = f"service-{self.name}-csar.csar"
makedirs(paths, exist_ok=True)
with open((paths + csar_filename), 'wb') as csar_file:
for chunk in result.iter_content(chunk_size=128):
@@ -639,8 +631,7 @@ class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too
def _check_distributed(self) -> bool:
"""Check if service is distributed and update status accordingly."""
- url = "{}/services/distribution/{}".format(self._base_create_url(),
- self.distribution_id)
+ url = f"{self._base_create_url()}/services/distribution/{self.distribution_id}"
headers = headers_sdc_creator(SdcResource.headers)
status = {}
@@ -649,8 +640,7 @@ class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too
try:
result = self.send_message_json("GET",
- "Check distribution for {}".format(
- self.name),
+ f"Check distribution for {self.name}",
url,
headers=headers)
except ResourceNotFound:
@@ -688,12 +678,10 @@ class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too
def load_metadata(self) -> None:
"""Load Metada of Service and retrieve informations."""
- url = "{}/services/{}/distribution".format(self._base_create_url(),
- self.identifier)
+ url = f"{self._base_create_url()}/services/{self.identifier}/distribution"
headers = headers_sdc_creator(SdcResource.headers)
result = self.send_message_json("GET",
- "Get Metadata for {}".format(
- self.name),
+ f"Get Metadata for {self.name}",
url,
headers=headers)
if ('distributionStatusOfServiceList' in result
@@ -712,7 +700,7 @@ class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too
str: the url
"""
- return "{}/{}".format(cls._base_url(), cls._sdc_path())
+ return f"{cls._base_url()}/{cls._sdc_path()}"
def _really_submit(self) -> None:
"""Really submit the SDC Service in order to enable it."""
diff --git a/src/onapsdk/sdc/vsp.py b/src/onapsdk/sdc/vsp.py
index 3100647..07b8d1e 100644
--- a/src/onapsdk/sdc/vsp.py
+++ b/src/onapsdk/sdc/vsp.py
@@ -180,16 +180,14 @@ class Vsp(SdcElement): # pylint: disable=too-many-instance-attributes
def _get_item_version_details(self) -> Dict[Any, Any]:
"""Get vsp item details."""
if self.created() and self.version:
- url = "{}/items/{}/versions/{}".format(self._base_url(),
- self.identifier,
- self.version)
+ url = f"{self._base_url()}/items/{self.identifier}/versions/{self.version}"
return self.send_message_json('GET', 'get item version', url)
return {}
def _upload_action(self, package_to_upload: BinaryIO):
"""Do upload for real."""
- url = "{}/{}/{}/orchestration-template-candidate".format(
- self._base_url(), Vsp._sdc_path(), self._version_path())
+ url = (f"{self._base_url()}/{Vsp._sdc_path()}/{self._version_path()}/"
+ "orchestration-template-candidate")
headers = self.headers.copy()
headers.pop("Content-Type")
headers["Accept-Encoding"] = "gzip, deflate"
@@ -217,8 +215,8 @@ class Vsp(SdcElement): # pylint: disable=too-many-instance-attributes
def _validate_action(self):
"""Do validate for real."""
- url = "{}/{}/{}/orchestration-template-candidate/process".format(
- self._base_url(), Vsp._sdc_path(), self._version_path())
+ url = (f"{self._base_url()}/{Vsp._sdc_path()}/{self._version_path()}/"
+ "orchestration-template-candidate/process")
validate_result = self.send_message_json('PUT',
'Validate artifacts for Vsp',
url)
@@ -323,9 +321,8 @@ class Vsp(SdcElement): # pylint: disable=too-many-instance-attributes
def _get_vsp_details(self) -> Dict[Any, Any]:
"""Get vsp details."""
if self.created() and self.version:
- url = "{}/vendor-software-products/{}/versions/{}".format(
- self._base_url(), self.identifier, self.version)
-
+ url = (f"{self._base_url()}/vendor-software-products/{self.identifier}/"
+ f"versions/{self.version}")
return self.send_message_json('GET', 'get vsp version', url)
return {}
diff --git a/src/onapsdk/sdnc/topology.py b/src/onapsdk/sdnc/topology.py
index f618c2c..2b79ec9 100644
--- a/src/onapsdk/sdnc/topology.py
+++ b/src/onapsdk/sdnc/topology.py
@@ -12,7 +12,7 @@
# 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 List, Dict, Iterable
+from typing import List, Dict, Iterable, Optional
from onapsdk.utils.headers_creator import headers_sdnc_creator
from onapsdk.utils.jinja import jinja_env
@@ -219,7 +219,7 @@ class Topology(SdncElement):
except KeyError:
return Topology(topology_id=topology_id)
- def get_node(self, node_id) -> "Node":
+ def get_node(self, node_id) -> Optional["Node"]:
"""Get the node with a specific node_id form the specific topology at SDNC via NETCONF-API.
Returns:
@@ -240,7 +240,8 @@ class Topology(SdncElement):
password=node["netconf-node-topology:password"],
topology_id=self.topology_id)
except KeyError:
- print(f"Error. Node creation skipped.")
+ self._logger.error("Error. Node creation skipped.")
+ return None
def get_all_nodes(self) -> Iterable["Node"]:
"""Get all nodes of the specific topology from SDNC using NETCONF-API.
@@ -264,4 +265,4 @@ class Topology(SdncElement):
password=node["netconf-node-topology:password"],
topology_id=self.topology_id)
except KeyError:
- print(f"Error. Node creation skipped. KeyError")
+ self._logger.error("Error. Node creation skipped. KeyError")
diff --git a/src/onapsdk/so/so_element.py b/src/onapsdk/so/so_element.py
index 68866ae..5c00654 100644
--- a/src/onapsdk/so/so_element.py
+++ b/src/onapsdk/so/so_element.py
@@ -107,9 +107,8 @@ class SoElement(OnapService):
str: the base url
"""
- return "{}/onap/so/infra/serviceInstantiation/{}/serviceInstances".format(
- cls.base_url, cls.api_version
- )
+ return (f"{cls.base_url}/onap/so/infra/serviceInstantiation/"
+ f"{cls.api_version}/serviceInstances")
@classmethod
def get_guis(cls) -> GuiItem:
diff --git a/src/onapsdk/utils/__init__.py b/src/onapsdk/utils/__init__.py
index bd7f9f5..05b4c26 100644
--- a/src/onapsdk/utils/__init__.py
+++ b/src/onapsdk/utils/__init__.py
@@ -35,6 +35,6 @@ def load_json_file(path_to_json_file: str) -> str:
Returns:
File content as string (str)
"""
- with open(path_to_json_file) as json_file:
+ with open(path_to_json_file, encoding="utf-8") as json_file:
data = json.load(json_file)
return json.dumps(data)
diff --git a/src/onapsdk/utils/tosca_file_handler.py b/src/onapsdk/utils/tosca_file_handler.py
index 1d1918b..e1665f6 100644
--- a/src/onapsdk/utils/tosca_file_handler.py
+++ b/src/onapsdk/utils/tosca_file_handler.py
@@ -70,7 +70,7 @@ def get_vf_list_from_tosca_file(model: str) -> List:
if "org.openecomp.resource.vf" in value:
print(node, value)
if node not in newlist:
- search_value = str(node).split(" ")[0]
+ search_value = str(node).split(" ", maxsplit=1)[0]
newlist.append(search_value)
return newlist