xtesting==0.91.0
avionix>=0.4.5
openstacksdk>=0.61.0
-onapsdk==12.7.3
+onapsdk==12.9.0
jinja2>3
kubernetes>=22.6.0
setuptools==65.3.0
}
return entry_points_result
-def run_test(test_name, validation, entry_point, settings_module):
+def run_test(test_name, validation, force_cleanup, entry_point, settings_module):
settings_env = "ONAP_PYTHON_SDK_SETTINGS"
if validation:
validation_env = "PYTHON_SDK_TESTS_VALIDATION"
os.environ[validation_env] = "True"
+ if force_cleanup:
+ validation_env = "PYTHON_SDK_TESTS_FORCE_CLEANUP"
+ os.environ[validation_env] = "True"
os.environ[settings_env] = f"onaptests.configuration.{test_name}_settings"
if not settings_module:
settings_module = importlib.import_module("onapsdk.configuration")
print("\nExample: python run_test.py basic_cps\n")
exit(1)
validation = len(argv) > 1
+ force_cleanup = len(argv) > 2
test_name = argv[0]
entry_points = get_entrypoints()
if test_name == "all":
settings_module = None
for test_name, entry_point in entry_points.items():
- settings_module = run_test(test_name, validation, entry_point, settings_module)
+ settings_module = run_test(test_name, validation, force_cleanup, entry_point, settings_module)
else:
entry_point = entry_points[test_name]
- run_test(test_name, validation, entry_point, None)
+ run_test(test_name, validation, force_cleanup, entry_point, None)
if __name__ == "__main__":
main(sys.argv[1:])
def model_yaml_template(self) -> dict:
return {}
- @property
- def service_name(self) -> dict:
- """Service name.
-
- Get from YAML template.
-
- Returns:
- str: Service name
-
- """
- return next(iter(self.yaml_template.keys()))
-
@property
def service_instance_name(self) -> str:
"""Service instance name.
def model_yaml_template(self) -> dict:
return {}
- @property
- def service_name(self) -> dict:
- """Service name.
-
- Get from YAML template.
-
- Returns:
- str: Service name
-
- """
- return next(iter(self.yaml_template.keys()))
-
@property
def service_instance_name(self) -> str:
"""Service instance name.
self._model_yaml_template: dict = load(model_yaml_template)
return self._model_yaml_template
- @property
- def service_name(self) -> dict:
- """Service name.
-
- Get from YAML template.
-
- Returns:
- str: Service name
-
- """
- return next(iter(self.yaml_template.keys()))
-
@property
def service_instance_name(self) -> str:
"""Service instance name.
def model_yaml_template(self) -> dict:
return {}
- @property
- def service_name(self) -> dict:
- """Service name.
-
- Get from YAML template.
-
- Returns:
- str: Service name
-
- """
- return next(iter(self.yaml_template.keys()))
-
@property
def service_instance_name(self) -> str:
"""Service instance name.
from abc import ABC, abstractmethod
from typing import Iterator, List, Optional
-from onapsdk.aai.business import Customer
+from onapsdk.aai.business import Customer, ServiceInstance, ServiceSubscription
from onapsdk.configuration import settings
from onapsdk.exceptions import SDKException, SettingsError
from onaptests.steps.reports_collection import (Report, ReportsCollection,
IF_VALIDATION = "PYTHON_SDK_TESTS_VALIDATION"
+IF_FORCE_CLEANUP = "PYTHON_SDK_TESTS_FORCE_CLEANUP"
class StoreStateHandler(ABC):
if cleanup:
self._start_cleanup_time = time.time()
try:
- if (self._executed and self._cleanup and
+ if (self._cleanup and self._state_execute and
(self._is_validation_only or
self.check_preconditions(cleanup=True))):
self._log_execution_state("START", cleanup)
- if not self._is_validation_only:
+ if not self._is_validation_only or self._is_force_cleanup:
fun(self, *args, **kwargs)
self._cleaned_up = True
execution_status = ReportStepStatus.PASS
self._nesting_level: int = 0
self._break_on_error: bool = break_on_error
self._is_validation_only = os.environ.get(IF_VALIDATION) is not None
+ self._is_force_cleanup = os.environ.get(IF_FORCE_CLEANUP) is not None
def add_step(self, step: "BaseStep") -> None:
"""Add substep.
class YamlTemplateBaseStep(BaseStep, ABC):
"""Base YAML template step."""
+ def __init__(self, cleanup: bool):
+ """Initialize step."""
+
+ super().__init__(cleanup=cleanup)
+ self._service_instance: ServiceInstance = None
+ self._service_subscription: ServiceSubscription = None
+ self._customer: Customer = None
+
+ def _load_customer_and_subscription(self, reload: bool = False):
+ if self._customer is None:
+ self._customer: Customer = \
+ Customer.get_by_global_customer_id(settings.GLOBAL_CUSTOMER_ID)
+ if self._service_subscription is None or reload:
+ self._service_subscription: ServiceSubscription = \
+ self._customer.get_service_subscription_by_service_type(self.service_name)
+
+ def _load_service_instance(self):
+ if self._service_instance is None:
+ self._service_instance: ServiceInstance = \
+ self._service_subscription.get_service_instance_by_name(self.service_instance_name)
+
+ @property
+ def service_name(self) -> str:
+ """Service name.
+
+ Get from YAML template if it's a root step, get from parent otherwise.
+
+ Returns:
+ str: Service name
+
+ """
+ if self.is_root:
+ return next(iter(self.yaml_template.keys()))
+ return self.parent.service_name
+
+ @property
+ def service_instance_name(self) -> str:
+ """Service instance name.
+
+ Generate service instance name.
+ If not applicable None is returned
+
+ Returns:
+ str: Service instance name
+
+ """
+ return None
+
@property
@abstractmethod
def yaml_template(self) -> dict:
from yaml import load, SafeLoader
from onapsdk.aai.cloud_infrastructure import CloudRegion, Tenant
-from onapsdk.aai.business import Customer, ServiceInstance, ServiceSubscription
+from onapsdk.aai.business import Customer
from onapsdk.aai.business.owning_entity import OwningEntity as AaiOwningEntity
from onapsdk.configuration import settings
from onapsdk.exceptions import ResourceNotFound
super().__init__(cleanup=settings.CLEANUP_FLAG)
self._yaml_template: dict = None
self._service_instance_name: str = None
- self._service_instance: str = None
if not settings.ONLY_INSTANTIATE:
self.add_step(YamlTemplateServiceOnboardStep())
self.add_step(ConnectServiceSubToCloudRegionStep())
def model_yaml_template(self) -> dict:
return {}
- @property
- def service_name(self) -> str:
- """Service name.
-
- Get from YAML template if it's a root step, get from parent otherwise.
-
- Returns:
- str: Service name
-
- """
- if self.is_root:
- return next(iter(self.yaml_template.keys()))
- return self.parent.service_name
-
@property
def service_instance_name(self) -> str:
"""Service instance name.
"""
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)
+ self._load_customer_and_subscription()
cloud_region: CloudRegion = CloudRegion.get_by_id(
cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
cloud_region_id=settings.CLOUD_REGION_ID,
service,
cloud_region,
tenant,
- customer,
+ self._customer,
owning_entity,
settings.PROJECT,
- service_subscription,
+ self._service_subscription,
service_instance_name=self.service_instance_name
)
try:
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)
+ self._load_customer_and_subscription(reload=True)
+ self._load_service_instance()
@YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
Exception: Service cleaning failed
"""
- service_deletion = self._service_instance.delete(a_la_carte=True)
- try:
- service_deletion.wait_for_finish(settings.ORCHESTRATION_REQUEST_TIMEOUT)
- except TimeoutError:
- self._logger.error("Service deletion %s timed out", self._service_instance_name)
- raise onap_test_exceptions.ServiceCleanupException
- if service_deletion.finished:
- self._logger.info("Service %s deleted", self._service_instance_name)
- else:
- self._logger.error("Service deletion %s failed", self._service_instance_name)
- raise onap_test_exceptions.ServiceCleanupException
+ self._load_customer_and_subscription()
+ self._load_service_instance()
+ if self._service_instance:
+ service_deletion = self._service_instance.delete(a_la_carte=True)
+ try:
+ service_deletion.wait_for_finish(settings.ORCHESTRATION_REQUEST_TIMEOUT)
+ except TimeoutError:
+ self._logger.error("Service deletion %s timed out", self._service_instance_name)
+ raise onap_test_exceptions.ServiceCleanupException
+ if service_deletion.finished:
+ self._logger.info("Service %s deleted", self._service_instance_name)
+ else:
+ self._logger.error("Service deletion %s failed", self._service_instance_name)
+ raise onap_test_exceptions.ServiceCleanupException
super().cleanup()
import time
from typing import List
from uuid import uuid4
-from onapsdk.aai.business.service import ServiceInstance
from yaml import load, SafeLoader
-from onapsdk.aai.business.customer import Customer, ServiceSubscription
from onapsdk.aai.business.owning_entity import OwningEntity
from onapsdk.aai.cloud_infrastructure.cloud_region import CloudRegion
from onapsdk.aai.cloud_infrastructure.tenant import Tenant
self._yaml_template: dict = None
self._model_yaml_template: dict = None
self._service_instance_name: str = None
- self._service_instance: str = None
if not settings.ONLY_INSTANTIATE:
self.add_step(YamlTemplateServiceOnboardStep())
return self._model_yaml_template
return self.parent.model_yaml_template
- @property
- def service_name(self) -> str:
- """Service name.
-
- Get from YAML template if it's a root step, get from parent otherwise.
-
- Returns:
- str: Service name
-
- """
- if self.is_root:
- return next(iter(self.yaml_template.keys()))
- return self.parent.service_name
-
@property
def service_instance_name(self) -> str:
"""Service instance name.
"""
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)
+ self._load_customer_and_subscription()
if any(
filter(lambda x: x in self.yaml_template[self.service_name].keys(),
["vnfs", "networks"])):
service_instantiation = ServiceInstantiation.instantiate_macro(
sdc_service=service,
- customer=customer,
+ customer=self._customer,
owning_entity=owning_entity,
project=settings.PROJECT,
line_of_business=settings.LINE_OF_BUSINESS,
platform=settings.PLATFORM,
cloud_region=cloud_region,
tenant=tenant,
- service_subscription=service_subscription,
+ service_subscription=self._service_subscription,
service_instance_name=self.service_instance_name,
vnf_parameters=vnf_params_list,
enable_multicloud=settings.USE_MULTICLOUD,
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)
+ self._load_customer_and_subscription(reload=True)
+ self._load_service_instance()
@YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
Exception: Service cleaning failed
"""
+ self._load_customer_and_subscription()
+ self._load_service_instance()
if self._service_instance:
service_deletion = self._service_instance.delete(a_la_carte=False)
try:
from yaml import load, SafeLoader
from onapsdk.aai.cloud_infrastructure import CloudRegion, Tenant
-from onapsdk.aai.business import Customer, ServiceInstance, ServiceSubscription
from onapsdk.configuration import settings
from onapsdk.so.instantiation import InstantiationParameter
self._yaml_template: dict = None
self._service_instance_name: str = None
- self._service_instance: ServiceInstance = None
if settings.CLOUD_REGION_TYPE == settings.K8S_REGION_TYPE:
# K8SProfileStep creates the requested profile and then calls
# YamlTemplateVnfAlaCarteInstantiateStep step
def model_yaml_template(self) -> dict:
return {}
- @property
- def service_name(self) -> str:
- """Service name.
-
- Get from YAML template if it's a root step, get from parent otherwise.
-
- Returns:
- str: Service name
-
- """
- if self.is_root:
- return next(iter(self.yaml_template.keys()))
- return self.parent.service_name
-
@property
def service_instance_name(self) -> str:
"""Service instance name.
"""
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)
+ self._load_customer_and_subscription()
+ self._load_service_instance()
cloud_region: CloudRegion = CloudRegion.get_by_id(
cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
cloud_region_id=settings.CLOUD_REGION_ID,
Exception: Vf module cleaning failed
"""
- for vnf_instance in self._service_instance.vnf_instances:
- self._logger.debug("VNF instance %s found in Service Instance ",
- vnf_instance.name)
- self._logger.info("Get VF Modules")
- for vf_module in vnf_instance.vf_modules:
- self._logger.info("Delete VF Module %s",
- vf_module.name)
- vf_module_deletion = vf_module.delete(a_la_carte=True)
-
- try:
- vf_module_deletion.wait_for_finish(settings.ORCHESTRATION_REQUEST_TIMEOUT)
- if vf_module_deletion.failed:
- self._logger.error("VfModule deletion %s failed", vf_module.name)
+ if self._service_instance:
+ for vnf_instance in self._service_instance.vnf_instances:
+ self._logger.debug("VNF instance %s found in Service Instance ",
+ vnf_instance.name)
+ self._logger.info("Get VF Modules")
+ for vf_module in vnf_instance.vf_modules:
+ self._logger.info("Delete VF Module %s",
+ vf_module.name)
+ vf_module_deletion = vf_module.delete(a_la_carte=True)
+
+ try:
+ vf_module_deletion.wait_for_finish(settings.ORCHESTRATION_REQUEST_TIMEOUT)
+ if vf_module_deletion.failed:
+ self._logger.error("VfModule deletion %s failed", vf_module.name)
+ raise onap_test_exceptions.VfModuleCleanupException
+ self._logger.info("VfModule %s deleted", vf_module.name)
+ except TimeoutError:
+ self._logger.error("VfModule deletion %s timed out", vf_module.name)
raise onap_test_exceptions.VfModuleCleanupException
- self._logger.info("VfModule %s deleted", vf_module.name)
- except TimeoutError:
- self._logger.error("VfModule deletion %s timed out", vf_module.name)
- raise onap_test_exceptions.VfModuleCleanupException
super().cleanup()
from uuid import uuid4
from yaml import load, SafeLoader
-from onapsdk.aai.business import Customer, ServiceInstance, ServiceSubscription
from onapsdk.configuration import settings
from onapsdk.so.instantiation import Subnet
from onapsdk.sdc.service import Service
super().__init__(cleanup=settings.CLEANUP_FLAG)
self._yaml_template: dict = None
self._service_instance_name: str = None
- self._service_instance: ServiceInstance = None
self.add_step(YamlTemplateServiceAlaCarteInstantiateStep())
@property
def model_yaml_template(self) -> dict:
return {}
- @property
- def service_name(self) -> str:
- """Service name.
-
- Get from YAML template if it's a root step, get from parent otherwise.
-
- Returns:
- str: Service name
-
- """
- if self.is_root:
- return next(iter(self.yaml_template.keys()))
- return self.parent.service_name
-
@property
def service_instance_name(self) -> str:
"""Service instance name.
"""
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)
- self._service_instance = service_instance
+ self._load_customer_and_subscription()
+ self._load_service_instance()
for idx, network in enumerate(service.networks):
# for network in self.yaml_template[self.service_name]["networks"]:
- net_instantiation = service_instance.add_network(
+ net_instantiation = self._service_instance.add_network(
network,
settings.LINE_OF_BUSINESS,
settings.PLATFORM,
Raises:
Exception: VL cleaning failed
"""
- if self._cleanup:
+ if self._service_instance:
for net_instance in self._service_instance.network_instances:
self._logger.info("Start network deletion %s", net_instance.name)
net_deletion = net_instance.delete(a_la_carte=True)
except TimeoutError:
self._logger.error("VL deletion %s timed out", net_instance.name)
raise onap_test_exceptions.NetworkCleanupException
- super().cleanup()
+ super().cleanup()
from yaml import load, SafeLoader
from onapsdk.aai.cloud_infrastructure import CloudRegion, Tenant
-from onapsdk.aai.business import Customer, ServiceInstance, ServiceSubscription
from onapsdk.configuration import settings
from onapsdk.sdc.service import Service
super().__init__(cleanup=settings.CLEANUP_FLAG)
self._yaml_template: dict = None
self._service_instance_name: str = None
- self._service_instance: ServiceInstance = None
self.add_step(YamlTemplateServiceAlaCarteInstantiateStep())
@property
def model_yaml_template(self) -> dict:
return {}
- @property
- def service_name(self) -> str:
- """Service name.
-
- Get from YAML template if it's a root step, get from parent otherwise.
-
- Returns:
- str: Service name
-
- """
- if self.is_root:
- return next(iter(self.yaml_template.keys()))
- return self.parent.service_name
-
@property
def service_instance_name(self) -> str:
"""Service instance name.
"""
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)
+ self._load_customer_and_subscription()
+ self._load_service_instance()
cloud_region: CloudRegion = CloudRegion.get_by_id(
cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
cloud_region_id=settings.CLOUD_REGION_ID,
Exception: VNF cleaning failed
"""
- for vnf_instance in self._service_instance.vnf_instances:
- vnf_deletion = vnf_instance.delete(a_la_carte=True)
-
- try:
- vnf_deletion.wait_for_finish(settings.ORCHESTRATION_REQUEST_TIMEOUT)
- if vnf_deletion.failed:
- self._logger.error("VNF deletion %s failed", vnf_instance.name)
+ if self._service_instance:
+ for vnf_instance in self._service_instance.vnf_instances:
+ vnf_deletion = vnf_instance.delete(a_la_carte=True)
+
+ try:
+ vnf_deletion.wait_for_finish(settings.ORCHESTRATION_REQUEST_TIMEOUT)
+ if vnf_deletion.failed:
+ self._logger.error("VNF deletion %s failed", vnf_instance.name)
+ raise onap_test_exceptions.VnfCleanupException
+ except TimeoutError:
+ self._logger.error("VNF deletion %s timed out", vnf_instance.name)
raise onap_test_exceptions.VnfCleanupException
- except TimeoutError:
- self._logger.error("VNF deletion %s timed out", vnf_instance.name)
- raise onap_test_exceptions.VnfCleanupException
super().cleanup()
def model_yaml_template(self) -> dict:
return {}
- @property
- def service_name(self) -> str:
- """Service name.
-
- Get from YAML template if it's a root step, get from parent otherwise.
-
- Returns:
- str: Service name
-
- """
- if self.is_root:
- return next(iter(self.yaml_template.keys()))
- else:
- return self.parent.service_name
-
def check(self, operational_policies: list, is_template: bool = False) -> str:
"""Check CLAMP requirements to create a loop."""
self._logger.info("Check operational policy")
def model_yaml_template(self) -> dict:
return {}
- @property
- def service_name(self) -> str:
- """Service name.
-
- Get from YAML template if it's a root step, get from parent otherwise.
-
- Returns:
- str: Service name
-
- """
- if self.is_root:
- return next(iter(self.yaml_template.keys()))
- else:
- return self.parent.service_name
-
@YamlTemplateBaseStep.store_state
def execute(self):
"""Onboard service."""
return self._model_yaml_template
return self.parent.model_yaml_template
- @property
- def service_name(self) -> str:
- """Service name.
-
- Get from YAML template if it's a root step, get from parent otherwise.
-
- Returns:
- str: Service name
-
- """
- if self.is_root:
- return next(iter(self.yaml_template.keys()))
- return self.parent.service_name
-
@YamlTemplateBaseStep.store_state
def execute(self):
"""Onboard service."""
Code Review / testsuite / pythonsdk-tests.git / commitdiff