Use Python SDK exceptions.
Issue-ID: TEST-302
Signed-off-by: Michal Jagiello <michal.jagiello@t-mobile.pl>
Change-Id: I76996817c0d7b91c0214ad521c3768682ff07e83
xtesting
openstacksdk
-onapsdk>=7.1.0
+onapsdk==7.4.0
jinja2
kubernetes
docker
\ No newline at end of file
from xtesting.core import testcase
from onapsdk.configuration import settings
+from onapsdk.exceptions import SDKException
import onaptests.utils.exceptions as onap_test_exceptions
from onaptests.steps.instantiate.vl_ala_carte import YamlTemplateVlAlaCarteInstantiateStep
time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
try:
self.test.cleanup()
- except ValueError as error:
- self.__logger.info("service deleted as expected {0}".format(error))
+ except SDKException as error:
+ self.__logger.info("service deletion error {0}".format(error))
self.result = 100
else:
self.__logger.info("No cleanup requested. Test completed.")
from xtesting.core import testcase
from onapsdk.configuration import settings
+from onapsdk.exceptions import SDKException
import onaptests.utils.exceptions as onap_test_exceptions
from onaptests.steps.instantiate.vf_module_ala_carte import YamlTemplateVfModuleAlaCarteInstantiateStep
except onap_test_exceptions.OnapTestException as exc:
self.result = 0
self.__logger.error(exc.error_message)
+ except SDKException as error:
+ self.__logger.info("service error {0}".format(error))
finally:
self.stop_time = time.time()
from abc import ABC, abstractmethod
from typing import List
-from onapsdk.configuration import settings
from onapsdk.aai.business import Customer
+from onapsdk.configuration import settings
+from onapsdk.exceptions import SettingsError
from .reports_collection import Report, ReportsCollection, ReportStepStatus
# Setup Proxy if SOCK_HTTP is defined in settings
try:
cls.set_proxy(settings.SOCK_HTTP)
- except AttributeError:
+ except SettingsError:
pass
def __init__(self, cleanup: bool = False) -> None:
"""A&AI cloud region creation module."""
from onapsdk.aai.cloud_infrastructure import CloudRegion
from onapsdk.configuration import settings
+from onapsdk.exceptions import ResourceNotFound
from ..base import BaseStep
cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
cloud_region_id=settings.CLOUD_REGION_ID,
)
- except ValueError:
+ except ResourceNotFound:
CloudRegion.create(
cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
cloud_region_id=settings.CLOUD_REGION_ID,
from onapsdk.aai.cloud_infrastructure import Complex
from onapsdk.configuration import settings
+from onapsdk.exceptions import APIError
from ..base import BaseStep
"""
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
- )
+ try:
+ Complex.create(
+ physical_location_id=settings.COMPLEX_PHYSICAL_LOCATION_ID,
+ data_center_code=settings.COMPLEX_DATA_CENTER_CODE,
+ name=settings.COMPLEX_PHYSICAL_LOCATION_ID)
+ except APIError:
+ self._logger.warn("Try to update the complex failed.")
from onapsdk.aai.business import Customer
from onapsdk.configuration import settings
+from onapsdk.exceptions import APIError
from ..base import BaseStep
- GLOBAL_CUSTOMER_ID.
"""
super().execute()
- Customer.create(settings.GLOBAL_CUSTOMER_ID, settings.GLOBAL_CUSTOMER_ID, "INFRA")
+ try:
+ Customer.create(settings.GLOBAL_CUSTOMER_ID, settings.GLOBAL_CUSTOMER_ID, "INFRA")
+ except APIError:
+ self._logger.warn("Try to update the Customer failed.")
"""Connectivity info creation module."""
from onapsdk.configuration import settings
+from onapsdk.exceptions import APIError
from onapsdk.msb.k8s import ConnectivityInfo
from ..base import BaseStep
try:
self._logger.info("Check if k8s connectivity information exists")
ConnectivityInfo.get_connectivity_info_by_region_id(settings.CLOUD_REGION_ID)
- except ValueError:
+ except APIError:
self._logger.info("Create the k8s connectivity information")
ConnectivityInfo.create(settings.CLOUD_REGION_ID,
settings.CLOUD_REGION_CLOUD_OWNER,
from onapsdk.aai.cloud_infrastructure import CloudRegion
from onapsdk.configuration import settings
+from onapsdk.exceptions import ResourceNotFound
from ..base import BaseStep
from onaptests.steps.cloud.cloud_region_create import CloudRegionCreateStep
- TENANT_NAME.
"""
super().execute()
- cloud_region: CloudRegion = CloudRegion.get_by_id(
- cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
- cloud_region_id=settings.CLOUD_REGION_ID,
- )
- if not list(cloud_region.esr_system_infos):
+ try:
+ cloud_region: CloudRegion = CloudRegion.get_by_id(
+ cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
+ cloud_region_id=settings.CLOUD_REGION_ID)
+ except ResourceNotFound:
cloud_region.add_esr_system_info(
esr_system_info_id=str(uuid4()),
user_name=settings.VIM_USERNAME,
# if it does not exist, create it
try:
cloud_region.get_tenant(settings.TENANT_ID)
- except ValueError:
+ except ResourceNotFound:
self._logger.warning("Impossible to retrieve the Specificed Tenant")
self._logger.debug("If no multicloud selected, add the tenant")
cloud_region.add_tenant(
try:
cloud_region.get_availability_zone_by_name(
settings.AVAILABILITY_ZONE_NAME)
- except ValueError:
+ except ResourceNotFound:
cloud_region.add_availability_zone(
settings.AVAILABILITY_ZONE_NAME,
settings.AVAILABILITY_ZONE_TYPE)
from onapsdk.aai.business import Customer, ServiceInstance, ServiceSubscription
from onapsdk.configuration import settings
+from onapsdk.exceptions import APIError, ResourceNotFound
from onapsdk.msb.k8s import Definition
from onapsdk.so.instantiation import VnfParameter
######## Check profile for Definition ###################################
try:
rbdef.get_profile_by_name(k8s_profile_name)
- except ValueError:
+ except ResourceNotFound:
######## Create profile for Definition ###################################
profile = rbdef.create_profile(k8s_profile_name,
k8s_profile_namespace,
try:
profile = rbdef.get_profile_by_name(k8s_profile_name)
profile.delete()
- except ValueError:
+ except APIError:
self._logger.error("K8s profile deletion %s failed", k8s_profile_name)
raise onap_test_exceptions.ProfileCleanupException
super().cleanup()
from onapsdk.aai.business import Customer, ServiceInstance, ServiceSubscription
from onapsdk.aai.business.owning_entity import OwningEntity as AaiOwningEntity
from onapsdk.configuration import settings
+from onapsdk.exceptions import ResourceNotFound
from onapsdk.sdc.service import Service
from onapsdk.so.instantiation import ServiceInstantiation
from onapsdk.vid import Project
try:
owning_entity = AaiOwningEntity.get_by_owning_entity_name(
settings.OWNING_ENTITY)
- except ValueError:
+ except ResourceNotFound:
self._logger.info("Owning entity not found, create it")
owning_entity = AaiOwningEntity.create(settings.OWNING_ENTITY)
vid_project = Project.create(settings.PROJECT)
try:
owning_entity = AaiOwningEntity.get_by_owning_entity_name(
settings.OWNING_ENTITY)
- except ValueError:
+ except ResourceNotFound:
self._logger.info("Owning entity not found, create it")
owning_entity = AaiOwningEntity.create(settings.OWNING_ENTITY)
vid_project = Project.create(settings.PROJECT)
from onapsdk.aai.cloud_infrastructure.cloud_region import CloudRegion
from onapsdk.aai.cloud_infrastructure.tenant import Tenant
from onapsdk.configuration import settings
+from onapsdk.exceptions import ResourceNotFound
from onapsdk.sdc.service import Service
from onapsdk.so.instantiation import ServiceInstantiation
from onapsdk.vid import LineOfBusiness, Platform, Project
try:
owning_entity = OwningEntity.get_by_owning_entity_name(
settings.OWNING_ENTITY)
- except ValueError:
+ except ResourceNotFound:
self._logger.info("Owning entity not found, create it")
owning_entity = OwningEntity.create(settings.OWNING_ENTITY)
vid_project: Project = Project(settings.PROJECT)
Code Review / testsuite / pythonsdk-tests.git / commitdiff