#!/usr/bin/env python
"""Basic Onboard test case."""
-from onaptests.scenario.scenario_base import ScenarioBase
-from onaptests.steps.onboard.service import YamlTemplateServiceOnboardStep
+from onapsdk.configuration import settings
+from yaml import SafeLoader, load
+
+from onaptests.scenario.scenario_base import (BaseStep, ScenarioBase,
+ YamlTemplateBaseScenarioStep)
+from onaptests.steps.onboard.service import (VerifyServiceDistributionStep,
+ YamlTemplateServiceOnboardStep)
+
+
+class BasicSdcOnboardStep(YamlTemplateBaseScenarioStep):
+ """Main basic onboard scenario step."""
+
+ def __init__(self):
+ """Initialize step.
+
+ Substeps:
+ - YamlTemplateServiceOnboardStep
+ - VerifyServiceDistributionStep (optional).
+ """
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self._yaml_template: dict = None
+ self.add_step(YamlTemplateServiceOnboardStep())
+ if settings.VERIFY_DISTRIBUTION:
+ self.add_step(VerifyServiceDistributionStep())
+
+ @property
+ def description(self) -> str:
+ """Step description.
+
+ Used for reports
+
+ Returns:
+ str: Step description
+
+ """
+ return "Basic SDC Onboard scenario step"
+
+ @property
+ def component(self) -> str:
+ """Component name.
+
+ Name of component which step is related with.
+ Most is the name of ONAP component.
+
+ Returns:
+ str: Component name
+
+ """
+ return "SDC"
+
+ @property
+ def yaml_template(self) -> dict:
+ """YAML template abstract property.
+
+ Every YAML template step need to implement that property.
+
+ Returns:
+ dict: YAML template
+
+ """
+ if not self._yaml_template:
+ with open(settings.SERVICE_YAML_TEMPLATE, "r", encoding="utf-8") as yaml_template:
+ self._yaml_template: dict = load(yaml_template, SafeLoader)
+ return self._yaml_template
+
+ @property
+ def model_yaml_template(self) -> dict:
+ return {}
class BasicOnboard(ScenarioBase):
"""Init BasicOnboard."""
# import basic_onboard_settings needed
super().__init__('basic_onboard', **kwargs)
- self.test = YamlTemplateServiceOnboardStep()
+ self.test = BasicSdcOnboardStep()
-import time
from uuid import uuid4
from onapsdk.aai.business.owning_entity import OwningEntity as AaiOwningEntity
from ..base import YamlTemplateBaseStep
from ..cloud.connect_service_subscription_to_cloud_region import \
ConnectServiceSubToCloudRegionStep
-from ..onboard.service import YamlTemplateServiceOnboardStep
+from ..onboard.service import (VerifyServiceDistributionStep,
+ YamlTemplateServiceOnboardStep)
class YamlTemplateServiceAlaCarteInstantiateStep(YamlTemplateBaseStep):
Substeps:
- YamlTemplateServiceOnboardStep,
- ConnectServiceSubToCloudRegionStep.
+ - VerifyServiceDistributionStep
+ - TestSdncStep
"""
super().__init__(cleanup=settings.CLEANUP_FLAG)
self._yaml_template: dict = None
if not settings.ONLY_INSTANTIATE:
self.add_step(YamlTemplateServiceOnboardStep())
self.add_step(ConnectServiceSubToCloudRegionStep())
+ self.add_step(VerifyServiceDistributionStep())
self.add_step(TestSdncStep(full=False))
@property
self._logger.info("Owning entity not found, create it")
owning_entity = AaiOwningEntity.create(settings.OWNING_ENTITY)
- # Before instantiating, be sure that the service has been distributed
- self._logger.info("******** Check Service Distribution *******")
- distribution_completed = False
- nb_try = 0
- nb_try_max = 10
- while distribution_completed is False and nb_try < nb_try_max:
- distribution_completed = service.distributed
- if distribution_completed is True:
- self._logger.info(
- "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)
- time.sleep(settings.SERVICE_DISTRIBUTION_SLEEP_TIME)
- nb_try += 1
-
- if distribution_completed is False:
- self._logger.error(
- "Service Distribution for %s failed !!", service.name)
- raise onap_test_exceptions.ServiceDistributionException
-
service_instantiation = ServiceInstantiation.instantiate_ala_carte(
service,
cloud_region,
from onaptests.steps.cloud.customer_service_subscription_create import \
CustomerServiceSubscriptionCreateStep
from onaptests.steps.instantiate.sdnc_service import TestSdncStep
-from onaptests.steps.onboard.service import YamlTemplateServiceOnboardStep
+from onaptests.steps.onboard.service import (VerifyServiceDistributionStep,
+ YamlTemplateServiceOnboardStep)
class YamlTemplateServiceMacroInstantiateStep(YamlTemplateBaseStep):
- YamlTemplateServiceOnboardStep,
- ConnectServiceSubToCloudRegionStep,
- CustomerServiceSubscriptionCreateStep.
+ - VerifyServiceDistributionStep
+ - TestSdncStep
"""
super().__init__(cleanup=settings.CLEANUP_FLAG)
self._yaml_template: dict = None
self.add_step(ConnectServiceSubToCloudRegionStep())
else: # only pnfs
self.add_step(CustomerServiceSubscriptionCreateStep())
+ self.add_step(VerifyServiceDistributionStep())
self.add_step(TestSdncStep(full=False))
@property
self._logger.info("Owning entity not found, create it")
owning_entity = OwningEntity.create(settings.OWNING_ENTITY)
- # Before instantiating, be sure that the service has been distributed
- self._logger.info("******** Check Service Distribution *******")
- distribution_completed = False
- nb_try = 0
- while distribution_completed is False and \
- nb_try < settings.SERVICE_DISTRIBUTION_NUMBER_OF_TRIES:
- distribution_completed = service.distributed
- if distribution_completed is True:
- self._logger.info(
- "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)
- time.sleep(settings.SERVICE_DISTRIBUTION_SLEEP_TIME)
- nb_try += 1
-
- if distribution_completed is False:
- self._logger.error(
- "Service Distribution for %s failed !!", service.name)
- raise onap_test_exceptions.ServiceDistributionException
-
so_service = None
vnf_params_list: List[VnfParameters] = []
if settings.MODEL_YAML_TEMPLATE:
-from typing import Any, Dict
+import time
+from typing import Any, Dict, Iterator
+from urllib.parse import urlencode
+from onapsdk.aai.service_design_and_creation import Model
from onapsdk.configuration import settings
-from onapsdk.exceptions import ResourceNotFound
+from onapsdk.exceptions import InvalidResponse, ResourceNotFound
from onapsdk.sdc2.component_instance import (ComponentInstance,
ComponentInstanceInput)
from onapsdk.sdc2.pnf import Pnf
from onapsdk.sdc2.service import Service, ServiceInstantiationType
from onapsdk.sdc2.vf import Vf
from onapsdk.sdc2.vl import Vl
+from onapsdk.so.catalog_db_adapter import CatalogDbAdapter
from yaml import SafeLoader, load
-from ..base import BaseStep, YamlTemplateBaseStep
-from .pnf import PnfOnboardStep, YamlTemplatePnfOnboardStep
-from .vf import VfOnboardStep, YamlTemplateVfOnboardStep
-
-
-class ServiceOnboardStep(BaseStep):
- """Service onboard step."""
-
- def __init__(self):
- """Initialize step.
-
- Substeps:
- - VfOnboardStep.
- """
- super().__init__(cleanup=settings.CLEANUP_FLAG)
- if settings.VF_NAME != "":
- self.add_step(VfOnboardStep())
- if settings.PNF_NAME != "":
- self.add_step(PnfOnboardStep())
-
- @property
- def description(self) -> str:
- """Step description."""
- return "Onboard service in SDC."
-
- @property
- def component(self) -> str:
- """Component name."""
- return "SDC"
-
- def check_preconditions(self, cleanup=False) -> bool:
- if not super().check_preconditions(cleanup):
- return False
- if cleanup:
- return settings.SDC_CLEANUP
- return True
-
- @BaseStep.store_state
- def execute(self):
- """Onboard service.
-
- Use settings values:
- - VL_NAME,
- - VF_NAME,
- - PNF_NAME,
- - SERVICE_NAME,
- - SERVICE_INSTANTIATION_TYPE.
+import onaptests.utils.exceptions as onap_test_exceptions
+from onaptests.scenario.scenario_base import BaseScenarioStep
- """
- super().execute()
- try:
- service: Service = Service.get_by_name(name=settings.SERVICE_NAME)
- if service.distributed:
- return
- except ResourceNotFound:
- service = Service.create(name=settings.SERVICE_NAME,
- instantiation_type=settings.SERVICE_INSTANTIATION_TYPE)
- if settings.VL_NAME != "":
- vl: Vl = Vl(name=settings.VL_NAME)
- service.add_resource(vl)
- if settings.VF_NAME != "":
- vf: Vf = Vf(name=settings.VF_NAME)
- service.add_resource(vf)
- if settings.PNF_NAME != "":
- pnf: Pnf = Pnf(name=settings.PNF_NAME)
- service.add_resource(pnf)
- if service.lifecycle_state != LifecycleState.CERTIFIED:
- service.lifecycle_operation(LifecycleOperation.CERTIFY)
- service.distribute()
-
- @BaseStep.store_state
- def cleanup(self) -> None:
- """Cleanup service onboard step."""
- try:
- service: Service = Service.get_by_name(name=settings.SERVICE_NAME)
- if service.lifecycle_state == LifecycleState.CERTIFIED:
- service.archive()
- service.delete()
- except ResourceNotFound:
- self._logger.info(f"Service {settings.SERVICE_NAME} not found")
- super().cleanup()
+from ..base import BaseStep, YamlTemplateBaseStep
+from .pnf import YamlTemplatePnfOnboardStep
+from .vf import YamlTemplateVfOnboardStep
class YamlTemplateServiceOnboardStep(YamlTemplateBaseStep):
except ResourceNotFound:
self._logger.info(f"Service {self.service_name} not found")
super().cleanup()
+
+
+class VerifyServiceDistributionStep(BaseScenarioStep):
+ """Service distribution check step."""
+
+ def __init__(self):
+ """Initialize step."""
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.add_step(ServiceDistributionWaitStep())
+ for notified_module in settings.SDC_SERVICE_DISTRIBUTION_COMPONENTS:
+ self.add_step(VerifyServiceDistributionStatusStep(
+ notified_module=notified_module))
+ self.add_step(VerifyServiceDistributionInSoStep())
+ self.add_step(VerifyServiceDistributionInAaiStep())
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Verify complete status of distribution"
+
+ @property
+ def component(self) -> str:
+ """Component name."""
+ return "SDC"
+
+
+class BaseServiceDistributionComponentCheckStep(BaseStep):
+ """Service distribution check step."""
+
+ def __init__(self, component_name: str, break_on_error: bool = True):
+ """Initialize step.
+
+ Args:
+ component_name (str): Name of tested component
+ break_on_error (bool): If step breaks execution when failed
+ """
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP,
+ break_on_error=break_on_error)
+ self.component_name = component_name
+ self.service: Service = None
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return f"Check service distribution in {self.component_name}."
+
+ @property
+ def component(self) -> str:
+ """Component name."""
+ return self.component_name
+
+ def execute(self):
+ """Check service distribution status."""
+ super().execute()
+ self.service = Service.get_by_name(name=settings.SERVICE_NAME)
+
+
+class ServiceDistributionWaitStep(BaseServiceDistributionComponentCheckStep):
+ """Service distribution wait step."""
+
+ def __init__(self):
+ """Initialize step."""
+ super().__init__(component_name="SDC", break_on_error=False)
+
+ @BaseStep.store_state
+ def execute(self):
+ """Wait for service distribution."""
+ super().execute()
+ # Before instantiating, be sure that the service has been distributed
+ self._logger.info("******** Check Service Distribution *******")
+ distribution_completed = False
+ nb_try = 0
+ while distribution_completed is False and \
+ nb_try < settings.SERVICE_DISTRIBUTION_NUMBER_OF_TRIES:
+ distribution_completed = self.service.distributed
+ if distribution_completed is True:
+ self._logger.info(
+ "Service Distribution for %s is sucessfully finished",
+ self.service.name)
+ break
+ self._logger.info(
+ "Service Distribution for %s ongoing, Wait for %d s",
+ self.service.name, settings.SERVICE_DISTRIBUTION_SLEEP_TIME)
+ time.sleep(settings.SERVICE_DISTRIBUTION_SLEEP_TIME)
+ nb_try += 1
+
+ if distribution_completed is False:
+ msg = f"Service Distribution for {self.service.name} failed after timeout!!"
+ self._logger.error(msg)
+ raise onap_test_exceptions.ServiceDistributionException(msg)
+
+
+class VerifyServiceDistributionStatusStep(BaseServiceDistributionComponentCheckStep):
+ """Check service distribution in SO step."""
+
+ def __init__(self, notified_module: str):
+ """Initialize step.
+
+ Args:
+ notified_module (str): Name of notified module
+ """
+
+ component_name = notified_module.split("-")[0].upper()
+ super().__init__(component_name=component_name)
+ self.component_id = notified_module
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return f"Check service distribution in {self.component_name} \
+{self.component_id}."
+
+ @BaseStep.store_state
+ def execute(self):
+ """Check service distribution status."""
+ super().execute()
+ if not self.service.distributed:
+ latest_distribution = self.service.latest_distribution
+ for status in latest_distribution.distribution_status_list:
+ if status.component_id == self.component_id and status.failed:
+ msg = f"Service {self.service.name} is not \
+distributed into [{self.component_id}]: {status.error_reason}"
+ self._logger.error(msg)
+ raise onap_test_exceptions.ServiceDistributionException(msg)
+ msg = f"Service {self.service.name} is distributed in SO and {self.component_id}."
+ self._logger.info(msg)
+
+
+class VerifyServiceDistributionInSoStep(BaseServiceDistributionComponentCheckStep):
+ """Check service distribution in SO step."""
+
+ def __init__(self):
+ """Initialize step."""
+ super().__init__(component_name="SO")
+
+ @BaseStep.store_state
+ def execute(self):
+ """Check service distribution status."""
+ super().execute()
+ try:
+ CatalogDbAdapter.get_service_info(self.service.uuid)
+ except ResourceNotFound as e:
+ msg = f"Service {self.service.name} is missing in SO."
+ self._logger.error(msg)
+ raise onap_test_exceptions.ServiceDistributionException(msg) from e
+ except InvalidResponse:
+ # looks like json returned by SO catalog DB adapter returns wrong json
+ # but we don't care here. It is important to just know if service is there
+ pass
+
+
+class VerifyServiceDistributionInAaiStep(BaseServiceDistributionComponentCheckStep):
+ """Check service distribution in AAI step."""
+
+ class ModelWithGet(Model):
+ """"Workaround to fix """
+
+ @classmethod
+ def get_all(cls,
+ invariant_id: str = None,
+ resource_version: str = None) -> Iterator["Model"]:
+ """Get all models.
+
+ Args:
+ invariant_id (str): model invariant ID
+ resource_version (str): object resource version
+
+ Yields:
+ Model: Model object
+
+ """
+ filter_parameters: dict = cls.filter_none_key_values(
+ {"model-invariant-id": invariant_id,
+ "resource-version": resource_version}
+ )
+ url: str = f"{cls.get_all_url()}?{urlencode(filter_parameters)}"
+ for model in cls.send_message_json("GET", "Get A&AI sdc models",
+ url).get("model", []):
+ yield Model(
+ invariant_id=model.get("model-invariant-id"),
+ model_type=model.get("model-type"),
+ resource_version=model.get("resource-version")
+ )
+
+ def __init__(self):
+ """Initialize step."""
+ BaseServiceDistributionComponentCheckStep.__init__(
+ self, component_name="AAI")
+
+ @BaseStep.store_state
+ def execute(self):
+ """Check service distribution status."""
+ super().execute()
+ try:
+ aai_services = self.ModelWithGet.get_all(
+ invariant_id=self.service.invariant_uuid)
+ for aai_service in aai_services:
+ self._logger.info(
+ f"Resolved {aai_service.invariant_id} aai service")
+ except ResourceNotFound as e:
+ msg = f"Service {self.service.name} is missing in AAI."
+ self._logger.error(msg)
+ raise onap_test_exceptions.ServiceDistributionException(msg) from e
Code Review / testsuite / pythonsdk-tests.git / commitdiff