aboutsummaryrefslogtreecommitdiffstats
path: root/src/onaptests/steps/instantiate
diff options
context:
space:
mode:
Diffstat (limited to 'src/onaptests/steps/instantiate')
-rw-r--r--src/onaptests/steps/instantiate/k8s_profile_create.py15
-rw-r--r--src/onaptests/steps/instantiate/service_ala_carte.py15
-rw-r--r--src/onaptests/steps/instantiate/service_macro.py50
-rw-r--r--src/onaptests/steps/instantiate/vf_module_ala_carte.py7
-rw-r--r--src/onaptests/steps/instantiate/vl_ala_carte.py12
-rw-r--r--src/onaptests/steps/instantiate/vnf_ala_carte.py6
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,