--- /dev/null
+import logging.config
+import time
+from onapsdk.configuration import settings
+from onaptests.steps.instantiate.vl_ala_carte import YamlTemplateVlAlaCarteInstantiateStep
+
+
+
+if __name__ == "__main__":
+ # logging configuration for onapsdk, it is not requested for onaptests
+ # Correction requested in onapsdk to avoid having this duplicate code
+ logging.config.dictConfig(settings.LOG_CONFIG)
+ logger = logging.getLogger("Instantiate Basic Network without multicloud")
+
+ basic_network_instantiate = YamlTemplateVlAlaCarteInstantiateStep(
+ cleanup=settings.CLEANUP_FLAG)
+ basic_network_instantiate.execute()
+ if settings.CLEANUP_FLAG:
+ time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
+ try:
+ basic_network_instantiate.cleanup()
+ except ValueError as error:
+ logger.info("service instance deleted as expected {0}".format(error))
+ basic_network_instantiate.reports_collection.generate_report()
[entry_points]
xtesting.testcase =
basic_vm = onaptests.scenario.basic_vm:BasicVm
+ basic_network = onaptests.scenario.basic_network:BasicNetwork
clearwater_ims = onaptests.scenario.clearwater_ims:ClearwaterIms
--- /dev/null
+import os
+import openstack
+import sys
+from yaml import load
+
+import onaptests.utils.exceptions as onap_test_exceptions
+from .settings import * # pylint: disable=W0614
+
+""" Specific ubuntu16 without multicloud."""
+
+# pylint: disable=bad-whitespace
+# The ONAP part
+SERVICE_DETAILS="Onboarding, distribution and instanitation of asic Network using à la carte"
+SERVICE_COMPONENTS="SDC, DMAAP, AAI, SO, SDNC"
+USE_MULTICLOUD = False
+# Set ONLY_INSTANTIATE to true to run an instantiation without repeating
+# onboarding and related AAI configuration (Cloud config)
+ONLY_INSTANTIATE= False
+
+# if a yaml file is define, retrieve info from this yaml files
+# if not declare the parameters in the settings
+SERVICE_YAML_TEMPLATE = (sys.path[-1] + "/onaptests/templates/vnf-services/" +
+ "basic_network-service.yaml")
+
+try:
+ # Try to retrieve the SERVICE NAME from the yaml file
+ with open(SERVICE_YAML_TEMPLATE, "r") as yaml_template:
+ yaml_config_file = load(yaml_template)
+ SERVICE_NAME = next(iter(yaml_config_file.keys()))
+except (FileNotFoundError, ValueError):
+ raise onap_test_exceptions.TestConfigurationException
+
+CLEANUP_FLAG = True
+CLEANUP_ACTIVITY_TIMER = 10 # nb of seconds before cleanup in case cleanup option is set
+VENDOR_NAME = "basicnw_vendor"
+
+CLOUD_REGION_CLOUD_OWNER = "basicnw-cloud-owner"
+CLOUD_REGION_TYPE = "openstack"
+CLOUD_REGION_VERSION = "openstack"
+CLOUD_OWNER_DEFINED_TYPE = "N/A"
+
+AVAILABILITY_ZONE_NAME = "basicvm-availability-zone"
+AVAILABILITY_ZONE_TYPE = "nova"
+COMPLEX_PHYSICAL_LOCATION_ID = "lannion"
+COMPLEX_DATA_CENTER_CODE = "1234-5"
+
+GLOBAL_CUSTOMER_ID = "basicnw-customer"
+
+OWNING_ENTITY = "basicnw-oe"
+PROJECT = "basicnw-project"
+LINE_OF_BUSINESS = "basicnw-lob"
+PLATFORM = "basicnw-platform"
+
+SERVICE_INSTANCE_NAME = "basicnw_service_instance"
+
+# The cloud Part
+# Assuming a cloud.yaml is available, use the openstack client
+# to retrieve cloud info and avoid data duplication
+TEST_CLOUD = os.getenv('OS_TEST_CLOUD')
+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')
--- /dev/null
+#!/usr/bin/env python
+"""Basic VM test case."""
+import logging
+import time
+
+from xtesting.core import testcase
+from onapsdk.configuration import settings
+
+import onaptests.utils.exceptions as onap_test_exceptions
+from onaptests.steps.instantiate.vl_ala_carte import YamlTemplateVlAlaCarteInstantiateStep
+
+class BasicNetwork(testcase.TestCase):
+ """Onboard then instantiate a simple Network with ONAP."""
+
+ __logger = logging.getLogger(__name__)
+
+ def __init__(self, **kwargs):
+ """Init Basic Network use case."""
+ # import basic_network_nomulticloud_settings needed
+ if "case_name" not in kwargs:
+ kwargs["case_name"] = 'basic_network'
+ super(BasicNetwork, self).__init__(**kwargs)
+ self.__logger.debug("BasicNetwork init started")
+ self.test = YamlTemplateVlAlaCarteInstantiateStep(
+ cleanup=settings.CLEANUP_FLAG)
+ self.start_time = None
+ self.stop_time = None
+ self.result = 0
+
+ def run(self):
+ """Run onap_tests with basic network."""
+ self.start_time = time.time()
+ self.__logger.debug("start time")
+ try:
+ self.test.execute()
+ self.__logger.info("Service basic_network successfully created")
+ # The cleanup is part of the test, not only a teardown action
+ if settings.CLEANUP_FLAG:
+ self.__logger.info("Service basic_network cleanup called")
+ time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
+ try:
+ self.test.cleanup()
+ except ValueError as error:
+ self.__logger.info("service deleted as expected {0}".format(error))
+ self.result = 100
+ else:
+ self.__logger.info("No cleanup requested. Test completed.")
+ self.result = 100
+ except onap_test_exceptions.ServiceInstantiateException:
+ self.__logger.error("Basic network service instantiation failed.")
+ self.result = 0
+ except onap_test_exceptions.ServiceInstantiateException:
+ self.__logger.error("Basic network service cleanup failed.")
+ self.result = 0
+ except onap_test_exceptions.NetworkInstantiateException:
+ self.__logger.error("Basic network VL instantiation failed.")
+ self.result = 0
+ except onap_test_exceptions.NetworkCleanupException:
+ self.__logger.error("Basic network VL cleanup failed.")
+ self.result = 0
+ finally:
+ self.stop_time = time.time()
+
+ def clean(self):
+ """Clean Additional resources if needed."""
+ self.__logger.info("Generate Test report")
+ self.test.reports_collection.generate_report()
--- /dev/null
+import time
+import re
+from typing import Iterable
+from uuid import uuid4
+from yaml import load
+
+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
+from onapsdk.vid import LineOfBusiness, Platform
+
+import onaptests.utils.exceptions as onap_test_exceptions
+from ..base import YamlTemplateBaseStep
+from .service_ala_carte import YamlTemplateServiceAlaCarteInstantiateStep
+
+
+class YamlTemplateVlAlaCarteInstantiateStep(YamlTemplateBaseStep):
+ """Instantiate vl a'la carte using YAML template."""
+
+ def __init__(self, cleanup=False):
+ """Initialize step.
+
+ Substeps:
+ - YamlTemplateServiceAlaCarteInstantiateStep.
+ """
+ super().__init__(cleanup=cleanup)
+ self._yaml_template: dict = None
+ self._service_instance_name: str = None
+ self._service_instance: ServiceInstance = None
+ self.add_step(YamlTemplateServiceAlaCarteInstantiateStep(cleanup))
+
+ @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") as yaml_template:
+ self._yaml_template: dict = load(yaml_template)
+ return self._yaml_template
+ return self.parent.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.
+
+ Generate using `service_name` and `uuid4()` function if it's a root step,
+ get from parent otherwise.
+
+ Returns:
+ str: Service instance name
+
+ """
+ if self.is_root:
+ if not self._service_instance_name:
+ self._service_instance_name: str = f"{self.service_name}-{str(uuid4())}"
+ return self._service_instance_name
+ return self.parent.service_instance_name
+
+ def get_subnets(self, network_name: str) -> Iterable[Subnet]:
+ """Get Network parameters from YAML template.
+
+ Args:
+ network_name (str): Network name to get parameters for.
+
+ Yields:
+ Iterator[Iterable[Subnet]]: Subnets
+
+ """
+ # workaround, as Network name differs from model name (added " 0")
+ 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:
+ print("No Subnet defined")
+ else:
+ for subnet in net['subnets']:
+ yield Subnet(
+ name=subnet['subnet-name'],
+ start_address=subnet['start-address'],
+ gateway_address=subnet['gateway-address'],
+ cidr_mask=subnet['cidr-mask'],
+ ip_version=subnet['ip-version'],
+ dhcp_enabled=subnet['dhcp-enabled'])
+
+ @YamlTemplateBaseStep.store_state
+ def execute(self) -> None:
+ """Instantiate Vl.
+
+ Use settings values:
+ - GLOBAL_CUSTOMER_ID.
+
+ Raises:
+ Exception: Vl instantiation failed
+
+ """
+ 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
+ line_of_business: LineOfBusiness = LineOfBusiness(settings.LINE_OF_BUSINESS)
+ platform: Platform = Platform(settings.PLATFORM)
+ for idx, network in enumerate(service.networks):
+ #for network in self.yaml_template[self.service_name]["networks"]:
+ net_instantiation = service_instance.add_network(
+ network,
+ line_of_business,
+ platform,
+ network_instance_name=f"{self.service_instance_name}_net_{idx}",
+ subnets=self.get_subnets(network.name))
+ while not net_instantiation.finished:
+ time.sleep(10)
+ if net_instantiation.failed:
+ raise onap_test_exceptions.NetworkInstantiateException
+
+ def cleanup(self) -> None:
+ """Cleanup VL.
+
+ Raises:
+ Exception: VL cleaning failed
+ """
+ if settings.CLEANUP_FLAG:
+ for net_instance in self._service_instance.network_instances:
+ self._logger.info("Start network deletion %s",net_instance.name)
+ net_deletion = net_instance.delete()
+ try:
+ net_deletion.wait_for_finish()
+ except TimeoutError:
+ raise onap_test_exceptions.NetworkCleanupException
+ super().cleanup()
from onapsdk.configuration import settings
from onapsdk.sdc.service import Service
from onapsdk.sdc.vf import Vf
+from onapsdk.sdc.vl import Vl
from ..base import BaseStep, YamlTemplateBaseStep
from .vf import VfOnboardStep, YamlTemplateVfOnboardStep
"""Onboard service.
Use settings values:
+ - VL_NAME,
- VF_NAME,
- SERVICE_NAME.
"""
super().execute()
- vf: Vf = Vf(name=settings.VF_NAME)
- service: Service = Service(name=settings.SERVICE_NAME, resources=[vf])
+ service: Service = Service(name=settings.SERVICE_NAME)
+ service.create()
+ 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)
+ service.checkin()
service.onboard()
def execute(self):
"""Onboard service."""
super().execute()
- service: Service = Service(name=self.service_name,
- resources=[Vf(name=vnf["vnf_name"]) \
- for vnf in self.yaml_template[self.service_name]["vnfs"]])
+ service: Service = Service(name=settings.SERVICE_NAME)
+ service.create()
+ 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'])
+ 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"])
+ service.add_resource(vf)
+ service.checkin()
service.onboard()
def execute(self):
"""Onboard Vfs from YAML template."""
super().execute()
- 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)
- vf.onboard()
+ 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)
+ vf.onboard()
"""
super().execute()
vendor: Vendor = Vendor(name=settings.VENDOR_NAME)
- for vnf in self.yaml_template["vnfs"]:
- with open(
- sys.path[-1] + "/" + vnf["heat_files_to_upload"], "rb") as package:
- vsp: Vsp = Vsp(name=f"{vnf['vnf_name']}_VSP",
- vendor=vendor,
- package=package)
- vsp.onboard()
+ if "vnfs" in self.yaml_template:
+ for vnf in self.yaml_template["vnfs"]:
+ with open(
+ sys.path[-1] + "/" + vnf["heat_files_to_upload"], "rb") as package:
+ vsp: Vsp = Vsp(name=f"{vnf['vnf_name']}_VSP",
+ vendor=vendor,
+ package=package)
+ vsp.onboard()
--- /dev/null
+---
+basic_network:
+ tosca_file_from_SDC: service-basic_network-template
+ version: "1.0"
+ subscription_type: "net"
+ networks:
+ - network_name: net_internal
+ vl_name: "Generic NeutronNet"
+ subnets: [
+ {
+ "subnet-name": "net_internal-subnet",
+ "start-address": "10.200.0.0",
+ "cidr-mask": "24",
+ "ip-version": "4",
+ "dhcp-enabled": False,
+ "gateway-address": "10.200.0.1",
+ }
+ ]
Code Review / testsuite / pythonsdk-tests.git / commitdiff