diff options
Diffstat (limited to 'src/onaptests/steps')
19 files changed, 924 insertions, 0 deletions
diff --git a/src/onaptests/steps/__init__.py b/src/onaptests/steps/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/src/onaptests/steps/__init__.py diff --git a/src/onaptests/steps/base.py b/src/onaptests/steps/base.py new file mode 100644 index 0000000..b32e6d3 --- /dev/null +++ b/src/onaptests/steps/base.py @@ -0,0 +1,84 @@ +from abc import ABC, abstractmethod +from typing import List + + +class BaseStep(ABC): + """Base step class.""" + + def __init__(self, cleanup: bool = False) -> None: + """Step initialization. + + Args: + cleanup(bool, optional): Determines if cleanup action should be called. + + """ + self._steps: List["BaseStep"] = [] + self._cleanup: bool = cleanup + self._parent: "BaseStep" = None + + def add_step(self, step: "BaseStep") -> None: + """Add substep. + + Add substep and mark step as a substep parent. + + Args: + step (BaseStep): Step object + """ + self._steps.append(step) + step._parent: "BaseStep" = self + + @property + def parent(self) -> "BaseStep": + """Step parent. + + If parent is not set the step is a root one. + """ + return self._parent + + @property + def is_root(self) -> bool: + """Is a root step. + + Step is a root if has no parent + + Returns: + bool: True if step is a root step, False otherwise + + """ + return self._parent is None + + def execute(self) -> None: + """Step's action. + + Run all substeps action before it's own action. + Override this method and remember to call `super().action()` before. + + """ + for step in self._steps: + step.execute() + + def cleanup(self) -> None: + """Step's cleanup. + + Not all steps has to have cleanup method + + """ + if self._cleanup: + for step in self._steps: + step.cleanup() + + +class YamlTemplateBaseStep(BaseStep, ABC): + """Base YAML template step.""" + + @property + @abstractmethod + def yaml_template(self) -> dict: + """YAML template abstract property. + + Every YAML template step need to implement that property. + + Returns: + dict: YAML template + + """ diff --git a/src/onaptests/steps/cloud/__init__.py b/src/onaptests/steps/cloud/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/src/onaptests/steps/cloud/__init__.py diff --git a/src/onaptests/steps/cloud/cloud_region_create.py b/src/onaptests/steps/cloud/cloud_region_create.py new file mode 100644 index 0000000..dafd6ec --- /dev/null +++ b/src/onaptests/steps/cloud/cloud_region_create.py @@ -0,0 +1,28 @@ +from onapsdk.aai.cloud_infrastructure import CloudRegion +from onapsdk.configuration import settings + +from ..base import BaseStep + + +class CloudRegionCreateStep(BaseStep): + """Cloud region creation step.""" + + def execute(self): + """Create cloud region. + + Use settings values: + - CLOUD_REGION_CLOUD_OWNER, + - CLOUD_REGION_ID, + - CLOUD_REGION_TYPE, + - CLOUD_REGION_VERSION. + + """ + super().execute() + CloudRegion.create( + cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER, + cloud_region_id=settings.CLOUD_REGION_ID, + orchestration_disabled=False, + in_maint=False, + cloud_type=settings.CLOUD_REGION_TYPE, + cloud_region_version=settings.CLOUD_REGION_VERSION + ) diff --git a/src/onaptests/steps/cloud/complex_create.py b/src/onaptests/steps/cloud/complex_create.py new file mode 100644 index 0000000..37ad49f --- /dev/null +++ b/src/onaptests/steps/cloud/complex_create.py @@ -0,0 +1,23 @@ +from onapsdk.aai.cloud_infrastructure import Complex +from onapsdk.configuration import settings + +from ..base import BaseStep + + +class ComplexCreateStep(BaseStep): + """Complex creation step.""" + + def execute(self): + """Create complex. + + Use settings values: + - COMPLEX_PHYSICAL_LOCATION_ID, + - COMPLEX_DATA_CENTER_CODE. + + """ + super().execute() + Complex.create( + physical_location_id=settings.COMPLEX_PHYSICAL_LOCATION_ID, + data_center_code=settings.COMPLEX_DATA_CENTER_CODE, + name=settings.COMPLEX_PHYSICAL_LOCATION_ID + ) diff --git a/src/onaptests/steps/cloud/connect_service_subscription_to_cloud_region.py b/src/onaptests/steps/cloud/connect_service_subscription_to_cloud_region.py new file mode 100644 index 0000000..23ba994 --- /dev/null +++ b/src/onaptests/steps/cloud/connect_service_subscription_to_cloud_region.py @@ -0,0 +1,46 @@ +from onapsdk.aai.business import Customer, ServiceSubscription +from onapsdk.aai.cloud_infrastructure import CloudRegion, Tenant +from onapsdk.configuration import settings + +from ..base import BaseStep +from .customer_service_subscription_create import CustomerServiceSubscriptionCreateStep +from .link_cloud_to_complex import LinkCloudRegionToComplexStep +from .register_cloud_to_multicloud import RegisterCloudRegionToMulticloudStep + + +class ConnectServiceSubToCloudRegionStep(BaseStep): + """Connect service subscription to cloud region step.""" + + def __init__(self, cleanup=False): + """Initialize step. + + Substeps: + - LinkCloudRegionToComplexStep, + - RegisterCloudRegionToMulticloudStep, + - CustomerServiceSubscriptionCreateStep. + + """ + super().__init__(cleanup=cleanup) + self.add_step(LinkCloudRegionToComplexStep(cleanup=cleanup)) + self.add_step(RegisterCloudRegionToMulticloudStep(cleanup=cleanup)) + self.add_step(CustomerServiceSubscriptionCreateStep(cleanup=cleanup)) + + def execute(self): + """Connect service subsription to cloud region and tenant. + + Use settings values: + - GLOBAL_CUSTOMER_ID, + - SERVICE_NAME, + - CLOUD_REGION_CLOUD_OWNER, + - CLOUD_REGION_ID. + + """ + super().execute() + customer: Customer = Customer.get_by_global_customer_id(settings.GLOBAL_CUSTOMER_ID) + service_subscription: ServiceSubscription = customer.get_service_subscription_by_service_type(settings.SERVICE_NAME) + cloud_region: CloudRegion = CloudRegion.get_by_id( + cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER, + cloud_region_id=settings.CLOUD_REGION_ID, + ) + tenant: Tenant = cloud_region.get_tenant(settings.TENANT_ID) + service_subscription.link_to_cloud_region_and_tenant(cloud_region=cloud_region, tenant=tenant) diff --git a/src/onaptests/steps/cloud/customer_create.py b/src/onaptests/steps/cloud/customer_create.py new file mode 100644 index 0000000..2bdf312 --- /dev/null +++ b/src/onaptests/steps/cloud/customer_create.py @@ -0,0 +1,16 @@ +from onapsdk.aai.business import Customer +from onapsdk.configuration import settings + +from ..base import BaseStep + + +class CustomerCreateStep(BaseStep): + """Customer creation step.""" + + def execute(self): + """Create cutomer. + + Use settings values: + - GLOBAL_CUSTOMER_ID. + """ + Customer.create(settings.GLOBAL_CUSTOMER_ID, settings.GLOBAL_CUSTOMER_ID, "INFRA") diff --git a/src/onaptests/steps/cloud/customer_service_subscription_create.py b/src/onaptests/steps/cloud/customer_service_subscription_create.py new file mode 100644 index 0000000..c797d56 --- /dev/null +++ b/src/onaptests/steps/cloud/customer_service_subscription_create.py @@ -0,0 +1,30 @@ +from onapsdk.aai.business import Customer +from onapsdk.sdc.service import Service +from onapsdk.configuration import settings + +from ..base import BaseStep +from .customer_create import CustomerCreateStep + + +class CustomerServiceSubscriptionCreateStep(BaseStep): + """Cutomer service subsription creation step""" + + def __init__(self, cleanup=False): + """Initialize step. + + Substeps: + - CustomerCreateStep. + """ + super().__init__(cleanup=cleanup) + self.add_step(CustomerCreateStep(cleanup=cleanup)) + + def execute(self): + """Create customer service subsription. + + Use settings values: + - GLOBAL_CUSTOMER_ID, + - SERVICE_NAME. + """ + service = Service(name=settings.SERVICE_NAME) + customer = Customer.get_by_global_customer_id(settings.GLOBAL_CUSTOMER_ID) + customer.subscribe_service(service) diff --git a/src/onaptests/steps/cloud/link_cloud_to_complex.py b/src/onaptests/steps/cloud/link_cloud_to_complex.py new file mode 100644 index 0000000..4da3804 --- /dev/null +++ b/src/onaptests/steps/cloud/link_cloud_to_complex.py @@ -0,0 +1,40 @@ +from onapsdk.aai.cloud_infrastructure import CloudRegion, Complex +from onapsdk.configuration import settings + +from ..base import BaseStep +from .cloud_region_create import CloudRegionCreateStep +from .complex_create import ComplexCreateStep + + +class LinkCloudRegionToComplexStep(BaseStep): + """Link cloud region to complex step""" + + def __init__(self, cleanup=False): + """Initialize step. + + Substeps: + - ComplexCreateStep, + - CloudRegionCreateStep. + """ + super().__init__(cleanup=cleanup) + self.add_step(ComplexCreateStep(cleanup=cleanup)) + self.add_step(CloudRegionCreateStep(cleanup=cleanup)) + + def execute(self): + """Link cloud region to complex. + + Use settings values: + - COMPLEX_PHYSICAL_LOCATION_ID, + - CLOUD_REGION_CLOUD_OWNER, + - CLOUD_REGION_ID. + """ + super().execute() + cmplx = Complex( + physical_location_id=settings.COMPLEX_PHYSICAL_LOCATION_ID, + name=settings.COMPLEX_PHYSICAL_LOCATION_ID + ) + cloud_region = CloudRegion.get_by_id( + cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER, + cloud_region_id=settings.CLOUD_REGION_ID, + ) + cloud_region.link_to_complex(cmplx) diff --git a/src/onaptests/steps/cloud/register_cloud_to_multicloud.py b/src/onaptests/steps/cloud/register_cloud_to_multicloud.py new file mode 100644 index 0000000..c3a001c --- /dev/null +++ b/src/onaptests/steps/cloud/register_cloud_to_multicloud.py @@ -0,0 +1,47 @@ +from uuid import uuid4 + +from onapsdk.aai.cloud_infrastructure import CloudRegion +from onapsdk.configuration import settings + +from ..base import BaseStep +from .cloud_region_create import CloudRegionCreateStep + + +class RegisterCloudRegionToMulticloudStep(BaseStep): + """Cloud region registration in multicloud step.""" + + def __init__(self, cleanup=False): + """Initialize step. + + Substeps: + - CloudRegionCreateStep. + """ + super().__init__(cleanup=cleanup) + self.add_step(CloudRegionCreateStep(cleanup=cleanup)) + + def execute(self): + """Register cloud region in multicloud. + + Use settings values: + - CLOUD_REGION_CLOUD_OWNER, + - CLOUD_REGION_ID, + - CLOUD_REGION_TYPE, + - CLOUD_DOMAIN, + - VIM_USERNAME, + - VIM_PASSWORD, + - VIM_SERVICE_URL. + """ + super().execute() + cloud_region = CloudRegion.get_by_id( + cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER, + cloud_region_id=settings.CLOUD_REGION_ID, + ) + cloud_region.add_esr_system_info( + esr_system_info_id=str(uuid4()), + user_name=settings.VIM_USERNAME, + password=settings.VIM_PASSWORD, + system_type=settings.CLOUD_REGION_TYPE, + service_url=settings.VIM_SERVICE_URL, + cloud_domain=settings.CLOUD_DOMAIN + ) + cloud_region.register_to_multicloud() diff --git a/src/onaptests/steps/instantiate/__init__.py b/src/onaptests/steps/instantiate/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/src/onaptests/steps/instantiate/__init__.py diff --git a/src/onaptests/steps/instantiate/service_ala_carte.py b/src/onaptests/steps/instantiate/service_ala_carte.py new file mode 100644 index 0000000..dad6563 --- /dev/null +++ b/src/onaptests/steps/instantiate/service_ala_carte.py @@ -0,0 +1,168 @@ +import time +from uuid import uuid4 +from yaml import load + +from onapsdk.aai.cloud_infrastructure import CloudRegion, Tenant +from onapsdk.aai.business import Customer +from onapsdk.aai.business.owning_entity import OwningEntity as AaiOwningEntity +from onapsdk.configuration import settings +from onapsdk.sdc.service import Service +from onapsdk.so.instantiation import ServiceInstantiation +from onapsdk.vid import Project + +from ..base import BaseStep, YamlTemplateBaseStep +from ..cloud.connect_service_subscription_to_cloud_region import ConnectServiceSubToCloudRegionStep +from ..onboard.service import ServiceOnboardStep, YamlTemplateServiceOnboardStep + + +class ServiceAlaCarteInstantiateStep(BaseStep): + """Instantiate service a'la carte.""" + + def __init__(self, cleanup=False): + """Initialize step. + + Substeps: + - ServiceOnboardStep, + - ConnectServiceSubToCloudRegionStep. + """ + super().__init__(cleanup=cleanup) + self.add_step(ServiceOnboardStep(cleanup)) + self.add_step(ConnectServiceSubToCloudRegionStep(cleanup)) + + def execute(self): + """Instantiate service. + + Use settings values: + - SERVICE_NAME, + - GLOBAL_CUSTOMER_ID, + - CLOUD_REGION_CLOUD_OWNER, + - CLOUD_REGION_ID, + - TENANT_ID, + - OWNING_ENTITY, + - PROJECT, + - SERVICE_INSTANCE_NAME. + """ + super().execute() + service = Service(settings.SERVICE_NAME) + customer: Customer = Customer.get_by_global_customer_id(settings.GLOBAL_CUSTOMER_ID) + cloud_region: CloudRegion = CloudRegion.get_by_id( + cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER, + cloud_region_id=settings.CLOUD_REGION_ID, + ) + tenant: Tenant = cloud_region.get_tenant(settings.TENANT_ID) + owning_entity = AaiOwningEntity.get_by_owning_entity_name(settings.OWNING_ENTITY) + vid_project = Project.create(settings.PROJECT) + + service_instantiation = ServiceInstantiation.instantiate_so_ala_carte( + service, + cloud_region, + tenant, + customer, + owning_entity, + vid_project, + service_instance_name=settings.SERVICE_INSTANCE_NAME + ) + while not service_instantiation.finished: + time.sleep(10) + + +class YamlTemplateServiceAlaCarteInstantiateStep(YamlTemplateBaseStep): + """Instantiate service a'la carte using YAML template.""" + + def __init__(self, cleanup=False): + """Initialize step. + + Substeps: + - YamlTemplateServiceOnboardStep, + - ConnectServiceSubToCloudRegionStep. + """ + super().__init__(cleanup=cleanup) + self._yaml_template: dict = None + self.add_step(YamlTemplateServiceOnboardStep(cleanup)) + self.add_step(ConnectServiceSubToCloudRegionStep(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: + return f"{self.service_name}-{str(uuid4())}" + return self.parent.service_instance_name + + def execute(self): + """Instantiate service. + + Use settings values: + - GLOBAL_CUSTOMER_ID, + - CLOUD_REGION_CLOUD_OWNER, + - CLOUD_REGION_ID, + - TENANT_ID, + - OWNING_ENTITY, + - PROJECT. + + Raises: + Exception: Service instantiation failed + + """ + super().execute() + service = Service(self.service_name) + customer: Customer = Customer.get_by_global_customer_id(settings.GLOBAL_CUSTOMER_ID) + cloud_region: CloudRegion = CloudRegion.get_by_id( + cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER, + cloud_region_id=settings.CLOUD_REGION_ID, + ) + tenant: Tenant = cloud_region.get_tenant(settings.TENANT_ID) + owning_entity = AaiOwningEntity.get_by_owning_entity_name(settings.OWNING_ENTITY) + vid_project = Project.create(settings.PROJECT) + + service_instantiation = ServiceInstantiation.instantiate_so_ala_carte( + service, + cloud_region, + tenant, + customer, + owning_entity, + vid_project, + service_instance_name=self.service_instance_name + ) + while not service_instantiation.finished: + time.sleep(10) + if service_instantiation.failed: + raise Exception("Service instantiation failed") diff --git a/src/onaptests/steps/instantiate/vf_module_ala_carte.py b/src/onaptests/steps/instantiate/vf_module_ala_carte.py new file mode 100644 index 0000000..fd4b368 --- /dev/null +++ b/src/onaptests/steps/instantiate/vf_module_ala_carte.py @@ -0,0 +1,112 @@ +import time +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 VnfParameter + +from ..base import YamlTemplateBaseStep +from .vnf_ala_carte import YamlTemplateVnfAlaCarteInstantiateStep + + +class YamlTemplateVfModuleAlaCarteInstantiateStep(YamlTemplateBaseStep): + """Instantiate vf module a'la carte using YAML template.""" + + def __init__(self, cleanup=False): + """Initialize step. + + Substeps: + - YamlTemplateVnfAlaCarteInstantiateStep. + """ + super().__init__(cleanup=cleanup) + self._yaml_template: dict = None + self.add_step(YamlTemplateVnfAlaCarteInstantiateStep(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: + return f"{self.service_name}-{str(uuid4())}" + return self.parent.service_instance_name + + def get_vnf_parameters(self, vnf_name: str) -> Iterable[VnfParameter]: + """Get VNF parameters from YAML template. + + Args: + vnf_name (str): VNF name to get parameters for. + + Yields: + Iterator[Iterable[VnfParameter]]: VNF parameter + + """ + for vnf in self.yaml_template[self.service_name]["vnfs"]: + if vnf["vnf_name"] == vnf_name: + for vnf_parameter in vnf["vnf_parameters"]: + yield VnfParameter( + name=vnf_parameter["name"], + value=vnf_parameter["value"] + ) + + def execute(self) -> None: + """Instantiate Vf module. + + Use settings values: + - GLOBAL_CUSTOMER_ID. + + Raises: + Exception: Vf module instantiation failed + + """ + 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) + service_instance: ServiceInstance = service_subscription.get_service_instance_by_name(self.service_instance_name) + for vnf_instance in service_instance.vnf_instances: + vf_module_instantiation = vnf_instance.add_vf_module( + vnf_instance.vnf.vf_module, + vnf_parameters= self.get_vnf_parameters(vnf_instance.vnf.vnf_name)) + while not vf_module_instantiation.finished: + time.sleep(10) + if vf_module_instantiation.failed: + raise Exception("Vf module instantiation failed") diff --git a/src/onaptests/steps/instantiate/vnf_ala_carte.py b/src/onaptests/steps/instantiate/vnf_ala_carte.py new file mode 100644 index 0000000..116b574 --- /dev/null +++ b/src/onaptests/steps/instantiate/vnf_ala_carte.py @@ -0,0 +1,96 @@ +import time +from uuid import uuid4 +from yaml import load + +from onapsdk.aai.business import Customer, ServiceInstance, ServiceSubscription +from onapsdk.configuration import settings +from onapsdk.sdc.service import Service +from onapsdk.vid import LineOfBusiness, Platform + +from ..base import YamlTemplateBaseStep +from .service_ala_carte import YamlTemplateServiceAlaCarteInstantiateStep + + +class YamlTemplateVnfAlaCarteInstantiateStep(YamlTemplateBaseStep): + """Instantiate vnf 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.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: + return f"{self.service_name}-{str(uuid4())}" + return self.parent.service_instance_name + + def execute(self): + """Instantiate vnf. + + Use settings values: + - GLOBAL_CUSTOMER_ID, + - LINE_OF_BUSINESS. + + Raises: + Exception: VNF 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) + line_of_business: LineOfBusiness = LineOfBusiness(settings.LINE_OF_BUSINESS) + platform: Platform = Platform(settings.PLATFORM) + for idx, vnf in service.vnfs: + vnf_instantiation = service_instance.add_vnf(vnf, line_of_business, platform, f"{self.service_instance_name}_vnf_{idx}") + while not vnf_instantiation.finished: + time.sleep(10) + if vnf_instantiation.failed: + raise Exception("Vnf instantiation failed") diff --git a/src/onaptests/steps/onboard/__init__.py b/src/onaptests/steps/onboard/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/src/onaptests/steps/onboard/__init__.py diff --git a/src/onaptests/steps/onboard/service.py b/src/onaptests/steps/onboard/service.py new file mode 100644 index 0000000..0143908 --- /dev/null +++ b/src/onaptests/steps/onboard/service.py @@ -0,0 +1,88 @@ +from yaml import load + +from onapsdk.configuration import settings +from onapsdk.sdc.service import Service +from onapsdk.sdc.vf import Vf + +from ..base import BaseStep, YamlTemplateBaseStep +from .vf import VfOnboardStep, YamlTemplateVfOnboardStep + + +class ServiceOnboardStep(BaseStep): + """Service onboard step.""" + + def __init__(self, cleanup=False): + """Initialize step. + + Substeps: + - VfOnboardStep. + """ + super().__init__(cleanup=cleanup) + self.add_step(VfOnboardStep(cleanup=cleanup)) + + def execute(self): + """Onboard service. + + Use settings values: + - VF_NAME, + - SERVICE_NAME. + + """ + super().execute() + vf: Vf = Vf(name=settings.VF_NAME) + service: Service = Service(name=settings.SERVICE_NAME, resources=[vf]) + service.onboard() + + +class YamlTemplateServiceOnboardStep(YamlTemplateBaseStep): + """Service onboard using YAML template step.""" + + def __init__(self, cleanup=False): + """Initialize step. + + Substeps: + - YamlTemplateVfOnboardStep. + """ + super().__init__(cleanup=cleanup) + self._yaml_template: dict = None + self.add_step(YamlTemplateVfOnboardStep(cleanup=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())) + else: + return self.parent.service_name + + 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.name]["vnfs"]]) + service.onboard() diff --git a/src/onaptests/steps/onboard/vendor.py b/src/onaptests/steps/onboard/vendor.py new file mode 100644 index 0000000..8a4cb48 --- /dev/null +++ b/src/onaptests/steps/onboard/vendor.py @@ -0,0 +1,19 @@ +from onapsdk.configuration import settings +from onapsdk.sdc.vendor import Vendor + +from ..base import BaseStep + + +class VendorOnboardStep(BaseStep): + """Vendor onboard step.""" + + def execute(self): + """Onboard vendor. + + Use settings values: + - VENDOR_NAME. + + """ + super().execute() + vendor: Vendor = Vendor(name=settings.VENDOR_NAME) + vendor.onboard() diff --git a/src/onaptests/steps/onboard/vf.py b/src/onaptests/steps/onboard/vf.py new file mode 100644 index 0000000..6d19b9a --- /dev/null +++ b/src/onaptests/steps/onboard/vf.py @@ -0,0 +1,65 @@ +from onapsdk.configuration import settings +from onapsdk.sdc.vf import Vf +from onapsdk.sdc.vsp import Vsp + +from ..base import BaseStep, YamlTemplateBaseStep +from .vsp import VspOnboardStep, YamlTemplateVspOnboardStep + + +class VfOnboardStep(BaseStep): + """Vf onboard step.""" + + def __init__(self, cleanup=False): + """Initialize step. + + Substeps: + - VspOnboardStep. + """ + super().__init__(cleanup=cleanup) + self.add_step(VspOnboardStep(cleanup=cleanup)) + + def execute(self): + """Onboard Vf. + + Use settings values: + - VSP_NAME, + - VF_NAME. + + """ + super().execute() + vsp: Vsp = Vsp(name=settings.VSP_NAME) + vf: Vf = Vf(name=settings.VF_NAME, vsp=vsp) + vf.onboard() + + +class YamlTemplateVfOnboardStep(YamlTemplateBaseStep): + """Vf onboard using YAML template step.""" + + def __init__(self, cleanup=False) -> None: + """Initialize step. + + Substeps: + - YamlTemplateVspOnboardStep. + """ + super().__init__(cleanup=cleanup) + self.add_step(YamlTemplateVspOnboardStep(cleanup=cleanup)) + + @property + def yaml_template(self) -> dict: + """YAML template. + + Get YAML template from parent using it's name. + + Returns: + dict: YAML template + + """ + return self.parent.yaml_template[self.parent.name] + + 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() diff --git a/src/onaptests/steps/onboard/vsp.py b/src/onaptests/steps/onboard/vsp.py new file mode 100644 index 0000000..f1eec1f --- /dev/null +++ b/src/onaptests/steps/onboard/vsp.py @@ -0,0 +1,62 @@ +from onapsdk.configuration import settings +from onapsdk.sdc.vendor import Vendor +from onapsdk.sdc.vsp import Vsp + +from ..base import BaseStep, YamlTemplateBaseStep +from .vendor import VendorOnboardStep + + +class VspOnboardStep(BaseStep): + """Vsp onboard step.""" + + def __init__(self, cleanup=False): + """Initialize step. + + Substeps: + - VendorOnboardStep. + """ + super().__init__(cleanup=cleanup) + self.add_step(VendorOnboardStep(cleanup=cleanup)) + + def execute(self): + """Onboard Vsp. + + Use settings values: + - VSP_NAME, + - VSP_FILE_PATH, + - VENDOR_NAME. + + """ + super().execute() + vendor: Vendor = Vendor(name=settings.VENDOR_NAME) + vsp: Vsp = Vsp(name=settings.VSP_NAME, vendor=vendor, package=open(settings.VSP_FILE_PATH, "rb")) + vsp.onboard() + + +class YamlTemplateVspOnboardStep(VspOnboardStep, YamlTemplateBaseStep): + """Vsp onboard using YAML template step.""" + + @property + def yaml_template(self) -> dict: + """YAML template. + + Get YAML template from parent. + + Returns: + dict: YAML template + + """ + return self.parent.yaml_template + + def execute(self): + """Onboard Vsps from YAML template. + + Use settings values: + - VENDOR_NAME. + """ + super().execute() + vendor: Vendor = Vendor(name=settings.VENDOR_NAME) + for vnf in self.yaml_template["vnfs"]: + with open(vnf["heat_files_to_upload"], "rb") as package: + vsp: Vsp = Vsp(name=f"{vnf['vnf_name']}_VSP", vendor=vendor, package=package) + vsp.onboard() |