diff options
author | Michal Jagiello <michal.jagiello@t-mobile.pl> | 2023-07-17 13:30:55 +0000 |
---|---|---|
committer | Michal Jagiello <michal.jagiello@t-mobile.pl> | 2023-07-17 13:50:59 +0000 |
commit | 939f1933c8635a18de428c696c97cc5d8600abff (patch) | |
tree | a58cb9342f1a9d2cec80fc6a792eae0daae7bc49 /src/onaptests/steps/instantiate | |
parent | b1e1a5fa99e29b3ffc2bc48d11b163e28cbb7744 (diff) |
Add pylama into checks
Checks code with pylama
Issue-ID: INT-2222
Signed-off-by: Michal Jagiello <michal.jagiello@t-mobile.pl>
Change-Id: If275ef90f21e3049abe72bcf373473b40846005c
Diffstat (limited to 'src/onaptests/steps/instantiate')
6 files changed, 67 insertions, 38 deletions
diff --git a/src/onaptests/steps/instantiate/k8s_profile_create.py b/src/onaptests/steps/instantiate/k8s_profile_create.py index f0ad61b..27831c5 100644 --- a/src/onaptests/steps/instantiate/k8s_profile_create.py +++ b/src/onaptests/steps/instantiate/k8s_profile_create.py @@ -12,6 +12,7 @@ import onaptests.utils.exceptions as onap_test_exceptions from ..base import BaseStep from .vnf_ala_carte import YamlTemplateVnfAlaCarteInstantiateStep + class K8SProfileStep(BaseStep): """CreateK8sProfileStep.""" @@ -116,8 +117,10 @@ class K8SProfileStep(BaseStep): self._logger.info("Create the k8s profile if it doesn't exist") super().execute() customer: Customer = Customer.get_by_global_customer_id(settings.GLOBAL_CUSTOMER_ID) - service_subscription: ServiceSubscription = customer.get_service_subscription_by_service_type(self.service_name) - self._service_instance: ServiceInstance = service_subscription.get_service_instance_by_name(self.service_instance_name) + service_subscription: ServiceSubscription = \ + customer.get_service_subscription_by_service_type(self.service_name) + self._service_instance: ServiceInstance = \ + service_subscription.get_service_instance_by_name(self.service_instance_name) for vnf_instance in self._service_instance.vnf_instances: # possible to have several modules for 1 VNF @@ -140,15 +143,15 @@ class K8SProfileStep(BaseStep): self._logger.error("Missing rb profile information") raise onap_test_exceptions.ProfileInformationException - ######## Check profile for Definition ################################### + # Check profile for Definition try: rbdef.get_profile_by_name(k8s_profile_name) except ResourceNotFound: - ######## Create profile for Definition ################################### + # Create profile for Definition profile = rbdef.create_profile(k8s_profile_name, k8s_profile_namespace, settings.K8S_PROFILE_K8S_VERSION) - ####### Upload artifact for created profile ############################## + # Upload artifact for created profile profile.upload_artifact(open(settings.K8S_PROFILE_ARTIFACT_PATH, 'rb').read()) @BaseStep.store_state(cleanup=True) @@ -172,7 +175,7 @@ class K8SProfileStep(BaseStep): if k8s_profile_name == "": self._logger.error("K8s profile deletion failed, missing rb profile name") raise onap_test_exceptions.ProfileInformationException - ######## Delete profile for Definition ################################### + # Delete profile for Definition try: profile = rbdef.get_profile_by_name(k8s_profile_name) profile.delete() diff --git a/src/onaptests/steps/instantiate/service_ala_carte.py b/src/onaptests/steps/instantiate/service_ala_carte.py index a2e1812..9908b6d 100644 --- a/src/onaptests/steps/instantiate/service_ala_carte.py +++ b/src/onaptests/steps/instantiate/service_ala_carte.py @@ -188,7 +188,8 @@ class YamlTemplateServiceAlaCarteInstantiateStep(YamlTemplateBaseStep): super().execute() service = Service(self.service_name) customer: Customer = Customer.get_by_global_customer_id(settings.GLOBAL_CUSTOMER_ID) - service_subscription: ServiceSubscription = customer.get_service_subscription_by_service_type(service.name) + service_subscription: ServiceSubscription = \ + customer.get_service_subscription_by_service_type(service.name) cloud_region: CloudRegion = CloudRegion.get_by_id( cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER, cloud_region_id=settings.CLOUD_REGION_ID, @@ -210,8 +211,8 @@ class YamlTemplateServiceAlaCarteInstantiateStep(YamlTemplateBaseStep): distribution_completed = service.distributed if distribution_completed is True: self._logger.info( - "Service Distribution for %s is sucessfully finished", - service.name) + "Service Distribution for %s is sucessfully finished", + service.name) break self._logger.info( "Service Distribution for %s ongoing, Wait for 60 s", @@ -221,7 +222,7 @@ class YamlTemplateServiceAlaCarteInstantiateStep(YamlTemplateBaseStep): if distribution_completed is False: self._logger.error( - "Service Distribution for %s failed !!",service.name) + "Service Distribution for %s failed !!", service.name) raise onap_test_exceptions.ServiceDistributionException service_instantiation = ServiceInstantiation.instantiate_ala_carte( @@ -243,8 +244,10 @@ class YamlTemplateServiceAlaCarteInstantiateStep(YamlTemplateBaseStep): self._logger.error("Service instantiation %s failed", self.service_instance_name) raise onap_test_exceptions.ServiceInstantiateException else: - service_subscription: ServiceSubscription = customer.get_service_subscription_by_service_type(self.service_name) - self._service_instance: ServiceInstance = service_subscription.get_service_instance_by_name(self.service_instance_name) + service_subscription: ServiceSubscription = \ + customer.get_service_subscription_by_service_type(self.service_name) + self._service_instance: ServiceInstance = \ + service_subscription.get_service_instance_by_name(self.service_instance_name) @YamlTemplateBaseStep.store_state(cleanup=True) def cleanup(self) -> None: diff --git a/src/onaptests/steps/instantiate/service_macro.py b/src/onaptests/steps/instantiate/service_macro.py index b32d936..d97a109 100644 --- a/src/onaptests/steps/instantiate/service_macro.py +++ b/src/onaptests/steps/instantiate/service_macro.py @@ -12,13 +12,23 @@ from onapsdk.aai.cloud_infrastructure.tenant import Tenant from onapsdk.configuration import settings from onapsdk.exceptions import ResourceNotFound from onapsdk.sdc.service import Service -from onapsdk.so.instantiation import InstantiationParameter, ServiceInstantiation, VfmoduleParameters, VnfParameters, SoService -from onaptests.steps.cloud.customer_service_subscription_create import CustomerServiceSubscriptionCreateStep +from onapsdk.so.instantiation import ( + InstantiationParameter, + ServiceInstantiation, + VfmoduleParameters, + VnfParameters, + SoService +) +from onaptests.steps.cloud.customer_service_subscription_create import ( + CustomerServiceSubscriptionCreateStep +) import onaptests.utils.exceptions as onap_test_exceptions from onaptests.steps.base import YamlTemplateBaseStep from onaptests.steps.onboard.service import YamlTemplateServiceOnboardStep -from onaptests.steps.cloud.connect_service_subscription_to_cloud_region import ConnectServiceSubToCloudRegionStep +from onaptests.steps.cloud.connect_service_subscription_to_cloud_region import ( + ConnectServiceSubToCloudRegionStep +) class YamlTemplateServiceMacroInstantiateStep(YamlTemplateBaseStep): @@ -48,7 +58,6 @@ class YamlTemplateServiceMacroInstantiateStep(YamlTemplateBaseStep): else: # only pnfs self.add_step(CustomerServiceSubscriptionCreateStep(cleanup)) - @property def description(self) -> str: """Step description.""" @@ -125,7 +134,7 @@ class YamlTemplateServiceMacroInstantiateStep(YamlTemplateBaseStep): return self.parent.service_instance_name @YamlTemplateBaseStep.store_state - def execute(self): + def execute(self): # noqa """Instantiate service. Use settings values: @@ -143,7 +152,8 @@ class YamlTemplateServiceMacroInstantiateStep(YamlTemplateBaseStep): super().execute() service = Service(self.service_name) customer: Customer = Customer.get_by_global_customer_id(settings.GLOBAL_CUSTOMER_ID) - service_subscription: ServiceSubscription = customer.get_service_subscription_by_service_type(service.name) + service_subscription: ServiceSubscription = \ + customer.get_service_subscription_by_service_type(service.name) if any( filter(lambda x: x in self.yaml_template[self.service_name].keys(), ["vnfs", "networks"])): @@ -153,8 +163,9 @@ class YamlTemplateServiceMacroInstantiateStep(YamlTemplateBaseStep): ) tenant: Tenant = cloud_region.get_tenant(settings.TENANT_ID) else: - cloud_region, tenant = None, None # Only PNF is going to be instantiated so - # neither cloud_region nor tenant are needed + # Only PNF is going to be instantiated so + # neither cloud_region nor tenant are needed + cloud_region, tenant = None, None try: owning_entity = OwningEntity.get_by_owning_entity_name( settings.OWNING_ENTITY) @@ -171,12 +182,12 @@ class YamlTemplateServiceMacroInstantiateStep(YamlTemplateBaseStep): distribution_completed = service.distributed if distribution_completed is True: self._logger.info( - "Service Distribution for %s is sucessfully finished", - service.name) + "Service Distribution for %s is sucessfully finished", + service.name) break self._logger.info( "Service Distribution for %s ongoing, Wait for %d s", - service.name,settings.SERVICE_DISTRIBUTION_SLEEP_TIME) + service.name, settings.SERVICE_DISTRIBUTION_SLEEP_TIME) time.sleep(settings.SERVICE_DISTRIBUTION_SLEEP_TIME) nb_try += 1 @@ -190,16 +201,19 @@ class YamlTemplateServiceMacroInstantiateStep(YamlTemplateBaseStep): if settings.MODEL_YAML_TEMPLATE: so_data = self.yaml_template[self.service_name] so_service = SoService(vnfs=so_data.get("vnfs", []), - subscription_service_type=so_data.get('subscription_service_type')) + subscription_service_type=so_data.get( + 'subscription_service_type')) else: for vnf_data in self.yaml_template[self.service_name].get("vnfs", []): vnf_params_list.append(VnfParameters( vnf_data["vnf_name"], - [InstantiationParameter(name=parameter["name"], value=parameter["value"]) for parameter in + [InstantiationParameter(name=parameter["name"], + value=parameter["value"]) for parameter in vnf_data.get("vnf_parameters", [])], [VfmoduleParameters(vf_module_data["vf_module_name"], - [InstantiationParameter(name=parameter["name"], value=parameter["value"]) for - parameter in vf_module_data.get("parameters", [])]) \ + [InstantiationParameter(name=parameter["name"], + value=parameter["value"]) for + parameter in vf_module_data.get("parameters", [])]) for vf_module_data in vnf_data.get("vf_module_parameters", [])] )) @@ -227,8 +241,10 @@ class YamlTemplateServiceMacroInstantiateStep(YamlTemplateBaseStep): self._logger.error("Service instantiation %s failed", self.service_instance_name) raise onap_test_exceptions.ServiceInstantiateException - service_subscription: ServiceSubscription = customer.get_service_subscription_by_service_type(self.service_name) - self._service_instance: ServiceInstance = service_subscription.get_service_instance_by_name(self.service_instance_name) + service_subscription: ServiceSubscription = \ + customer.get_service_subscription_by_service_type(self.service_name) + self._service_instance: ServiceInstance = \ + service_subscription.get_service_instance_by_name(self.service_instance_name) @YamlTemplateBaseStep.store_state(cleanup=True) def cleanup(self) -> None: diff --git a/src/onaptests/steps/instantiate/vf_module_ala_carte.py b/src/onaptests/steps/instantiate/vf_module_ala_carte.py index 38bdee9..b4a7c77 100644 --- a/src/onaptests/steps/instantiate/vf_module_ala_carte.py +++ b/src/onaptests/steps/instantiate/vf_module_ala_carte.py @@ -12,6 +12,7 @@ from ..base import YamlTemplateBaseStep from .vnf_ala_carte import YamlTemplateVnfAlaCarteInstantiateStep from .k8s_profile_create import K8SProfileStep + class YamlTemplateVfModuleAlaCarteInstantiateStep(YamlTemplateBaseStep): """Instantiate vf module a'la carte using YAML template.""" @@ -129,8 +130,10 @@ class YamlTemplateVfModuleAlaCarteInstantiateStep(YamlTemplateBaseStep): """ super().execute() customer: Customer = Customer.get_by_global_customer_id(settings.GLOBAL_CUSTOMER_ID) - service_subscription: ServiceSubscription = customer.get_service_subscription_by_service_type(self.service_name) - self._service_instance: ServiceInstance = service_subscription.get_service_instance_by_name(self.service_instance_name) + service_subscription: ServiceSubscription = \ + customer.get_service_subscription_by_service_type(self.service_name) + self._service_instance: ServiceInstance = \ + service_subscription.get_service_instance_by_name(self.service_instance_name) cloud_region: CloudRegion = CloudRegion.get_by_id( cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER, cloud_region_id=settings.CLOUD_REGION_ID, diff --git a/src/onaptests/steps/instantiate/vl_ala_carte.py b/src/onaptests/steps/instantiate/vl_ala_carte.py index 7186c24..022a8b2 100644 --- a/src/onaptests/steps/instantiate/vl_ala_carte.py +++ b/src/onaptests/steps/instantiate/vl_ala_carte.py @@ -101,7 +101,7 @@ class YamlTemplateVlAlaCarteInstantiateStep(YamlTemplateBaseStep): """ # workaround, as Network name differs from model name (added " 0") - network_name=re.sub(r"\s\d$", r"", network_name) + network_name = re.sub(r"\s\d$", r"", network_name) for net in self.yaml_template[self.service_name]["networks"]: if net["vl_name"] == network_name: if net['subnets'] is None: @@ -130,11 +130,13 @@ class YamlTemplateVlAlaCarteInstantiateStep(YamlTemplateBaseStep): super().execute() service: Service = Service(self.service_name) customer: Customer = Customer.get_by_global_customer_id(settings.GLOBAL_CUSTOMER_ID) - service_subscription: ServiceSubscription = customer.get_service_subscription_by_service_type(self.service_name) - service_instance: ServiceInstance = service_subscription.get_service_instance_by_name(self.service_instance_name) + service_subscription: ServiceSubscription = \ + customer.get_service_subscription_by_service_type(self.service_name) + service_instance: ServiceInstance = \ + service_subscription.get_service_instance_by_name(self.service_instance_name) self._service_instance = service_instance for idx, network in enumerate(service.networks): - #for network in self.yaml_template[self.service_name]["networks"]: + # for network in self.yaml_template[self.service_name]["networks"]: net_instantiation = service_instance.add_network( network, settings.LINE_OF_BUSINESS, @@ -159,7 +161,7 @@ class YamlTemplateVlAlaCarteInstantiateStep(YamlTemplateBaseStep): """ if self._cleanup: for net_instance in self._service_instance.network_instances: - self._logger.info("Start network deletion %s",net_instance.name) + self._logger.info("Start network deletion %s", net_instance.name) net_deletion = net_instance.delete(a_la_carte=True) try: net_deletion.wait_for_finish(settings.ORCHESTRATION_REQUEST_TIMEOUT) diff --git a/src/onaptests/steps/instantiate/vnf_ala_carte.py b/src/onaptests/steps/instantiate/vnf_ala_carte.py index eb9896f..379e285 100644 --- a/src/onaptests/steps/instantiate/vnf_ala_carte.py +++ b/src/onaptests/steps/instantiate/vnf_ala_carte.py @@ -103,8 +103,10 @@ class YamlTemplateVnfAlaCarteInstantiateStep(YamlTemplateBaseStep): super().execute() service: Service = Service(self.service_name) customer: Customer = Customer.get_by_global_customer_id(settings.GLOBAL_CUSTOMER_ID) - service_subscription: ServiceSubscription = customer.get_service_subscription_by_service_type(self.service_name) - self._service_instance: ServiceInstance = service_subscription.get_service_instance_by_name(self.service_instance_name) + service_subscription: ServiceSubscription = \ + customer.get_service_subscription_by_service_type(self.service_name) + self._service_instance: ServiceInstance = \ + service_subscription.get_service_instance_by_name(self.service_instance_name) cloud_region: CloudRegion = CloudRegion.get_by_id( cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER, cloud_region_id=settings.CLOUD_REGION_ID, |