xtesting==0.91.0
avionix>=0.4.5
openstacksdk>=0.61.0
-onapsdk==12.9.0
+onapsdk==12.11.0
jinja2>3
kubernetes>=22.6.0
setuptools==65.3.0
def get_entrypoints():
config = configparser.ConfigParser()
config.read('setup.cfg')
- entry_points = config['options.entry_points']['xtesting.testcase']
+ entry_points = config['entry_points']['xtesting.testcase']
config = configparser.ConfigParser()
config.read_string(f"[entry_points]\n{entry_points}")
entry_points = config['entry_points']
def main(argv):
"""Script is used to run one or all the tests.
-
+
You need to specify a name of the test like 'basic_cps' or
keyword 'all' that tells to run all the tests. You can also
pass a second argument of any value that tells the script to run
clearwater_ims = onaptests.scenario.clearwater_ims:ClearwaterIms
basic_onboard = onaptests.scenario.basic_onboard:BasicOnboard
pnf_macro = onaptests.scenario.pnf_macro:PnfMacro
- basic_clamp = onaptests.scenario.basic_clamp:BasicClamp
cds_resource_resolution = onaptests.scenario.cds_resource_resolution:CDSResourceResolution
multi_vnf_macro = onaptests.scenario.multi_vnf_macro:MultiVnfUbuntuMacro
basic_cnf_macro = onaptests.scenario.basic_cnf_macro:BasicCnfMacro
+++ /dev/null
-from yaml import SafeLoader, load
-
-from onaptests.utils.resources import get_resource_location
-
-from .settings import * # noqa
-
-CLEANUP_FLAG = False
-CLAMP_DISTRIBUTION_TIMER = 10
-
-# The ONAP part
-SERVICE_DETAILS = ("Onboarding, enriching a model with TCA." +
- "Design a loop with Clamp and deploy it in Policy and DCAE")
-
-VENDOR_NAME = "basiclamp_vendor"
-
-VSP_NAME = "basiclamp_vsp"
-
-OPERATIONAL_POLICIES = [
- {
- "name": "MinMax",
- "policy_type": "onap.policies.controlloop.guard.common.MinMax",
- "policy_version": "1.0.0",
- "config_function": "add_minmax_config", # func
- "configuration": {
- "min": 1,
- "max": 10
- }
- },
- {
- "name": "FrequencyLimiter",
- "policy_type": "onap.policies.controlloop.guard.common.FrequencyLimiter",
- "policy_version": "1.0.0",
- "config_function": "add_frequency_limiter", # func
- "configuration": {}
- }
-]
-
-# if a yaml file is define, retrieve info from this yaml files
-# if not declare the parameters in the settings
-SERVICE_YAML_TEMPLATE = get_resource_location("templates/vnf-services/basic_clamp-service.yaml")
-CONFIGURATION_PATH = get_resource_location("configuration/")
-
-try:
- # Try to retrieve the SERVICE NAME from the yaml file
- with open(SERVICE_YAML_TEMPLATE, "r", encoding="utf-8") as yaml_template:
- yaml_config_file = load(yaml_template, SafeLoader)
- SERVICE_NAME = next(iter(yaml_config_file.keys()))
- VF_NAME = SERVICE_NAME
-except ValueError:
- SERVICE_NAME = "" # Fill me
- VF_NAME = "" # Fill me
-
-MODEL_YAML_TEMPLATE = None
SERVICE_YAML_TEMPLATE = Path(get_resource_location(
"templates/vnf-services/basic_cnf_macro-service.yaml"))
+generate_service_config_yaml_file(service_name="basic_cnf_macro", # noqa
+ service_template="basic_cnf_macro-service.yaml.j2",
+ service_config=SERVICE_YAML_TEMPLATE)
try:
# Try to retrieve the SERVICE NAME from the yaml file
-import random
-import string
-from jinja2 import Environment, PackageLoader
from yaml import SafeLoader, load
import onaptests.utils.exceptions as onap_test_exceptions
from .settings import * # noqa
-# We need to create a service file with a random service name,
-# to be sure that we force onboarding
-def generate_service_config_yaml_file():
- """ generate the service file with a random service name
- from a jinja template"""
-
- env = Environment(
- loader=PackageLoader('onaptests', 'templates/vnf-services'),
- )
- template = env.get_template('basic_onboard-service.yaml.j2')
-
- # get a random string to randomize the vnf name
- # Random string with the combination of lower and upper case
- letters = string.ascii_letters
- result_str = ''.join(random.choice(letters) for i in range(6))
- service_name = 'basic_onboard_' + result_str
-
- rendered_template = template.render(service_name=service_name)
-
- with open(SERVICE_YAML_TEMPLATE, 'w+', encoding="utf-8") as file_to_write:
- file_to_write.write(rendered_template)
-
-
# The ONAP part
SERVICE_DETAILS = "Basic onboard service to only onboard a service in SDC"
# if a yaml file is define, retrieve info from this yaml files
# if not declare the parameters in the settings
+MODEL_YAML_TEMPLATE = None
+CLEANUP_FLAG = True
+SDC_CLEANUP = True
+
SERVICE_YAML_TEMPLATE = get_resource_location("templates/vnf-services/basic-onboard-service.yaml")
-generate_service_config_yaml_file()
+generate_service_config_yaml_file(service_name="basic_onboard", # noqa
+ service_template="basic_onboard-service.yaml.j2",
+ service_config=SERVICE_YAML_TEMPLATE,
+ generate_random_names=SDC_CLEANUP)
try:
# Try to retrieve the SERVICE NAME from the yaml file
# CLEANUP_FLAG = True
# CLEANUP_ACTIVITY_TIMER = 10 # nb of seconds before cleanup in case cleanup option is set
VENDOR_NAME = "basic_onboard_vendor"
-
-MODEL_YAML_TEMPLATE = None
-CLEANUP_FLAG = True
-import os
from pathlib import Path
from uuid import uuid4
-import openstack
from yaml import SafeLoader, load
import onaptests.utils.exceptions as onap_test_exceptions
from onaptests.utils.resources import get_resource_location
from .settings import * # noqa
-from .settings import IF_VALIDATION
SERVICE_DETAILS = "Onboarding, distribution and instanitation of an Ubuntu VM using macro"
GLOBAL_CUSTOMER_ID = "basicvm-customer"
-if not IF_VALIDATION:
- TEST_CLOUD = os.getenv('OS_TEST_CLOUD')
- cloud = openstack.connect(cloud=TEST_CLOUD)
- VIM_USERNAME = cloud.config.auth.get('username', 'Fill me')
- VIM_PASSWORD = cloud.config.auth.get('password', 'Fill me')
- VIM_SERVICE_URL = cloud.config.auth.get('auth_url', 'Fill me')
- TENANT_ID = cloud.config.auth.get('project_id', 'Fill me')
- TENANT_NAME = cloud.config.auth.get('project_name', 'Fill me')
- CLOUD_REGION_ID = cloud.config.auth.get('region_name', 'RegionOne')
- CLOUD_DOMAIN = cloud.config.auth.get('project_domain_name', 'Default')
+# if not IF_VALIDATION:
+# TEST_CLOUD = os.getenv('OS_TEST_CLOUD')
+# cloud = openstack.connect(cloud=TEST_CLOUD)
+# VIM_USERNAME = cloud.config.auth.get('username', 'Fill me')
+# VIM_PASSWORD = cloud.config.auth.get('password', 'Fill me')
+# VIM_SERVICE_URL = cloud.config.auth.get('auth_url', 'Fill me')
+# TENANT_ID = cloud.config.auth.get('project_id', 'Fill me')
+# TENANT_NAME = cloud.config.auth.get('project_name', 'Fill me')
+# CLOUD_REGION_ID = cloud.config.auth.get('region_name', 'RegionOne')
+# CLOUD_DOMAIN = cloud.config.auth.get('project_domain_name', 'Default')
OWNING_ENTITY = "basicvm-oe"
PROJECT = "basicvm-project"
CLOUD_DOMAIN = "Default"
SERVICE_YAML_TEMPLATE = Path(get_resource_location(
"templates/vnf-services/basic_vm_macro-service.yaml"))
+generate_service_config_yaml_file(service_name="basic_vm_macro", # noqa
+ service_template="basic_vm_macro-service.yaml.j2",
+ service_config=SERVICE_YAML_TEMPLATE)
try:
# Try to retrieve the SERVICE NAME from the yaml file
# if a yaml file is define, retrieve info from this yaml files
# if not declare the parameters in the settings
SERVICE_YAML_TEMPLATE = get_resource_location("templates/vnf-services/basic_vm-service.yaml")
+generate_service_config_yaml_file(service_name="basic_vm", # noqa
+ service_template="basic_vm-service.yaml.j2",
+ service_config=SERVICE_YAML_TEMPLATE)
try:
# Try to retrieve the SERVICE NAME from the yaml file
from pathlib import Path
import openstack
-from jinja2 import Environment, PackageLoader
from onaptests.utils.resources import get_resource_location
SERVICE_DETAILS = "Onboarding, distribution and instanitation of an Mutli VM service using macro"
-# We need to create a service file with a random service name,
-# to be sure that we force onboarding
-def generate_service_config_yaml_file(filename):
- """generate the service file with a random service name
- from a jinja template"""
-
- env = Environment(
- loader=PackageLoader('onaptests', 'templates/vnf-services'),
- )
- template = env.get_template(f"{filename}.yaml.j2")
-
- rendered_template = template.render(service_name=SERVICE_NAME)
-
- file_name = get_resource_location(f"templates/vnf-services/{filename}.yaml")
-
- with open(file_name, 'w+', encoding="utf-8") as file_to_write:
- file_to_write.write(rendered_template)
-
-
CLEANUP_FLAG = True
CDS_DD_FILE = Path(get_resource_location("templates/artifacts/dd_nso_ubuntu.json"))
f"{VNF_FILENAME_PREFIX}-model.yaml"))
-generate_service_config_yaml_file(f"{VNF_FILENAME_PREFIX}-service")
-generate_service_config_yaml_file(f"{VNF_FILENAME_PREFIX}-model")
+generate_service_config_yaml_file(f"{VNF_FILENAME_PREFIX}-service", # noqa
+ f"{VNF_FILENAME_PREFIX}-service.yaml.j2",
+ str(SERVICE_YAML_TEMPLATE))
+generate_service_config_yaml_file(f"{VNF_FILENAME_PREFIX}-model", # noqa
+ f"{VNF_FILENAME_PREFIX}-model.yaml.j2",
+ str(MODEL_YAML_TEMPLATE))
SERVICE_INSTANCE_NAME = f"{SERVICE_NAME}_svc"
SERVICE_DETAILS = "Onboarding, distribution and registration of PNF using macro"
SERVICE_INSTANCE_NAME = "TestPNFMacroInstantiation"
SERVICE_YAML_TEMPLATE = Path(get_resource_location("templates/vnf-services/pnf-service.yaml"))
+generate_service_config_yaml_file(service_name="pnf_macro", # noqa
+ service_template="pnf-service.yaml.j2",
+ service_config=SERVICE_YAML_TEMPLATE)
CDS_DD_FILE = Path(get_resource_location("templates/artifacts/dd.json"))
CDS_CBA_UNENRICHED = Path(get_resource_location("templates/artifacts/PNF_DEMO.zip"))
# #
######################
+import random
+import string
+from jinja2 import Environment, PackageLoader
# Variables to set logger information
# Possible values for logging levels in onapsdk: INFO, DEBUG , WARNING, ERROR
IN_CLUSTER = False
VES_BASIC_AUTH = {'username': 'sample1', 'password': 'sample1'}
IF_VALIDATION = False
+
+
+# We need to create a service file with a random service name,
+# to be sure that we force onboarding
+def generate_service_config_yaml_file(service_name: str,
+ service_template: str,
+ service_config: str,
+ generate_random_names: bool = False):
+ """Generate service config YAML file.
+
+ Service configurations (both models and instances) are stored in YAML files
+ mostly generated by filling Jinja templates with service names. For most
+ cases we are generate the same configuration for all runs
+ (so generate_random_names is set to False, as default) but it is possible to
+ create all resources on each test execution.
+
+ Args:
+ service_name (str): Name of the service
+ service_template (str): Template which would be used to generate configuration
+ service_config (str): Configuration output file path
+ generate_random_names (bool, optional): Flag indicating whether service name
+ should have a random suffix or not. Defaults to False.
+
+ """
+
+ env = Environment(
+ loader=PackageLoader('onaptests', 'templates/vnf-services'),
+ )
+ template = env.get_template(service_template)
+
+ if generate_random_names:
+ # get a random string to randomize the vnf name
+ # Random string with the combination of lower and upper case
+ letters = string.ascii_letters
+ result_str = ''.join(random.choice(letters) for i in range(6))
+ service_name = f"{service_name}_{result_str}"
+
+ rendered_template = template.render(service_name=service_name)
+
+ with open(service_config, 'w+', encoding="utf-8") as file_to_write:
+ file_to_write.write(rendered_template)
+++ /dev/null
-"""Basic Clamp test case."""
-from onaptests.scenario.scenario_base import ScenarioBase
-from onaptests.steps.loop.clamp import ClampStep
-
-
-class BasicClamp(ScenarioBase):
- """Onboard, update a model with a loop, design the loop and deploy it."""
-
- def __init__(self, **kwargs):
- """Init Basic Clamp, onboard a VM, design and deploy a loop with CLAMP."""
- super().__init__('basic_clamp', **kwargs)
- self.test = ClampStep()
def __init__(self, **kwargs):
"""Init CDS resource resolution use case."""
- super().__init__('basic_cds', **kwargs)
+ super().__init__('cds_resource_resolution', **kwargs)
self.test = CDSResourceResolutionStep()
def __init__(self, **kwargs):
"""Init Basic Macro use case."""
- super().__init__('nso_ubuntu_macro', **kwargs)
+ super().__init__('multi_vnf_macro', **kwargs)
self.test = MultiVnfUbuntuMacroStep()
+++ /dev/null
-#!/usr/bin/python
-#
-# This program and the accompanying materials
-# are made available under the terms of the Apache License, Version 2.0
-# which accompanies this distribution, and is available at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-"""Clamp Scenario class."""
-import random
-import string
-import time
-
-from onapsdk.clamp.clamp_element import Clamp
-from onapsdk.configuration import settings
-from onapsdk.sdc.service import Service
-from yaml import SafeLoader, load
-
-import onaptests.utils.exceptions as onap_test_exceptions
-from onaptests.steps.base import YamlTemplateBaseStep
-from onaptests.steps.loop.instantiate_loop import InstantiateLoop
-from onaptests.steps.onboard.clamp import OnboardClampStep
-
-
-class ClampStep(YamlTemplateBaseStep):
- """class defining the different CLAMP scenarios."""
-
- count: int = 0
-
- def __init__(self):
- super().__init__(cleanup=settings.CLEANUP_FLAG)
- self._yaml_template: dict = None
- self.add_step(OnboardClampStep())
- Clamp()
- self.loop_instance = None
-
- @property
- def description(self) -> str:
- """Step description."""
- return "Retrieve TCA, Policy then create a loop"
-
- @property
- def component(self) -> str:
- """Component name."""
- return "CLAMP"
-
- @property
- def yaml_template(self) -> dict:
- """Step YAML template.
-
- Load from file if it's a root step, get from parent otherwise.
-
- Returns:
- dict: Step YAML template
-
- """
- if self.is_root:
- 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
- return self.parent.yaml_template
-
- @property
- def model_yaml_template(self) -> dict:
- return {}
-
- def check(self, operational_policies: list, is_template: bool = False) -> str:
- """Check CLAMP requirements to create a loop."""
- self._logger.info("Check operational policy")
- for policy in operational_policies:
- exist = Clamp.check_policies(policy_name=policy["name"],
- req_policies=30) # 30 required policy
- self._logger.info("Operational policy found.")
- if not exist:
- raise ValueError("Couldn't load the policy %s" % policy)
- # retrieve the service..based on service name
- service: Service = Service(self.service_name)
- if is_template:
- loop_template = Clamp.check_loop_template(service=service)
- self._logger.info("Loop template checked.")
- return loop_template
- return None
-
- def instantiate_clamp(self, loop_template: str, loop_name: str, operational_policies: list):
- """Instantite a closed loopin CLAMP."""
- letters = string.ascii_letters
- loop_name_random = loop_name.join(
- random.choice(letters) for i in range(6))
- loop = InstantiateLoop(template=loop_template,
- loop_name=loop_name_random,
- operational_policies=operational_policies)
- return loop.instantiate_loop()
-
- def loop_counter(self, action: str) -> None:
- """ Count number of loop instances."""
- if action == "plus":
- self.count += 1
- if action == "minus":
- self.count -= 1
-
- @YamlTemplateBaseStep.store_state
- def execute(self):
- super().execute() # TODO work only the 1st time, not if already onboarded
-
- # Before instantiating, be sure that the service has been distributed
- service = Service(self.service_name)
- 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
-
- # time to wait for template load in CLAMP
- self._logger.info("Wait a little bit to give a chance to the distribution")
- time.sleep(settings.CLAMP_DISTRIBUTION_TIMER)
- # Test 1
- operational_policies = settings.OPERATIONAL_POLICIES
- loop_template = self.check(operational_policies, True)
- # Test 2
- loop_name = "instance_" + self.service_name + str(self.count)
- self.loop_counter(action="plus")
- self.loop_instance = self.instantiate_clamp(
- loop_template=loop_template,
- loop_name=loop_name,
- operational_policies=operational_policies)
-
- @YamlTemplateBaseStep.store_state(cleanup=True)
- def cleanup(self) -> None:
- """Cleanup Service.
-
- Raises:
- Exception: Clamp cleaning failed
-
- """
- self.loop_counter(action="minus")
- self.loop_instance.undeploy_microservice_from_dcae()
- self.loop_instance.delete()
- super().cleanup()
+++ /dev/null
-#!/usr/bin/python
-# http://www.apache.org/licenses/LICENSE-2.0
-"""Instantiation class."""
-import logging
-import logging.config
-
-from onapsdk.clamp.loop_instance import LoopInstance
-from onapsdk.configuration import settings
-
-import onaptests.utils.exceptions as onap_test_exceptions
-
-
-# pylint: disable=protected-access
-class InstantiateLoop():
- """class instantiating a closed loop in clamp."""
-
- def __init__(self, template: str, loop_name: str, operational_policies: list):
- self.template = template
- self.loop_name = loop_name
- self.operational_policies = operational_policies
-
- self._logger: logging.Logger = logging.getLogger("")
- logging.config.dictConfig(settings.LOG_CONFIG)
-
- def add_policies(self, loop: LoopInstance) -> None:
- """add necessary wanted operational policies."""
- for policy in self.operational_policies:
- self._logger.info("******** ADD OPERATIONAL POLICY %s *******", policy["name"])
- added = loop.add_operational_policy(policy_type=policy["policy_type"],
- policy_version=policy["policy_version"])
- if not added:
- self._logger.error("an error occured while adding an operational policy")
- self._logger.info("ADD OPERATION SUCCESSFULY DONE")
-
- def configure_policies(self, loop: LoopInstance) -> None:
- """Configure all policies."""
- self._logger.info("******** UPDATE MICROSERVICE POLICY *******")
- loop._update_loop_details()
- loop.update_microservice_policy()
- self._logger.info("******** UPDATE OPERATIONAL POLICIES CONFIG *******")
- for policy in self.operational_policies:
- # loop.add_op_policy_config(loop.LoopInstance.__dict__[policy["config_function"]])
- # possible configurations for the moment
- if policy["name"] == "MinMax":
- loop.add_op_policy_config(loop.add_minmax_config)
- if policy["name"] == "Drools":
- loop.add_op_policy_config(loop.add_drools_conf)
- if policy["name"] == "FrequencyLimiter":
- loop.add_op_policy_config(loop.add_frequency_limiter)
- self._logger.info("Policies are well configured")
-
- def deploy(self, loop: LoopInstance) -> None:
- """Deploy closed loop."""
- self._logger.info("******** SUBMIT POLICIES TO PE *******")
- submit = loop.act_on_loop_policy(loop.submit)
- self._logger.info("******** CHECK POLICIES SUBMITION *******")
- if submit:
- self._logger.info("Policies successfully submited to PE")
-
- else:
- self._logger.error("An error occured while submitting the loop instance")
- raise onap_test_exceptions.PolicyException
- self._logger.info("******** DEPLOY LOOP INSTANCE *******")
- deploy = loop.deploy_microservice_to_dcae()
- if deploy:
- self._logger.info("Loop instance %s successfully deployed on DCAE !!", self.loop_name)
- else:
- self._logger.error("An error occured while deploying the loop instance")
- raise onap_test_exceptions.DcaeException
-
- def instantiate_loop(self):
- """Instantiate the control loop."""
- loop = LoopInstance(template=self.template,
- name=self.loop_name,
- details={})
- loop.create()
- if loop.details:
- self._logger.info("Loop instance %s successfully created !!", self.loop_name)
- else:
- self._logger.error("An error occured while creating the loop instance")
-
- self.add_policies(loop=loop)
- self.configure_policies(loop=loop)
- self.deploy(loop=loop)
-
- loop.details = loop._update_loop_details()
- return loop
+++ /dev/null
-#!/usr/bin/python
-# http://www.apache.org/licenses/LICENSE-2.0
-"""Clamp Onboard service class."""
-from onapsdk.configuration import settings
-from onapsdk.sdc.service import Service
-from onapsdk.sdc.vf import Vf
-from yaml import SafeLoader, load
-
-from ..base import BaseStep, YamlTemplateBaseStep
-from .service import YamlTemplateVfOnboardStep
-
-
-class OnboardClampStep(YamlTemplateBaseStep):
- """Onboard class to create CLAMP templates."""
-
- def __init__(self):
- """Initialize Clamp Onboard object."""
- super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
- self._yaml_template: dict = None
- self.add_step(YamlTemplateVfOnboardStep())
- # if "service_name" in kwargs:
- # self.service_name = kwargs['service_name']
- # else:
- # raise ValueError("Service Name to define")
- # self.vf_list = []
- # self.vsp_list = []
- # self.set_logger()
-
- @property
- def description(self) -> str:
- """Step description."""
- return "Onboard service in SDC including a TCA blueprint for CLAMP."
-
- @property
- def component(self) -> str:
- """Component name."""
- return "SDC"
-
- @property
- def yaml_template(self) -> dict:
- """Step YAML template.
-
- Load from file if it's a root step, get from parent otherwise.
-
- Returns:
- dict: Step YAML template
-
- """
- if self.is_root:
- 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
- return self.parent.yaml_template
-
- @property
- def model_yaml_template(self) -> dict:
- return {}
-
- @YamlTemplateBaseStep.store_state
- def execute(self):
- """Onboard service."""
- super().execute()
- # retrieve the Vf
- vf = None
- for sdc_vf in Vf.get_all():
- if sdc_vf.name == settings.VF_NAME:
- vf = sdc_vf
- self._logger.debug("Vf retrieved %s", vf)
-
- service: Service = Service(name=self.service_name,
- resources=[vf])
- service.create()
- self._logger.info(" Service %s created", service)
-
- if not service.distributed:
- service.add_resource(vf)
-
- # we add the artifact to the first VNF
- self._logger.info("Try to add blueprint to %s", vf.name)
- with open(settings.CONFIGURATION_PATH + 'tca-microservice.yaml',
- 'rb') as payload_file:
- data = payload_file.read()
- self._logger.info("DCAE INVENTORY BLUEPRINT file retrieved")
- service.add_artifact_to_vf(vnf_name=vf.name,
- artifact_type="DCAE_INVENTORY_BLUEPRINT",
- artifact_name="tca-microservice.yaml",
- artifact=data)
- service.checkin()
- service.onboard()
- self._logger.info("DCAE INVENTORY BLUEPRINT ADDED")
"""PNF onboarding step module."""
+import time
+from pathlib import Path
+
from onapsdk.configuration import settings
-from onapsdk.sdc.pnf import Pnf
+from onapsdk.exceptions import ResourceNotFound
+from onapsdk.sdc2.pnf import Pnf
+from onapsdk.sdc2.sdc_resource import LifecycleOperation, LifecycleState
from onapsdk.sdc.vendor import Vendor
from onapsdk.sdc.vsp import Vsp
+from onaptests.utils.resources import get_resource_location
from ..base import BaseStep, YamlTemplateBaseStep
from .vsp import VspOnboardStep, YamlTemplateVspOnboardStep
"""
super().execute()
- vendor: Vendor = Vendor(name=settings.VENDOR_NAME)
- pnf: Pnf = Pnf(name=settings.PNF_NAME, vendor=vendor)
- if not pnf.created():
- pnf.create()
+ try:
+ pnf: Pnf = Pnf.get_by_name(name=settings.PNF_NAME)
+ if pnf.lifecycle_state == LifecycleState.CERTIFIED:
+ return
+ except ResourceNotFound:
+ vsp: Vsp = Vsp(name=settings.VSP_NAME)
+ pnf = Pnf.create(settings.PNF_NAME, vsp=vsp, vendor=vsp.vendor)
pnf.add_deployment_artifact(
artifact_type=settings.PNF_ARTIFACT_TYPE,
artifact_name=settings.PNF_ARTIFACT_NAME,
artifact_label=settings.PNF_ARTIFACT_LABEL,
- artifact=settings.PNF_ARTIFACT_FILE_PATH
+ artifact_file_path=settings.PNF_ARTIFACT_FILE_PATH
)
- pnf.onboard()
+ pnf.lifecycle_operation(LifecycleOperation.CERTIFY)
@BaseStep.store_state(cleanup=True)
def cleanup(self):
- pnf: Pnf = Pnf(name=settings.PNF_NAME)
- if pnf.exists():
+ try:
+ pnf = Pnf.get_by_name(settings.PNF_NAME)
pnf.archive()
pnf.delete()
+ except ResourceNotFound:
+ self._logger.warning("VF not created")
super().cleanup()
"""Onboard PNFs from YAML template."""
super().execute()
if "pnfs" in self.yaml_template:
- vendor: Vendor = Vendor(name=settings.VENDOR_NAME)
for pnf in self.yaml_template["pnfs"]:
if "heat_files_to_upload" in pnf:
vsp: Vsp = Vsp(name=f"{pnf['pnf_name']}_VSP")
else:
vsp = None
- pnf_obj: Pnf = Pnf(name=pnf["pnf_name"], vendor=vendor, vsp=vsp)
- if not pnf_obj.created():
- pnf_obj.create()
- pnf_obj.add_deployment_artifact(
- artifact_type=pnf["pnf_artifact_type"],
- artifact_name=pnf["pnf_artifact_name"],
- artifact_label=pnf["pnf_artifact_label"],
- artifact=pnf["pnf_artifact_file_path"]
- )
- pnf_obj.onboard()
+ try:
+ pnf_obj: Pnf = Pnf.get_by_name(name=pnf["pnf_name"])
+ if pnf_obj.lifecycle_state == LifecycleState.CERTIFIED:
+ self._logger.info("PNF already created")
+ return
+ except ResourceNotFound:
+ pnf_obj: Pnf = Pnf.create(name=pnf["pnf_name"],
+ vsp=vsp,
+ vendor=Vendor(name=pnf["pnf_name"]))
+ if all(x in pnf for x in ["pnf_artifact_type",
+ "pnf_artifact_name",
+ "pnf_artifact_label",
+ "pnf_artifact_file_path"]):
+ artifact_file_path: Path = Path(pnf["pnf_artifact_file_path"])
+ if not artifact_file_path.exists():
+ artifact_file_path = Path(get_resource_location(artifact_file_path))
+ pnf_obj.add_deployment_artifact(
+ artifact_type=pnf["pnf_artifact_type"],
+ artifact_name=pnf["pnf_artifact_name"],
+ artifact_label=pnf["pnf_artifact_label"],
+ artifact_file_path=str(artifact_file_path)
+ )
+ time.sleep(10)
+ pnf_obj.lifecycle_operation(LifecycleOperation.CERTIFY)
@YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self):
if "pnfs" in self.yaml_template:
for pnf in self.yaml_template["pnfs"]:
- pnf_obj: Pnf = Pnf(name=pnf["pnf_name"])
- if pnf_obj.exists():
+ try:
+ pnf_obj: Pnf = Pnf.get_by_name(name=pnf["pnf_name"])
pnf_obj.archive()
pnf_obj.delete()
+ except ResourceNotFound:
+ self._logger.warning(f"PNF {pnf['pnf_name']} does not exist")
super().cleanup()
-import time
from typing import Any, Dict
+from yaml import SafeLoader, load
-import onapsdk.constants as onapsdk_const
from onapsdk.configuration import settings
-from onapsdk.exceptions import APIError, ResourceNotFound
-from onapsdk.sdc.component import Component
-from onapsdk.sdc.pnf import Pnf
-from onapsdk.sdc.properties import ComponentProperty
-from onapsdk.sdc.service import Service, ServiceInstantiationType
-from onapsdk.sdc.vf import Vf
-from onapsdk.sdc.vl import Vl
-from yaml import SafeLoader, load
+from onapsdk.exceptions import ResourceNotFound
+from onapsdk.sdc2.pnf import Pnf
+from onapsdk.sdc2.component_instance import ComponentInstance, ComponentInstanceInput
+from onapsdk.sdc2.sdc_resource import LifecycleOperation, LifecycleState
+from onapsdk.sdc2.service import Service, ServiceInstantiationType
+from onapsdk.sdc2.vf import Vf
+from onapsdk.sdc2.vl import Vl
from ..base import BaseStep, YamlTemplateBaseStep
from .pnf import PnfOnboardStep, YamlTemplatePnfOnboardStep
"""
super().execute()
- service: Service = Service(name=settings.SERVICE_NAME,
- instantiation_type=settings.SERVICE_INSTANTIATION_TYPE)
- if not service.created():
- service.create()
+ 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.PNF_NAME != "":
pnf: Pnf = Pnf(name=settings.PNF_NAME)
service.add_resource(pnf)
- # If the service is already distributed, do not try to checkin/onboard (replay of tests)
- # checkin is done if needed
- # If service is replayed, no need to try to re-onboard the model
- # Double check because of:
- # https://gitlab.com/Orange-OpenSource/lfn/onap/python-onapsdk/-/issues/176
- if not service.distributed and service.status != onapsdk_const.DISTRIBUTED:
- if service.status == onapsdk_const.DRAFT:
- try:
- service.checkin()
- except (APIError, ResourceNotFound):
- # Retry as checkin may be a bit long
- # Temp workaround to avoid internal race in SDC
- time.sleep(10)
- service.checkin()
- service.onboard()
+ if service.lifecycle_state != LifecycleState.CERTIFIED:
+ service.lifecycle_operation(LifecycleOperation.CERTIFY)
+ service.distribute()
@BaseStep.store_state
def cleanup(self) -> None:
"""Cleanup service onboard step."""
- service: Service = Service(name=settings.SERVICE_NAME)
- if service.exists():
+ try:
+ service: Service = Service.get_by_name(name=settings.SERVICE_NAME)
service.archive()
service.delete()
+ except ResourceNotFound:
+ self._logger.info(f"Service {settings.SERVICE_NAME} not found")
super().cleanup()
self.yaml_template[self.service_name]["instantiation_type"])
else:
instantiation_type: ServiceInstantiationType = ServiceInstantiationType.A_LA_CARTE
- service: Service = Service(name=self.service_name, instantiation_type=instantiation_type)
- if not service.created():
- service.create()
+ try:
+ service: Service = Service.get_by_name(name=self.service_name)
+ if service.distributed:
+ return
+ except ResourceNotFound:
+ service = Service.create(name=self.service_name, instantiation_type=instantiation_type)
self.declare_resources(service)
self.assign_properties(service)
- # If the service is already distributed, do not try to checkin/onboard (replay of tests)
- # checkin is done if needed
- # If service is replayed, no need to try to re-onboard the model
- # Double check because of:
- # https://gitlab.com/Orange-OpenSource/lfn/onap/python-onapsdk/-/issues/176
- if not service.distributed and service.status != onapsdk_const.DISTRIBUTED:
- if service.status == onapsdk_const.DRAFT:
- try:
- service.checkin()
- except (APIError, ResourceNotFound):
- # Retry as checkin may be a bit long
- # Temp workaround to avoid internal race in SDC
- time.sleep(10)
- service.checkin()
- service.onboard()
+ if service.lifecycle_state != LifecycleState.CERTIFIED:
+ service.lifecycle_operation(LifecycleOperation.CERTIFY)
+ service.distribute()
def declare_resources(self, service: Service) -> None:
"""Declare resources.
"""
if "networks" in self.yaml_template[self.service_name]:
for net in self.yaml_template[self.service_name]["networks"]:
- vl: Vl = Vl(name=net['vl_name'])
+ vl: Vl = Vl.get_by_name(name=net['vl_name'])
service.add_resource(vl)
if "vnfs" in self.yaml_template[self.service_name]:
for vnf in self.yaml_template[self.service_name]["vnfs"]:
- vf: Vf = Vf(name=vnf["vnf_name"])
+ vf: Vf = Vf.get_by_name(name=vnf["vnf_name"])
service.add_resource(vf)
if "pnfs" in self.yaml_template[self.service_name]:
for pnf in self.yaml_template[self.service_name]["pnfs"]:
- pnf_obj: Pnf = Pnf(name=pnf["pnf_name"])
+ pnf_obj: Pnf = Pnf.get_by_name(name=pnf["pnf_name"])
service.add_resource(pnf_obj)
def assign_properties(self, service: Service) -> None:
if "networks" in self.yaml_template[self.service_name]:
for net in self.yaml_template[self.service_name]["networks"]:
if "properties" in net:
- vl: Vl = Vl(name=net['vl_name'])
- vl_component: Component = service.get_component(vl)
+ vl_component: ComponentInstance = service.get_component_by_name(net['vl_name'])
self.assign_properties_to_component(vl_component, net["properties"])
if "vnfs" in self.yaml_template[self.service_name]:
for vnf in self.yaml_template[self.service_name]["vnfs"]:
if "properties" in vnf:
- vf: Vf = Vf(name=vnf["vnf_name"])
- vf_component: Component = service.get_component(vf)
+ vf_component: ComponentInstance = service.get_component_by_name(vnf["vnf_name"])
self.assign_properties_to_component(vf_component, vnf["properties"])
if "pnfs" in self.yaml_template[self.service_name]:
for pnf in self.yaml_template[self.service_name]["pnfs"]:
if "properties" in pnf:
- pnf_obj: Pnf = Pnf(name=pnf["pnf_name"])
- pnf_component: Component = service.get_component(pnf_obj)
+ pnf_component: ComponentInstance = \
+ service.get_component_by_name(pnf["pnf_name"])
self.assign_properties_to_component(pnf_component, pnf["properties"])
def assign_properties_to_component(self,
- component: Component,
+ component: ComponentInstance,
component_properties: Dict[str, Any]) -> None:
"""Assign properties to component.
"""
for property_name, property_value in component_properties.items():
- prop: ComponentProperty = component.get_property(property_name)
+ prop: ComponentInstanceInput = component.get_input_by_name(property_name)
prop.value = property_value
@YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Cleanup service onboard step."""
- service: Service = Service(name=self.service_name)
- if service.exists():
+ try:
+ service: Service = Service.get_by_name(name=self.service_name)
service.archive()
service.delete()
+ except ResourceNotFound:
+ self._logger.info(f"Service {self.service_name} not found")
super().cleanup()
from onapsdk.configuration import settings
from onapsdk.sdc.vendor import Vendor
-from ..base import BaseStep
+from ..base import BaseStep, YamlTemplateBaseStep
class VendorOnboardStep(BaseStep):
vendor.archive()
vendor.delete()
super().cleanup()
+
+
+class YamlTemplateVendorOnboardStep(YamlTemplateBaseStep):
+ """Vendor onboard using YAML template step."""
+
+ def __init__(self):
+ """Initialize step. """
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Onboard vendor described in YAML file 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
+
+ @property
+ def yaml_template(self) -> dict:
+ """YAML template.
+
+ Get YAML template from parent.
+
+ Returns:
+ dict: YAML template
+
+ """
+ if settings.MODEL_YAML_TEMPLATE:
+ return self.model_yaml_template
+ return self.parent.yaml_template
+
+ @property
+ def model_yaml_template(self) -> dict:
+ """Model YAML template.
+
+ Get model YAML template from parent.
+
+ Returns:
+ dict: YAML template
+
+ """
+ return self.parent.model_yaml_template
+
+ @YamlTemplateBaseStep.store_state
+ def execute(self):
+ """Onboard Vsps from YAML template.
+
+ Use settings values:
+ - VENDOR_NAME.
+ """
+ super().execute()
+ if "vnfs" in self.yaml_template:
+ for vnf in self.yaml_template["vnfs"]:
+ vendor: Vendor = Vendor(name=f"{vnf['vnf_name']}")
+ vendor.onboard()
+ elif "pnfs" in self.yaml_template:
+ for pnf in self.yaml_template["pnfs"]:
+ vendor: Vendor = Vendor(name=f"{pnf['pnf_name']}")
+ vendor.onboard()
+
+ @YamlTemplateBaseStep.store_state(cleanup=True)
+ def cleanup(self) -> None:
+ if "vnfs" in self.yaml_template:
+ for vnf in self.yaml_template["vnfs"]:
+ vendor: Vendor = Vendor(name=f"{vnf['vnf_name']}")
+ if vendor.exists():
+ vendor.archive()
+ vendor.delete()
+ elif "pnfs" in self.yaml_template:
+ for pnf in self.yaml_template["pnfs"]:
+ vendor: Vendor = Vendor(name=f"{pnf['pnf_name']}")
+ if vendor.exists():
+ vendor.archive()
+ vendor.delete()
+ super().cleanup()
from pathlib import Path
from onapsdk.configuration import settings
-from onapsdk.sdc.vf import Vf
+from onapsdk.exceptions import ResourceNotFound
+from onapsdk.sdc2.vf import Vf
+from onapsdk.sdc2.sdc_resource import LifecycleOperation, LifecycleState
from onapsdk.sdc.vsp import Vsp
from onaptests.utils.resources import get_resource_location
"""
super().execute()
- vsp: Vsp = Vsp(name=settings.VSP_NAME)
- vf: Vf = Vf(name=settings.VF_NAME, vsp=vsp)
- if not vf.created():
- vf.onboard()
+ try:
+ vf: Vf = Vf.get_by_name(name=settings.VF_NAME)
+ if vf.lifecycle_state == LifecycleState.CERTIFIED:
+ return
+ except ResourceNotFound:
+ vsp: Vsp = Vsp(name=settings.VSP_NAME)
+ vf = Vf.create(settings.VF_NAME, vsp=vsp)
+ vf.lifecycle_operation(LifecycleOperation.CERTIFY)
@BaseStep.store_state(cleanup=True)
def cleanup(self):
- vf: Vf = Vf(name=settings.VF_NAME)
- if vf.exists():
+ try:
+ vf = Vf.get_by_name(settings.VF_NAME)
vf.archive()
vf.delete()
+ except ResourceNotFound:
+ self._logger.warning("VF not created")
super().cleanup()
if "vnfs" in self.yaml_template:
for vnf in self.yaml_template["vnfs"]:
vsp: Vsp = Vsp(name=f"{vnf['vnf_name']}_VSP")
- vf: Vf = Vf(name=vnf['vnf_name'], vsp=vsp)
- if not vf.created():
+ try:
+ vf: Vf = Vf.get_by_name(name=vnf['vnf_name'])
+ if vf.lifecycle_state == LifecycleState.CERTIFIED:
+ self._logger.info("VF already certified")
+ return
+ except ResourceNotFound:
+ vf: Vf = Vf.create(name=vnf['vnf_name'], vsp=vsp, vendor=vsp.vendor)
if all(x in vnf for x in ["vnf_artifact_type",
"vnf_artifact_name",
"vnf_artifact_label",
"vnf_artifact_file_path"]):
- vf.create()
artifact_file_path: Path = Path(vnf["vnf_artifact_file_path"])
if not artifact_file_path.exists():
artifact_file_path = Path(get_resource_location(artifact_file_path))
artifact_type=vnf["vnf_artifact_type"],
artifact_name=vnf["vnf_artifact_name"],
artifact_label=vnf["vnf_artifact_label"],
- artifact=str(artifact_file_path)
+ artifact_file_path=str(artifact_file_path)
)
- time.sleep(10)
- vf.onboard()
+ time.sleep(10)
+ vf.lifecycle_operation(LifecycleOperation.CERTIFY)
@YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self):
if "vnfs" in self.yaml_template:
for vnf in self.yaml_template["vnfs"]:
- vf_obj: Vf = Vf(name=vnf["vnf_name"])
- if vf_obj.exists():
+ try:
+ vf_obj: Vf = Vf.get_by_name(name=vnf["vnf_name"])
vf_obj.archive()
vf_obj.delete()
+ except ResourceNotFound:
+ self._logger.warning(f"VF {vnf['vnf_name']} does not exist")
super().cleanup()
from onaptests.utils.resources import get_resource_location
from ..base import BaseStep, YamlTemplateBaseStep
-from .vendor import VendorOnboardStep
+from .vendor import VendorOnboardStep, YamlTemplateVendorOnboardStep
class VspOnboardStep(BaseStep):
- VendorOnboardStep.
"""
super().__init__(cleanup=settings.CLEANUP_FLAG)
- self.add_step(VendorOnboardStep())
+ self.add_step(YamlTemplateVendorOnboardStep())
@property
def description(self) -> str:
- VENDOR_NAME.
"""
super().execute()
- vendor: Vendor = Vendor(name=settings.VENDOR_NAME)
if "vnfs" in self.yaml_template:
for vnf in self.yaml_template["vnfs"]:
with open(get_resource_location(vnf["heat_files_to_upload"]), "rb") as package:
vsp: Vsp = Vsp(name=f"{vnf['vnf_name']}_VSP",
- vendor=vendor,
+ vendor=Vendor(name=f"{vnf['vnf_name']}"),
package=package)
vsp.onboard()
elif "pnfs" in self.yaml_template:
if "heat_files_to_upload" in pnf:
with open(get_resource_location(pnf["heat_files_to_upload"]), "rb") as package:
vsp: Vsp = Vsp(name=f"{pnf['pnf_name']}_VSP",
- vendor=vendor,
+ vendor=Vendor(name=f"{pnf['pnf_name']}"),
package=package)
vsp.onboard()
---
-basic_cnf_macro:
+{{ service_name }}:
tosca_file_from_SDC: service-basic_cnf_macro-template
version: "1.0"
subscription_type: "basic_cnf_macro"
vnfs:
- - vnf_name: basic_cnf_macro
+ - vnf_name: {{ service_name }}
properties:
controller_actor: "CDS"
skip_post_instantiation_configuration: False
---
-basic_vm:
+{{ service_name }}:
vnfs:
- - vnf_name: basic_vm
+ - vnf_name: {{ service_name }}
heat_files_to_upload: templates/heat-files/ubuntu20/ubuntu20agent.zip
vnf_parameters: [
{"name": "ubuntu20_image_name",
---
-basic_vm_macro:
+{{ service_name }}:
instantiation_type: "Macro"
vnfs:
- - vnf_name: basic_vm_macro
+ - vnf_name: {{ service_name }}
properties:
controller_actor: "CDS"
skip_post_instantiation_configuration: False
---
-test_pnf_macro:
+{{ service_name }}:
tosca_file_from_SDC: service-basic_network-template
version: "1.0"
subscription_type: "net"
instantiation_type: "Macro"
pnfs:
- - pnf_name: "test-pnf"
+ - pnf_name: "{{ service_name }}"
heat_files_to_upload: templates/artifacts/pNF.csar
pnf_artifact_type: "CONTROLLER_BLUEPRINT_ARCHIVE"
pnf_artifact_name: "CBA_enriched.zip"
[tox]
minversion = 3.2.0
-envlist = json,yaml,md,pylama
+envlist = json,yaml,md,pylama,validate
skipsdist = true
requires = pip >= 8
Code Review / testsuite / pythonsdk-tests.git / commitdiff