--- /dev/null
+[pylama]
+linters = pylint,pycodestyle,pyflakes,mccabe
+
+[pylama:pycodestyle]
+max_line_length = 100
+
+[pylama:pylint]
+disable =
+ fixme,
+ logging-fstring-interpolation,
+ logging-format-interpolation,
+ consider-using-f-string,
+ unused-argument,
+ broad-exception-caught,
+ missing-module-docstring,
+ too-many-instance-attributes,
+ too-few-public-methods,
+ too-many-nested-blocks,
+ too-many-return-statements,
+ too-many-branches,
+ too-many-arguments,
+ too-many-locals,
+ too-many-statements,
+ too-many-boolean-expressions
+
+bad_functions = print
+load_plugins =
+ pylint.extensions.bad_builtin
import os
import sys
+from onapsdk.exceptions import ModuleError
import onaptests.utils.exceptions as onap_test_exceptions
+SETTING_FILE_EXCEPTIONS = {
+ "clearwater_ims": "clearwater_ims_nomulticloud_settings",
+ "basic_cnf": "basic_cnf_yaml_settings",
+ "basic_cds": "cba_enrichment_settings",
+ "basic_network": "basic_network_nomulticloud_settings",
+ "multi_vnf_macro": "multi_vnf_ubuntu_settings"
+}
+
+MODULES_TO_RELOAD = [
+ "onapsdk",
+ "onaptests"
+]
def get_entrypoints():
config = configparser.ConfigParser()
}
return entry_points_result
-def run_test(test_name, validation, force_cleanup, entry_point, settings_module):
+def run_test(test_name, validation, force_cleanup, entry_point):
settings_env = "ONAP_PYTHON_SDK_SETTINGS"
- if validation:
- validation_env = "PYTHON_SDK_TESTS_VALIDATION"
- os.environ[validation_env] = "True"
if force_cleanup:
validation_env = "PYTHON_SDK_TESTS_FORCE_CLEANUP"
os.environ[validation_env] = "True"
- os.environ[settings_env] = f"onaptests.configuration.{test_name}_settings"
- if not settings_module:
+
+ setting_file_name = f"{test_name}_settings"
+ if test_name in SETTING_FILE_EXCEPTIONS:
+ setting_file_name = SETTING_FILE_EXCEPTIONS[test_name]
+ os.environ[settings_env] = f"onaptests.configuration.{setting_file_name}"
+ try:
+ basic_settings = importlib.import_module("onaptests.configuration.settings")
+ if validation:
+ basic_settings.IF_VALIDATION = True
settings_module = importlib.import_module("onapsdk.configuration")
- else:
- settings_module = importlib.reload(settings_module)
- settings = settings_module.settings
+ except ModuleError:
+ raise onap_test_exceptions.TestConfigurationException(
+ f"Expected setting file {os.environ[settings_env]} not found")
+
+ if validation:
+ settings_module.settings.CLEANUP_FLAG = True
+ settings_module.settings.CLEANUP_ACTIVITY_TIMER = 1
+ settings_module.settings.SDC_CLEANUP = True
+ settings_module.settings.SERVICE_DISTRIBUTION_SLEEP_TIME = 1
+
# 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)
+ logging.config.dictConfig(settings_module.settings.LOG_CONFIG)
logger = logging.getLogger(test_name)
logger.info(f"Running {test_name} test")
test_module = importlib.import_module(entry_point["module"])
test_instance = getattr(test_module, entry_point["class"])()
- try:
- test_instance.run()
- test_instance.clean()
- if validation:
- test_instance.validate_execution()
- except onap_test_exceptions.TestConfigurationException:
- logger.error("Status Check configuration error")
- return settings_module
+ if validation:
+ validate_scenario_base_class(test_name, test_instance)
+ test_instance.run()
+ test_instance.clean()
+ if validation:
+ logger.info(f"Validating {test_name} test")
+ test_instance.validate()
+
+def validate_scenario_base_class(test_name, scenario):
+ has_scenario_base = False
+ for base in scenario.__class__.__bases__:
+ if base.__name__ in "ScenarioBase":
+ has_scenario_base = True
+ break
+ if not has_scenario_base:
+ raise TestConfigurationException(
+ f"[{test_name}] {scenario.__class__.__name__} class does not inherit from ScenarioBase")
def main(argv):
"""Script is used to run one or all the tests.
test_name = argv[0]
entry_points = get_entrypoints()
if test_name == "all":
- settings_module = None
+ modules_reload = False
for test_name, entry_point in entry_points.items():
- settings_module = run_test(test_name, validation, force_cleanup, entry_point, settings_module)
+ if modules_reload:
+ modules_to_keep = dict()
+ for module in sys.modules:
+ reload_module = False
+ for module_to_reload in MODULES_TO_RELOAD:
+ if module_to_reload in module:
+ reload_module = True
+ break
+ if not reload_module:
+ modules_to_keep[module] = sys.modules[module]
+ sys.modules.clear()
+ sys.modules.update(modules_to_keep)
+ run_test(
+ test_name, validation, force_cleanup, entry_point)
+ modules_reload = True
else:
entry_point = entry_points[test_name]
- run_test(test_name, validation, force_cleanup, entry_point, None)
+ run_test(test_name, validation, force_cleanup, entry_point)
if __name__ == "__main__":
main(sys.argv[1:])
from .settings import * # noqa
-""" Specific Basic clamp settings."""
+
CLEANUP_FLAG = False
CLAMP_DISTRIBUTION_TIMER = 10
-# pylint: disable=bad-whitespace
# The ONAP part
SERVICE_DETAILS = ("Onboarding, enriching a model with TCA." +
"Design a loop with Clamp and deploy it in Policy and DCAE")
-SERVICE_COMPONENTS = "SDC, CLAMP, POLICY, DCAE, DMAAP"
VENDOR_NAME = "basiclamp_vendor"
try:
# Try to retrieve the SERVICE NAME from the yaml file
- with open(SERVICE_YAML_TEMPLATE, "r") as yaml_template:
+ 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
import onaptests.utils.exceptions as onap_test_exceptions
from .settings import * # noqa
-""" Specific basic_cnf_macro with multicloud-k8s and yaml config scenario."""
+
+# Specific basic_cnf_macro with multicloud-k8s and yaml config scenario.
SERVICE_DETAILS = ("Onboarding, distribution and instantiation of a Apache CNF " +
"using macro and native CNF path: cnf-adapter + K8sPlugin")
-SERVICE_COMPONENTS = "SDC, DMAAP, AAI, SO, SDNC, CDS, Multicloud K8S"
CLEANUP_FLAG = True
try:
# Try to retrieve the SERVICE NAME from the yaml file
- with open(SERVICE_YAML_TEMPLATE, "r") as yaml_template:
+ 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()))
-except (FileNotFoundError, ValueError):
- raise onap_test_exceptions.TestConfigurationException
+except (FileNotFoundError, ValueError) as exc:
+ raise onap_test_exceptions.TestConfigurationException from exc
SERVICE_INSTANCE_NAME = f"basic_cnf_macro_{str(uuid4())}"
import onaptests.utils.exceptions as onap_test_exceptions
from .settings import * # noqa
-""" Specific basic_cnf with multicloud-k8s and yaml config scenario."""
+
+# Specific basic_cnf with multicloud-k8s and yaml config scenario.
SERVICE_DETAILS = ("Onboarding, distribution and instantiation of a CNF" +
"using à la carte and Multicloud K8S module")
-SERVICE_COMPONENTS = "SDC, DMAAP, AAI, SO, SDNC, Multicloud K8S"
# This scenario uses multicloud-k8s and not multicloud
# (no registration requested)
USE_MULTICLOUD = False
try:
# Try to retrieve the SERVICE NAME from the yaml file
- with open(SERVICE_YAML_TEMPLATE, "r") as yaml_template:
+ 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()))
-except (FileNotFoundError, ValueError):
- raise onap_test_exceptions.TestConfigurationException
+except (FileNotFoundError, ValueError) as exc:
+ raise onap_test_exceptions.TestConfigurationException from exc
CLEANUP_FLAG = True
# nb of seconds before cleanup in case cleanup option is set
-from .settings import * # noqa
-
import json
from pathlib import Path
from onaptests.utils.resources import get_resource_location
+from .settings import * # noqa
+
CLEANUP_FLAG = True
ANCHOR_DATA = json.dumps(
SCHEMA_SET_FILE = Path(get_resource_location("templates/artifacts/cps/bookstore.yang"))
SERVICE_NAME = "Basic CPS test"
-SERVICE_COMPONENTS = "CPS"
+SERVICE_DETAILS = "Validation of the most important CPS REST API requests"
QUERY_1 = "/bookstore/categories[@code='1']/books"
QUERY_2 = "//categories[@code='1']/books[@price=5 and @title='Dune']"
QUERY_3 = "//bookstore"
DB_PORT = 5432
DB_LOGIN = "login"
DB_PASSWORD = "password"
-CHECK_POSTGRESQL = False
\ No newline at end of file
+CHECK_POSTGRESQL = False
from yaml import load, SafeLoader
from onaptests.utils.resources import get_resource_location
import onaptests.utils.exceptions as onap_test_exceptions
+from .settings import IF_VALIDATION
from .settings import * # noqa
-""" Specific Basic Network without multicloud."""
-# pylint: disable=bad-whitespace
# The ONAP part
SERVICE_DETAILS = "Onboarding, distribution and instantiation of Basic 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)
try:
# Try to retrieve the SERVICE NAME from the yaml file
- with open(SERVICE_YAML_TEMPLATE, "r") as yaml_template:
+ 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()))
-except (FileNotFoundError, ValueError):
- raise onap_test_exceptions.TestConfigurationException
+except (FileNotFoundError, ValueError) as exc:
+ raise onap_test_exceptions.TestConfigurationException from exc
CLEANUP_FLAG = True
CLEANUP_ACTIVITY_TIMER = 10 # nb of seconds before cleanup in case cleanup option is set
# 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')
+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')
MODEL_YAML_TEMPLATE = None
from onaptests.utils.resources import get_resource_location
from .settings import * # noqa
-""" Creation of service to onboard"""
-
# We need to create a service file with a random service name,
# to be sure that we force onboarding
rendered_template = template.render(service_name=service_name)
- with open(SERVICE_YAML_TEMPLATE, 'w+') as file_to_write:
+ with open(SERVICE_YAML_TEMPLATE, 'w+', encoding="utf-8") as file_to_write:
file_to_write.write(rendered_template)
-"""Basic onboard service to only onboard a service in SDC"""
-
-# pylint: disable=bad-whitespace
# The ONAP part
-SERVICE_DETAILS = "Onboarding of an Ubuntu VM"
-SERVICE_COMPONENTS = "SDC"
+SERVICE_DETAILS = "Basic onboard service to only onboard a service in SDC"
# USE_MULTICLOUD = False
# Set ONLY_INSTANTIATE to true to run an instantiation without repeating
try:
# Try to retrieve the SERVICE NAME from the yaml file
- with open(SERVICE_YAML_TEMPLATE, "r") as yaml_template:
+ 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()))
-except (FileNotFoundError, ValueError):
- raise onap_test_exceptions.TestConfigurationException
+except (FileNotFoundError, ValueError) as exc:
+ raise onap_test_exceptions.TestConfigurationException from exc
# CLEANUP_FLAG = True
# CLEANUP_ACTIVITY_TIMER = 10 # nb of seconds before cleanup in case cleanup option is set
SERVICE_NAME = "Basic SDNC test"
-SERVICE_COMPONENTS = "SDNC"
+SERVICE_DETAILS = "Test basic functionality of SDNC controller"
SERVICE_ID = "pythonsdk-tests-service-01"
import os
-import openstack
from pathlib import Path
from uuid import uuid4
-from yaml import load, SafeLoader
+
+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"
CLEANUP_FLAG = True
GLOBAL_CUSTOMER_ID = "basicvm-customer"
-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"
try:
# Try to retrieve the SERVICE NAME from the yaml file
- with open(SERVICE_YAML_TEMPLATE, "r") as yaml_template:
+ 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()))
-except (FileNotFoundError, ValueError):
- raise onap_test_exceptions.TestConfigurationException
+except (FileNotFoundError, ValueError) as exc:
+ raise onap_test_exceptions.TestConfigurationException from exc
SERVICE_INSTANCE_NAME = f"basic_macro_{str(uuid4())}"
-from .basic_vm_macro_settings import * # noqa
-
from pathlib import Path
from onaptests.utils.resources import get_resource_location
+from .basic_vm_macro_settings import * # noqa
SERVICE_YAML_TEMPLATE = Path(get_resource_location(
"templates/vnf-services/basic_vm_macro_stability-service.yaml"))
from onaptests.utils.resources import get_resource_location
from .settings import * # noqa
-""" Specific Basic VM with multicloud and yaml config scenario."""
+
+# Specific Basic VM with multicloud and yaml config scenario.
SERVICE_DETAILS = ("Onboarding, distribution and instantiation of a VM" +
"using à la carte and Multicloud module")
-SERVICE_COMPONENTS = "SDC, DMAAP, AAI, SO, SDNC, Multicloud"
USE_MULTICLOUD = True
# Set ONLY_INSTANTIATE to true to run an instantiation without repeating
from yaml import load, SafeLoader
from onaptests.utils.resources import get_resource_location
import onaptests.utils.exceptions as onap_test_exceptions
+from .settings import IF_VALIDATION
from .settings import * # noqa
-""" Specific basic_vm without multicloud."""
-# pylint: disable=bad-whitespace
# The ONAP part
SERVICE_DETAILS = "Onboarding, distribution and instanitation of an Ubuntu VM using à la carte"
-SERVICE_COMPONENTS = "SDC, DMAAP, AAI, SO, SDNC"
USE_MULTICLOUD = False
# Set ONLY_INSTANTIATE to true to run an instantiation without repeating
try:
# Try to retrieve the SERVICE NAME from the yaml file
- with open(SERVICE_YAML_TEMPLATE, "r") as yaml_template:
+ 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()))
-except (FileNotFoundError, ValueError):
- raise onap_test_exceptions.TestConfigurationException
+except (FileNotFoundError, ValueError) as exc:
+ raise onap_test_exceptions.TestConfigurationException from exc
CLEANUP_FLAG = True
CLEANUP_ACTIVITY_TIMER = 10 # nb of seconds before cleanup in case cleanup option is set
# 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')
+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')
MODEL_YAML_TEMPLATE = None
SERVICE_NAME = "CDS blueprint enrichment"
+SERVICE_DETAILS = "Tests CDS blueprint enrichment and its upload to blueprint processor"
+
CLEANUP_FLAG = True
CDS_DD_FILE = Path(get_resource_location("templates/artifacts/dd.json"))
CLEANUP_FLAG = True
SERVICE_NAME = "CDS resource resolution"
+SERVICE_DETAILS = "Tests execution or resource resolution in CDS"
CLOUD_REGION_CLOUD_OWNER = "basicnf-owner" # must not contain _
CLOUD_REGION_ID = "k8sregion-cds"
CLOUD_REGION_TYPE = "k8s"
}
}
}
-CDS_NODE_PORT = 30449
import openstack
from yaml import load, SafeLoader
from onaptests.utils.resources import get_resource_location
+from .settings import IF_VALIDATION
from .settings import * # noqa
-""" Specific clearwater IMS without multicloud."""
-# pylint: disable=bad-whitespace
+SERVICE_DETAILS = "Onboarding, distribution and instantiation of a Clearwater IMS"
# The ONAP part
USE_MULTICLOUD = False
# Set ONLY_INSTANTIATE to true to run an instantiation without repeating
try:
# Try to retrieve the SERVICE NAME from the yaml file
- with open(SERVICE_YAML_TEMPLATE, "r") as yaml_template:
+ 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()))
except ValueError:
# 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')
-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.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.get('region_name', 'RegionOne')
+ CLOUD_DOMAIN = cloud.config.auth.get('project_domain_name', 'Default')
MODEL_YAML_TEMPLATE = None
import openstack
from jinja2 import Environment, PackageLoader
from onaptests.utils.resources import get_resource_location
+from .settings import IF_VALIDATION
from .settings import * # noqa
+
VNF_FILENAME_PREFIX = "multi-vnf-ubuntu"
SERVICE_NAME = f"multivnfubuntu{str(uuid.uuid4().hex)[:6]}"
+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
+ """generate the service file with a random service name
from a jinja template"""
env = Environment(
file_name = get_resource_location(f"templates/vnf-services/{filename}.yaml")
- with open(file_name, 'w+') as file_to_write:
+ with open(file_name, 'w+', encoding="utf-8") as file_to_write:
file_to_write.write(rendered_template)
GLOBAL_CUSTOMER_ID = "ubuntu-customer"
-TEST_CLOUD = os.getenv('OS_TEST_CLOUD') # Get values from clouds.yaml
-cloud = openstack.connect(cloud=TEST_CLOUD)
-VIM_USERNAME = cloud.config.auth.get('username', 'nso')
-VIM_PASSWORD = cloud.config.auth.get('password', 'Password123')
-VIM_SERVICE_URL = cloud.config.auth.get('auth_url', 'https://10.195.194.215:5000')
-TENANT_ID = cloud.config.auth.get('project_id', 'e2710e84063b421fab08189818761d55')
-TENANT_NAME = cloud.config.auth.get('project_name', 'nso')
-CLOUD_REGION_ID = cloud.config.auth.get('region_name', 'nso215')
-CLOUD_DOMAIN = cloud.config.auth.get('project_domain_name', 'Default')
+if not IF_VALIDATION:
+ TEST_CLOUD = os.getenv('OS_TEST_CLOUD') # Get values from clouds.yaml
+ cloud = openstack.connect(cloud=TEST_CLOUD)
+ VIM_USERNAME = cloud.config.auth.get('username', 'nso')
+ VIM_PASSWORD = cloud.config.auth.get('password', 'Password123')
+ VIM_SERVICE_URL = cloud.config.auth.get('auth_url', 'https://10.195.194.215:5000')
+ TENANT_ID = cloud.config.auth.get('project_id', 'e2710e84063b421fab08189818761d55')
+ TENANT_NAME = cloud.config.auth.get('project_name', 'nso')
+ CLOUD_REGION_ID = cloud.config.auth.get('region_name', 'nso215')
+ CLOUD_DOMAIN = cloud.config.auth.get('project_domain_name', 'Default')
OWNING_ENTITY = "seb"
PROJECT = "Project-UbuntuDemo"
VENDOR_NAME = "pnf_macro_vendor"
SERVICE_NAME = "test_pnf_macro"
+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"))
-"""Specific settings module.""" # pylint: disable=bad-whitespace
+"""Specific settings module."""
######################
# #
SERVICE_DISTRIBUTION_NUMBER_OF_TRIES = 30
SERVICE_DISTRIBUTION_SLEEP_TIME = 60
EXPOSE_SERVICES_NODE_PORTS = True
+CDS_NODE_PORT = 30449
IN_CLUSTER = False
VES_BASIC_AUTH = {'username': 'sample1', 'password': 'sample1'}
+IF_VALIDATION = False
from .settings import * # noqa
-""" Specific Status Check """
+
SERVICE_NAME = "Status Check"
SERVICE_DETAILS = "Checks status of all k8s resources in the selected namespace"
-SERVICE_COMPONENTS = "ALL"
STATUS_RESULTS_DIRECTORY = "/tmp"
STORE_ARTIFACTS = True
CHECK_POD_VERSIONS = True
"""Instantiate basic cnf using SO macro flow."""
+from yaml import SafeLoader, load
+
from onapsdk.configuration import settings
-from onaptests.scenario.scenario_base import BaseStep, ScenarioBase, YamlTemplateBaseScenarioStep
+from onaptests.scenario.scenario_base import (BaseStep, ScenarioBase,
+ YamlTemplateBaseScenarioStep)
from onaptests.steps.instantiate.service_macro import \
YamlTemplateServiceMacroInstantiateStep
from onaptests.steps.onboard.cds import CbaPublishStep
-from yaml import SafeLoader, load
class BasicCnfMacroStep(YamlTemplateBaseScenarioStep):
+ """Main basic cnf macro scenario step."""
def __init__(self):
"""Initialize step.
"""
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r") as 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
#!/usr/bin/env python
"""Basic CPS test case."""
from onapsdk.configuration import settings
-from onaptests.scenario.scenario_base import BaseStep, ScenarioBase
+from onaptests.scenario.scenario_base import BaseStep, ScenarioBase, BaseScenarioStep
from onaptests.steps.onboard.cps import (CheckPostgressDataBaseConnectionStep,
QueryCpsAnchorNodeStep)
-class BasicCpsStep(BaseStep):
+class BasicCpsStep(BaseScenarioStep):
+ """Main basic cps scenario step."""
+
def __init__(self):
"""Initialize step.
"""Instantiate basic vm using SO macro flow."""
+from yaml import SafeLoader, load
+
from onapsdk.configuration import settings
-from onaptests.scenario.scenario_base import BaseStep, ScenarioBase, YamlTemplateBaseScenarioStep
+from onaptests.scenario.scenario_base import (BaseStep, ScenarioBase,
+ YamlTemplateBaseScenarioStep)
from onaptests.steps.instantiate.service_macro import \
YamlTemplateServiceMacroInstantiateStep
from onaptests.steps.onboard.cds import CbaPublishStep
-from yaml import SafeLoader, load
class BasicVmMacroStep(YamlTemplateBaseScenarioStep):
+ """Main basic vm macro scenario step."""
def __init__(self):
"""Initialize step.
"""
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r") as 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
"""Instantiate basic vm using SO macro flow."""
+from yaml import SafeLoader, load
+
from onapsdk.configuration import settings
-from onaptests.scenario.scenario_base import BaseStep, ScenarioBase, YamlTemplateBaseScenarioStep
+from onaptests.scenario.scenario_base import (BaseStep, ScenarioBase,
+ YamlTemplateBaseScenarioStep)
from onaptests.steps.instantiate.service_macro import \
YamlTemplateServiceMacroInstantiateStep
from onaptests.steps.onboard.cds import CbaPublishStep
-from yaml import SafeLoader, load
class MultiVnfUbuntuMacroStep(YamlTemplateBaseScenarioStep):
+ """Main step for multi vnf instantiation with macro method."""
def __init__(self):
"""Initialize step.
"""
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r") as 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:
if not self._model_yaml_template:
- with open(settings.MODEL_YAML_TEMPLATE, "r") as model_yaml_template:
- self._model_yaml_template: dict = load(model_yaml_template)
+ with open(settings.MODEL_YAML_TEMPLATE, "r", encoding="utf-8") as model_yaml_template:
+ self._model_yaml_template: dict = load(model_yaml_template, SafeLoader)
return self._model_yaml_template
@property
"""Instantiate service with PNF using SO macro flow."""
-from onapsdk.configuration import settings
from yaml import SafeLoader, load
-from onaptests.scenario.scenario_base import BaseStep, ScenarioBase, YamlTemplateBaseScenarioStep
+from onapsdk.configuration import settings
+from onaptests.scenario.scenario_base import (BaseStep, ScenarioBase,
+ YamlTemplateBaseScenarioStep)
from onaptests.steps.instantiate.pnf_register_ves import \
SendPnfRegisterVesEvent
from onaptests.steps.instantiate.service_macro import \
"""
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r") as 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
import logging
import time
+from xtesting.core import testcase
+
from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
+from onapsdk.exceptions import SDKException, SettingsError
from onaptests.steps.base import BaseStep, YamlTemplateBaseStep
-from onaptests.utils.exceptions import OnapTestException
-from xtesting.core import testcase
+from onaptests.utils.exceptions import (OnapTestException,
+ TestConfigurationException)
class ScenarioBase(testcase.TestCase):
"""Init base scenario."""
if "case_name" not in kwargs:
kwargs["case_name"] = case_name_override
- self.scenario_name = kwargs["case_name"].replace("_", " ")
+ self.case_name = kwargs["case_name"]
+ self.scenario_name = self.case_name.replace("_", " ")
self.scenario_name = str.title(self.scenario_name)
self.__logger.info("%s Global Configuration:", self.scenario_name)
- for val in settings._settings:
- self.__logger.info("%s: %s", val, settings._settings[val])
+ for val_name, val in settings._settings.items():
+ self.__logger.info("%s: %s", val_name, val)
self.__logger.debug("%s init started", self.scenario_name)
super().__init__(**kwargs)
self.general_exception = None
+ self.test: BaseStep = None
- def run(self):
+ def run(self, **kwargs):
"""Run scenario and cleanup resources afterwards"""
self.start_time = time.time()
try:
self.__logger.info("Generate %s Test report", self.scenario_name)
self.test.reports_collection.generate_report()
- def validate_execution(self):
+ def validate(self):
+ """Validate implementation of the scenario."""
+
+ self._validate_service_details()
self.test.validate_execution()
+ self.test.validate_cleanup()
+
+ def _validate_service_details(self):
+ self._check_setting("SERVICE_NAME")
+ self._check_setting("SERVICE_DETAILS")
+
+ def _check_setting(self, name: str):
+ try:
+ if getattr(settings, name) == "":
+ raise TestConfigurationException(
+ f"[{self.case_name}] {name} setting is not defined")
+ except (KeyError, AttributeError, SettingsError) as exc:
+ raise TestConfigurationException(
+ f"[{self.case_name}] {name} setting is not defined") from exc
class BaseScenarioStep(BaseStep):
+ """Main scenario step that has no own execution method."""
def __init__(self, cleanup=False):
"""Initialize BaseScenarioStep step."""
class YamlTemplateBaseScenarioStep(YamlTemplateBaseStep, BaseScenarioStep):
+ """Main scenario yaml template step that has no own execution method."""
def __init__(self, cleanup=False):
"""Initialize YamlTemplateBaseScenarioStep step."""
OnapTestExceptionGroup,
SkipExecutionException,
SubstepExecutionException,
- SubstepExecutionExceptionGroup)
+ SubstepExecutionExceptionGroup,
+ TestConfigurationException)
-IF_VALIDATION = "PYTHON_SDK_TESTS_VALIDATION"
+# pylint: disable=protected-access
IF_FORCE_CLEANUP = "PYTHON_SDK_TESTS_FORCE_CLEANUP"
class StoreStateHandler(ABC):
+ """Decorator for storing the state of executed test step."""
@classmethod
def store_state(cls, fun=None, *, cleanup=False): # noqa
class BaseStep(StoreStateHandler, ABC):
"""Base step class."""
- """Indicates that Step has no dedicated cleanup method"""
+ # Indicates that Step has no dedicated cleanup method
HAS_NO_CLEANUP = False
_logger: logging.Logger = logging.getLogger("")
self._state_clean: bool = False
self._nesting_level: int = 0
self._break_on_error: bool = break_on_error
- self._is_validation_only = os.environ.get(IF_VALIDATION) is not None
+ self._is_validation_only = settings.IF_VALIDATION
self._is_force_cleanup = os.environ.get(IF_FORCE_CLEANUP) is not None
def add_step(self, step: "BaseStep") -> None:
if not self.is_root:
return self.parent.reports_collection
if not self._reports_collection:
- self._reports_collection = ReportsCollection()
+ self._reports_collection = ReportsCollection(self._component_list())
for step_report in itertools.chain(self.execution_reports, self.cleanup_reports):
self._reports_collection.put(step_report)
return self._reports_collection
"""
+ def _component_list(self, components: dict = None):
+ if not components:
+ components = {}
+ for step in self._steps:
+ components[step.component] = step.component
+ step._component_list(components)
+ if not self.is_root or not components:
+ components[self.component] = self.component
+ return list(components)
+
def _step_title(self, cleanup=False):
cleanup_label = " Cleanup:" if cleanup else ":"
return f"[{self.component}] {self.name}{cleanup_label} {self.description}"
substep_error = True
if step._break_on_error:
raise SubstepExecutionException from substep_err
- else:
- self._logger.exception(substep_err)
+ self._logger.exception(substep_err)
if self._steps:
if substep_error and self._break_on_error:
raise SubstepExecutionException("Cannot continue due to failed substeps")
if len(exceptions_to_raise) > 0:
if len(exceptions_to_raise) == 1:
raise exceptions_to_raise[0]
- else:
- raise SubstepExecutionExceptionGroup("Substep Exceptions", exceptions_to_raise)
+ raise SubstepExecutionExceptionGroup("Substep Exceptions", exceptions_to_raise)
def cleanup(self) -> None:
"""Step's cleanup.
Customer.set_proxy(onap_proxy)
def validate_execution(self):
+ """Validate if each step was executed by decorator."""
+
if self._is_validation_only:
- self._log_execution_state(f"VALIDATE [{self._executed}, {self._cleanup}]")
- if self._executed and self._cleanup and not self._cleaned_up:
- self._logger.error(f"{self._step_title()} Cleanup Not Executed")
- assert self._cleaned_up
- for step in reversed(self._steps):
+ self._log_execution_state(f"VALIDATE EXECUTION [{self._state_execute}]")
+ if not self._state_execute:
+ raise TestConfigurationException(
+ f"{self._step_title()} - Execute decorator was not called")
+ for step in self._steps:
step.validate_execution()
+ def validate_cleanup(self):
+ """Validate if each step was cleaned by decorator."""
+
+ if self._is_validation_only:
+ for step in reversed(self._steps):
+ step.validate_cleanup()
+ if self._cleanup:
+ self._log_execution_state(
+ f"VALIDATE CLEANUP [{self._state_clean}, {self._cleanup}]")
+ if not self._state_clean:
+ raise TestConfigurationException(
+ f"{self._step_title()} - Cleanup decorator was not called")
+
class YamlTemplateBaseStep(BaseStep, ABC):
"""Base YAML template step."""
class CheckK8sResourcesStep(BaseStep):
+ """Base step for check of k8s resources in the selected namespace."""
__logger = logging.getLogger(__name__)
class CheckBasicK8sResourcesStep(CheckK8sResourcesStep):
+ """Basic check of k8s resources in the selected namespace."""
def __init__(self, namespace: str, resource_type: str, k8s_res_class):
"""Init CheckBasicK8sResourcesStep."""
class CheckK8sConfigMapsStep(CheckBasicK8sResourcesStep):
+ """Check of k8s configmap in the selected namespace."""
def __init__(self, namespace: str):
"""Init CheckK8sConfigMapsStep."""
class CheckK8sSecretsStep(CheckBasicK8sResourcesStep):
+ """Check of k8s secrets in the selected namespace."""
def __init__(self, namespace: str):
"""Init CheckK8sSecretsStep."""
class CheckK8sIngressesStep(CheckBasicK8sResourcesStep):
+ """Check of k8s ingress in the selected namespace."""
def __init__(self, namespace: str):
"""Init CheckK8sIngressesStep."""
class CheckK8sPvcsStep(CheckK8sResourcesStep):
+ """Check of k8s pvcs in the selected namespace."""
def __init__(self, namespace: str):
"""Init CheckK8sPvcsStep."""
class CheckK8sResourcesUsingPodsStep(CheckK8sResourcesStep):
+ """Check of k8s respurces with pods in the selected namespace."""
def __init__(self, namespace: str, resource_type: str, pods_source):
"""Init CheckK8sResourcesUsingPodsStep."""
class CheckK8sJobsStep(CheckK8sResourcesUsingPodsStep):
+ """Check of k8s jobs in the selected namespace."""
__logger = logging.getLogger(__name__)
if not any(waiver_elt in job.name for waiver_elt in settings.WAIVER_LIST):
self.all_resources.append(job)
else:
- self.__logger.warn(
+ self.__logger.warning(
"Waiver pattern found in job, exclude %s", job.name)
jobs_pods += job_pods
class CheckK8sPodsStep(CheckK8sResourcesUsingPodsStep):
+ """Check of k8s pods in the selected namespace."""
__logger = logging.getLogger(__name__)
self.jinja_env.get_template('pod.html.j2').stream(pod=pod).dump(
'{}/pod-{}.html'.format(self.res_dir, pod.name))
if any(waiver_elt in pod.name for waiver_elt in settings.WAIVER_LIST):
- self.__logger.warn("Waiver pattern found in pod, exclude %s", pod.name)
+ self.__logger.warning("Waiver pattern found in pod, exclude %s", pod.name)
else:
self.all_resources.append(pod)
containers=containers).dump('{}/container_versions.html'.format(
self.res_dir))
# create a json file for version tracking
- with open(self.res_dir + "/onap_versions.json", "w") as write_file:
+ with open(self.res_dir + "/onap_versions.json", "w", encoding="utf-8") as write_file:
json.dump(pod_versions, write_file)
def _get_container_logs(self, pod, container, full=True, previous=False):
)
except UnicodeDecodeError:
logs = "{0} has an unicode decode error...".format(pod.name)
- self.__logger.error(
- "{0} has an unicode decode error in the logs...", pod.name,
- )
+ self.__logger.error(logs)
return logs
def _parse_container(self, pod, k8s_container, init=False): # noqa
with open(
"{}/pod-{}-{}.log".format(self.res_dir,
pod.name, container.name),
- 'w') as log_result:
+ 'w', encoding="utf-8") as log_result:
log_result.write(logs)
if (not container.ready) and container.restart_count > 0:
old_logs = self._get_container_logs(pod=pod, container=container,
"{}/pod-{}-{}.old.log".format(self.res_dir,
pod.name,
container.name),
- 'w') as log_result:
+ 'w', encoding="utf-8") as log_result:
log_result.write(old_logs)
- if (container.name in settings.FULL_LOGS_CONTAINERS):
+ if container.name in settings.FULL_LOGS_CONTAINERS:
logs = self._get_container_logs(pod=pod, container=container)
with open(
"{}/pod-{}-{}.log".format(self.res_dir,
pod.name, container.name),
- 'w') as log_result:
+ 'w', encoding="utf-8") as log_result:
log_result.write(logs)
- if (container.name in settings.SPECIFIC_LOGS_CONTAINERS):
+ if container.name in settings.SPECIFIC_LOGS_CONTAINERS:
for log_file in settings.SPECIFIC_LOGS_CONTAINERS[container.name]:
exec_command = ['/bin/sh', '-c', "cat {}".format(log_file)]
log_files[log_file] = stream(
"{}/pod-{}-{}-{}.log".format(
self.res_dir, pod.name,
container.name, log_file_slug),
- 'w') as log_result:
+ 'w', encoding="utf-8") as log_result:
log_result.write(log_files[log_file])
except client.rest.ApiException as exc:
self.__logger.warning("%scontainer %s of pod %s has an exception: %s",
log_files=log_files).dump('{}/pod-{}-{}-logs.html'.format(
self.res_dir, pod.name, container.name))
if any(waiver_elt in container.name for waiver_elt in settings.WAIVER_LIST):
- self.__logger.warn(
+ self.__logger.warning(
"Waiver pattern found in container, exclude %s", container.name)
else:
containers_list.append(container)
class CheckK8sServicesStep(CheckK8sResourcesUsingPodsStep):
+ """Check of k8s services in the selected namespace."""
def __init__(self, namespace: str, pods):
"""Init CheckK8sServicesStep."""
class CheckK8sDeploymentsStep(CheckK8sResourcesUsingPodsStep):
+ """Check of k8s deployments in the selected namespace."""
def __init__(self, namespace: str, pods):
"""Init CheckK8sDeploymentsStep."""
class CheckK8sReplicaSetsStep(CheckK8sResourcesUsingPodsStep):
+ """Check of k8s replicasets in the selected namespace."""
def __init__(self, namespace: str, pods):
"""Init CheckK8sReplicaSetsStep."""
class CheckK8sStatefulSetsStep(CheckK8sResourcesUsingPodsStep):
+ """Check of k8s statefulsets in the selected namespace."""
def __init__(self, namespace: str, pods):
"""Init CheckK8sStatefulSetsStep."""
class CheckK8sDaemonSetsStep(CheckK8sResourcesUsingPodsStep):
+ """Check of k8s daemonsets in the selected namespace."""
def __init__(self, namespace: str, pods):
"""Init CheckK8sDaemonSetsStep."""
daemonset=daemonset).dump('{}/daemonset-{}.html'.format(
self.res_dir, daemonset.name))
- if (k8s.status.number_ready < k8s.status.desired_number_scheduled):
+ if k8s.status.number_ready < k8s.status.desired_number_scheduled:
self._add_failing_resource(daemonset)
self.all_resources.append(daemonset)
for namespace in ([self.namespace] + settings.EXTRA_NAMESPACE_LIST):
self._init_namespace_steps(namespace)
+ self.pods = []
+ self.services = []
+ self.jobs = []
+ self.deployments = []
+ self.replicasets = []
+ self.statefulsets = []
+ self.daemonsets = []
+ self.pvcs = []
+ self.configmaps = []
+ self.secrets = []
+ self.ingresses = []
+ self.failing_statefulsets = []
+ self.failing_jobs = []
+ self.failing_deployments = []
+ self.failing_replicasets = []
+ self.failing_daemonsets = []
+ self.failing_pvcs = []
+
def _init_namespace_steps(self, namespace: str):
self.job_list_step = CheckK8sJobsStep(namespace)
self.pod_list_step = CheckK8sPodsStep(namespace, self.job_list_step)
ns_details = ns_details[step.namespace]
store_results(ns_details, step)
- with (Path(self.res_dir).joinpath(settings.STATUS_DETAILS_JSON)).open('w') as file:
+ with (Path(self.res_dir).joinpath(settings.STATUS_DETAILS_JSON)
+ ).open('w', encoding="utf-8") as file:
json.dump(details, file, indent=4)
if self.failing:
raise StatusCheckException
def map_by_name(self, resources):
+ """Get resources' names."""
return list(map(lambda resource: resource.name, resources))
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.")
+ self._logger.warning("Try to update the complex failed.")
try:
Customer.create(settings.GLOBAL_CUSTOMER_ID, settings.GLOBAL_CUSTOMER_ID, "INFRA")
except APIError:
- self._logger.warn("Try to update the Customer failed.")
+ self._logger.warning("Try to update the Customer failed.")
self.service_name = service_name
self.port = port
self.node_port = node_port
- if settings.IN_CLUSTER:
- config.load_incluster_config()
- else:
- config.load_kube_config(config_file=settings.K8S_CONFIG)
self.k8s_client: client.CoreV1Api = client.CoreV1Api()
@property
settings.K8S_ONAP_NAMESPACE
)
return service_data.spec.type == "NodePort"
- except ApiException:
+ except ApiException as exc:
self._logger.exception("Kubernetes API exception")
- raise OnapTestException
+ raise OnapTestException from exc
@BaseStep.store_state
def execute(self) -> None:
"""
super().execute()
+ if settings.IN_CLUSTER:
+ config.load_incluster_config()
+ else:
+ config.load_kube_config(config_file=settings.K8S_CONFIG)
if not self.is_service_node_port_type():
try:
self.k8s_client.patch_namespaced_service(
"nodePort": self.node_port}],
"type": "NodePort"}}
)
- except ApiException:
+ except ApiException as exc:
self._logger.exception("Kubernetes API exception")
- raise OnapTestException
- except urllib3.exceptions.HTTPError:
+ raise OnapTestException from exc
+ except urllib3.exceptions.HTTPError as exc:
self._logger.exception("Can't connect with k8s")
- raise OnapTestException
+ raise OnapTestException from exc
else:
self._logger.debug("Service already patched, skip")
+ @BaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Step cleanup.
}
]
)
- except ApiException:
+ except ApiException as exc:
self._logger.exception("Kubernetes API exception")
- raise OnapTestException
- except urllib3.exceptions.HTTPError:
+ raise OnapTestException from exc
+ except urllib3.exceptions.HTTPError as exc:
self._logger.exception("Can't connect with k8s")
- raise OnapTestException
+ raise OnapTestException from exc
else:
self._logger.debug("Service is not 'NodePort' type, skip")
return super().cleanup()
except APIError:
if settings.IN_CLUSTER:
token_file = "/var/run/secrets/kubernetes.io/serviceaccount/token"
- with open(token_file, "r") as file:
+ with open(token_file, "r", encoding="utf-8") as file:
user_token_value = file.read().strip()
jinja_env = Environment(autoescape=select_autoescape(['json.j2']),
loader=PackageLoader('onaptests.templates', 'kubeconfig'))
kubeconfig_data.encode('utf-8'))
else:
self._logger.info("Create the k8s connectivity information")
- ConnectivityInfo.create(settings.CLOUD_REGION_ID,
- settings.CLOUD_REGION_CLOUD_OWNER,
- open(settings.K8S_CONFIG, 'rb').read())
+ with open(settings.K8S_CONFIG, 'rb') as k8s_config:
+ ConnectivityInfo.create(settings.CLOUD_REGION_ID,
+ settings.CLOUD_REGION_CLOUD_OWNER,
+ k8s_config.read())
@BaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
from onapsdk.aai.cloud_infrastructure import CloudRegion
from onapsdk.configuration import settings
from onapsdk.exceptions import ResourceNotFound
+from onaptests.steps.cloud.cloud_region_create import CloudRegionCreateStep
from ..base import BaseStep
-from onaptests.steps.cloud.cloud_region_create import CloudRegionCreateStep
class RegisterCloudRegionStep(BaseStep):
def specific_k8s_init(self):
"""Do the specific part for k8s resource when k8s object is present."""
- pass
def __repr__(self):
return self.name
return self.name
def __eq__(self, other):
- if (isinstance(other, K8sResource)):
+ if isinstance(other, K8sResource):
return self.name == other.name
- else:
- return False
+ return False
class K8sPodParentResource(K8sResource):
"""
if self.is_root:
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r") as 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
k8s_profile_namespace,
settings.K8S_PROFILE_K8S_VERSION)
# Upload artifact for created profile
- profile.upload_artifact(open(settings.K8S_PROFILE_ARTIFACT_PATH, 'rb').read())
+ with open(settings.K8S_PROFILE_ARTIFACT_PATH, 'rb') as k8s_profile:
+ profile.upload_artifact(k8s_profile.read())
@BaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
try:
profile = rbdef.get_profile_by_name(k8s_profile_name)
profile.delete()
- except APIError:
+ except APIError as exc:
self._logger.error("K8s profile deletion %s failed", k8s_profile_name)
- raise onap_test_exceptions.ProfileCleanupException
+ raise onap_test_exceptions.ProfileCleanupException from exc
super().cleanup()
+import logging
+
from onapsdk.configuration import settings
from onapsdk.exceptions import APIError
from onapsdk.sdnc import VfModulePreload
if exc.response_status_code == 409:
self._logger.warning("SDNC service already exists.")
else:
- raise OnapTestException("SDNC service creation failed.")
+ raise OnapTestException("SDNC service creation failed.") from exc
@BaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
service.service_data = settings.SERVICE_CHANGED_DATA
service.update()
self._logger.info("SDNC service update is completed.")
- except APIError:
- raise OnapTestException("SDNC service update is failed.")
+ except APIError as exc:
+ raise OnapTestException("SDNC service update is failed.") from exc
class UploadVfModulePreloadStep(BaseSdncStep):
Get preload information from SDNC over GR-API.
"""
+ __logger = logging.getLogger(__name__)
+
def __init__(self):
"""Initialize step.
self._logger.info("Get existing SDNC service instance and update it over GR-API")
preloads = PreloadInformation.get_all()
for preload_information in preloads:
- print(preload_information)
+ self.__logger.debug(preload_information)
class TestSdncStep(BaseScenarioStep):
from yaml import load, SafeLoader
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.exceptions import ResourceNotFound
from onapsdk.so.instantiation import ServiceInstantiation
import onaptests.utils.exceptions as onap_test_exceptions
-from ..base import BaseStep, YamlTemplateBaseStep
+from ..base import 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)
- if not settings.ONLY_INSTANTIATE:
- self.add_step(ServiceOnboardStep(cleanup))
- self.add_step(ConnectServiceSubToCloudRegionStep(cleanup))
-
- @property
- def description(self) -> str:
- """Step description."""
- return "Instantiate service using SO a'la carte method."
-
- @property
- def component(self) -> str:
- """Component name."""
- return "SO"
-
- @BaseStep.store_state
- 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)
- try:
- owning_entity = AaiOwningEntity.get_by_owning_entity_name(
- settings.OWNING_ENTITY)
- except ResourceNotFound:
- self._logger.info("Owning entity not found, create it")
- owning_entity = AaiOwningEntity.create(settings.OWNING_ENTITY)
-
- service_instantiation = ServiceInstantiation.instantiate_ala_carte(
- service,
- cloud_region,
- tenant,
- customer,
- owning_entity,
- settings.PROJECT,
- service_instance_name=settings.SERVICE_INSTANCE_NAME
- )
- try:
- service_instantiation.wait_for_finish(settings.ORCHESTRATION_REQUEST_TIMEOUT)
- except TimeoutError:
- self._logger.error("Service instantiation %s timed out", self.service_instance_name)
- raise onap_test_exceptions.ServiceInstantiateException
- if service_instantiation.failed:
- self._logger.error("Service instantiation %s failed", self.service_instance_name)
- raise onap_test_exceptions.ServiceInstantiateException
+from ..onboard.service import YamlTemplateServiceOnboardStep
class YamlTemplateServiceAlaCarteInstantiateStep(YamlTemplateBaseStep):
"""
if self.is_root:
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r") as 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
service.name)
break
self._logger.info(
- "Service Distribution for %s ongoing, Wait for 60 s",
- service.name)
- time.sleep(60)
+ "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:
)
try:
service_instantiation.wait_for_finish(settings.ORCHESTRATION_REQUEST_TIMEOUT)
- except TimeoutError:
+ except TimeoutError as exc:
self._logger.error("Service instantiation %s timed out", self.service_instance_name)
- raise onap_test_exceptions.ServiceCleanupException
+ raise onap_test_exceptions.ServiceCleanupException from exc
if service_instantiation.failed:
self._logger.error("Service instantiation %s failed", self.service_instance_name)
raise onap_test_exceptions.ServiceInstantiateException
- else:
- self._load_customer_and_subscription(reload=True)
- self._load_service_instance()
+ self._load_customer_and_subscription(reload=True)
+ self._load_service_instance()
@YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
service_deletion = self._service_instance.delete(a_la_carte=True)
try:
service_deletion.wait_for_finish(settings.ORCHESTRATION_REQUEST_TIMEOUT)
- except TimeoutError:
+ except TimeoutError as exc:
self._logger.error("Service deletion %s timed out", self._service_instance_name)
- raise onap_test_exceptions.ServiceCleanupException
+ raise onap_test_exceptions.ServiceCleanupException from exc
if service_deletion.finished:
self._logger.info("Service %s deleted", self._service_instance_name)
else:
"""
if self.is_root:
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r") as 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
"""
if self.is_root:
if not self._model_yaml_template:
- with open(settings.MODEL_YAML_TEMPLATE, "r") as model_yaml_template:
- self._model_yaml_template: dict = load(model_yaml_template)
+ with open(settings.MODEL_YAML_TEMPLATE, "r",
+ encoding="utf-8") as model_yaml_template:
+ self._model_yaml_template: dict = load(model_yaml_template, SafeLoader)
return self._model_yaml_template
return self.parent.model_yaml_template
)
try:
service_instantiation.wait_for_finish(timeout=settings.ORCHESTRATION_REQUEST_TIMEOUT)
- except TimeoutError:
+ except TimeoutError as exc:
self._logger.error("Service instantiation %s timed out", self.service_instance_name)
- raise onap_test_exceptions.ServiceInstantiateException
+ raise onap_test_exceptions.ServiceInstantiateException from exc
if service_instantiation.failed:
self._logger.error("Service instantiation %s failed", self.service_instance_name)
raise onap_test_exceptions.ServiceInstantiateException
service_deletion = self._service_instance.delete(a_la_carte=False)
try:
service_deletion.wait_for_finish(timeout=settings.ORCHESTRATION_REQUEST_TIMEOUT)
- except TimeoutError:
+ except TimeoutError as exc:
self._logger.error("Service deletion %s timed out", self._service_instance_name)
- raise onap_test_exceptions.ServiceCleanupException
+ raise onap_test_exceptions.ServiceCleanupException from exc
if service_deletion.finished:
self._logger.info("Service %s deleted", self._service_instance_name)
else:
"""
if self.is_root:
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r") as 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
if vf_module_instantiation.failed:
self._logger.error("VfModule instantiation %s failed", vf_module.name)
raise onap_test_exceptions.VfModuleInstantiateException
- except TimeoutError:
+ except TimeoutError as exc:
self._logger.error("VfModule instantiation %s timed out", vf_module.name)
- raise onap_test_exceptions.VfModuleInstantiateException
+ raise onap_test_exceptions.VfModuleInstantiateException from exc
@YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
self._logger.error("VfModule deletion %s failed", vf_module.name)
raise onap_test_exceptions.VfModuleCleanupException
self._logger.info("VfModule %s deleted", vf_module.name)
- except TimeoutError:
+ except TimeoutError as exc:
self._logger.error("VfModule deletion %s timed out", vf_module.name)
- raise onap_test_exceptions.VfModuleCleanupException
+ raise onap_test_exceptions.VfModuleCleanupException from exc
super().cleanup()
"""
if self.is_root:
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r") as 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
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")
+ self._logger.warning("No Subnet defined")
else:
for subnet in net['subnets']:
yield Subnet(
if net_instantiation.failed:
self._logger.error("VL instantiation %s failed", net_instantiation.name)
raise onap_test_exceptions.NetworkInstantiateException
- except TimeoutError:
+ except TimeoutError as exc:
self._logger.error("VL instantiation %s timed out", net_instantiation.name)
- raise onap_test_exceptions.NetworkInstantiateException
+ raise onap_test_exceptions.NetworkInstantiateException from exc
@YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
if net_deletion.failed:
self._logger.error("VL deletion %s failed", net_instance.name)
raise onap_test_exceptions.NetworkCleanupException
- except TimeoutError:
+ except TimeoutError as exc:
self._logger.error("VL deletion %s timed out", net_instance.name)
- raise onap_test_exceptions.NetworkCleanupException
+ raise onap_test_exceptions.NetworkCleanupException from exc
super().cleanup()
"""
if self.is_root:
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r") as 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
if vnf_instantiation.failed:
self._logger.error("VNF instantiation %s failed", vnf.name)
raise onap_test_exceptions.VnfInstantiateException
- except TimeoutError:
+ except TimeoutError as exc:
self._logger.error("VNF instantiation %s timed out", vnf.name)
- raise onap_test_exceptions.VnfInstantiateException
+ raise onap_test_exceptions.VnfInstantiateException from exc
@YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
if vnf_deletion.failed:
self._logger.error("VNF deletion %s failed", vnf_instance.name)
raise onap_test_exceptions.VnfCleanupException
- except TimeoutError:
+ except TimeoutError as exc:
self._logger.error("VNF deletion %s timed out", vnf_instance.name)
- raise onap_test_exceptions.VnfCleanupException
+ raise onap_test_exceptions.VnfCleanupException from exc
super().cleanup()
#
# http://www.apache.org/licenses/LICENSE-2.0
"""Clamp Scenario class."""
-from yaml import load, SafeLoader
import random
import string
import time
-from onapsdk.clamp.clamp_element import Clamp
-from onapsdk.sdc.service import Service
+from yaml import SafeLoader, load
import onaptests.utils.exceptions as onap_test_exceptions
+from onapsdk.clamp.clamp_element import Clamp
from onapsdk.configuration import settings
-from onaptests.steps.onboard.clamp import OnboardClampStep
+from onapsdk.sdc.service import Service
+from onaptests.steps.base import YamlTemplateBaseStep
from onaptests.steps.loop.instantiate_loop import InstantiateLoop
-
-from ..base import YamlTemplateBaseStep
+from onaptests.steps.onboard.clamp import OnboardClampStep
class ClampStep(YamlTemplateBaseStep):
"""
if self.is_root:
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r") as 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
req_policies=30) # 30 required policy
self._logger.info("Operational policy found.")
if not exist:
- raise ValueError("Couldn't load the policy %s", policy)
+ 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."""
service.name)
break
self._logger.info(
- "Service Distribution for %s ongoing, Wait for 60 s",
- service.name)
- time.sleep(60)
+ "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:
import onaptests.utils.exceptions as onap_test_exceptions
+# pylint: disable=protected-access
class InstantiateLoop():
"""class instantiating a closed loop in clamp."""
loop = LoopInstance(template=self.template,
name=self.loop_name,
details={})
- details = loop.create()
- if 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")
def __init__(self) -> None:
"""Initialize CBA publish step."""
super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
- """Let's skip enrichment if enriched CBA is already present"""
+ # Let's skip enrichment if enriched CBA is already present
if Path.is_file(settings.CDS_CBA_UNENRICHED):
self.add_step(CbaEnrichStep())
elif settings.EXPOSE_SERVICES_NODE_PORTS:
"""
if self.is_root:
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r") as 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
# we add the artifact to the first VNF
self._logger.info("Try to add blueprint to %s", vf.name)
- payload_file = open(settings.CONFIGURATION_PATH + 'tca-microservice.yaml', 'rb')
- 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)
- payload_file.close()
+ 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")
def __init__(self) -> None:
"""Initialize step."""
super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.login = None
+ self.password = None
@property
def description(self) -> str:
return "Establish connection with Postgress and execute the query"
def get_database_credentials(self):
+ """Resolve CPS datbase credentials from k8s secret."""
+
if settings.IN_CLUSTER:
config.load_incluster_config()
else:
raise EnvironmentPreparationException("Error accessing secret") from e
def connect_to_postgress(self):
+ """Connect to CPS database and execute select query."""
+
self.get_database_credentials()
if self.login and self.password:
db_params = {
"""
if settings.MODEL_YAML_TEMPLATE:
return self.model_yaml_template
- elif self.is_root:
+ if self.is_root:
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r") as 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
"""
if self.is_root:
if not self._model_yaml_template:
- with open(settings.MODEL_YAML_TEMPLATE, "r") as model_yaml_template:
+ with open(settings.MODEL_YAML_TEMPLATE, "r",
+ encoding="utf-8") as model_yaml_template:
self._model_yaml_template: dict = load(model_yaml_template, SafeLoader)
return self._model_yaml_template
return self.parent.model_yaml_template
"""
if settings.MODEL_YAML_TEMPLATE:
return self.model_yaml_template
- else:
- return self.parent.yaml_template[self.parent.service_name]
+ return self.parent.yaml_template[self.parent.service_name]
@property
def model_yaml_template(self) -> dict:
vsp: Vsp = Vsp(name=f"{vnf['vnf_name']}_VSP")
vf: Vf = Vf(name=vnf['vnf_name'], vsp=vsp)
if not vf.created():
- if all([x in vnf for x in ["vnf_artifact_type",
- "vnf_artifact_name",
- "vnf_artifact_label",
- "vnf_artifact_file_path"]]):
+ 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():
"""
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()
+ with open(settings.VSP_FILE_PATH, "rb") as vsp_file:
+ vsp: Vsp = Vsp(name=settings.VSP_NAME,
+ vendor=vendor,
+ package=vsp_file)
+ vsp.onboard()
@BaseStep.store_state(cleanup=True)
def cleanup(self):
"""
if settings.MODEL_YAML_TEMPLATE:
return self.model_yaml_template
- else:
- return self.parent.yaml_template
+ return self.parent.yaml_template
@property
def model_yaml_template(self) -> dict:
class ReportsCollection:
"""Collection to store steps execution statuses."""
- def __init__(self) -> None:
+ def __init__(self, components: list) -> None:
"""Initialize collection."""
self._collection: list = []
+ self._components = components
def put(self, item: Report) -> None:
"""Put execution status dictionary.
step_report.step_execution_status == ReportStepStatus.FAIL))
def generate_report(self) -> None:
+ """Generate report files after execution of the test."""
+
usecase = settings.SERVICE_NAME
try:
details = settings.SERVICE_DETAILS
except (KeyError, AttributeError, SettingsError):
details = ""
- try:
- components = settings.SERVICE_COMPONENTS
- except (KeyError, AttributeError, SettingsError):
- components = ""
+ components = ""
+ for component in self._components:
+ components = f"{component}, {components}"
+ components = components.rstrip(", ")
jinja_env = Environment(
autoescape=select_autoescape(['html']),
]
}
with (Path(settings.REPORTING_FILE_DIRECTORY).joinpath(
- settings.JSON_REPORTING_FILE_NAME)).open('w') as file:
+ settings.JSON_REPORTING_FILE_NAME)).open('w', encoding="utf-8") as file:
json.dump(report_dict, file, indent=4)
"httpResponse": {
"body": expectation["response"]
}
- }
+ },
+ timeout=settings.DEFAULT_REQUEST_TIMEOUT
)
response.raise_for_status()
except (requests.ConnectionError, requests.HTTPError) as http_error:
self._logger.debug(f"Can't register cds-mockserver expectation: {str(http_error)}")
- raise OnapTestException("CDS mockserver not configured")
+ raise OnapTestException("CDS mockserver not configured") from http_error
from typing import Tuple
import requests
-from kubernetes import client, config, watch
-from onapsdk.configuration import settings
import urllib3
+from kubernetes import client, config, watch
+from onapsdk.configuration import settings
from onaptests.steps.base import BaseStep
from onaptests.steps.instantiate.msb_k8s import CreateInstanceStep
-from onaptests.utils.exceptions import EnvironmentPreparationException, OnapTestException
+from onaptests.utils.exceptions import (EnvironmentPreparationException,
+ OnapTestException)
class PnfSimulatorCnfRegisterStep(BaseStep):
if event["object"].status.phase == "Running":
return True
return status
- except urllib3.exceptions.HTTPError:
+ except urllib3.exceptions.HTTPError as exc:
self._logger.error("Can't connect with k8s")
- raise OnapTestException
+ raise OnapTestException from exc
def get_ves_protocol_ip_and_port(self) -> Tuple[str, str, str]:
"""Static method to get VES protocol, ip address and port.
proto = "https"
return proto, service.spec.cluster_ip, service.spec.ports[0].port
raise EnvironmentPreparationException("Couldn't get VES protocol, ip and port")
- except Exception:
+ except Exception as exc:
self._logger.exception("Can't connect with k8s")
- raise OnapTestException
+ raise OnapTestException from exc
@BaseStep.store_state
def execute(self) -> None:
}
}
}
- }
+ },
+ timeout=settings.DEFAULT_REQUEST_TIMEOUT
)
response.raise_for_status()
registered_successfully = True
chart_info_path = get_local_dir() / chart_info_file
try:
- with open(chart_info_path, 'r') as stream:
+ with open(chart_info_path, 'r', encoding="utf-8") as stream:
chart_info = yaml.safe_load(stream)
except IOError as err:
msg = f"{chart_info_file} not found."
msg = "Error during helm release installation."
raise EnvironmentPreparationException(msg) from err
+ @BaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Uninstall helm release."""
try:
"""Start simulators via simulators' API."""
from typing import Union, Optional, Dict
import requests
+from onapsdk.configuration import settings
from onaptests.steps.base import BaseStep
from onaptests.utils.exceptions import TestConfigurationException
def execute(self) -> None:
"""Send a start command to the simulator application."""
super().execute()
- requests.request(self.method.upper(), self.url, **self.data)
+ requests.request(self.method.upper(), self.url, **self.data,
+ timeout=settings.DEFAULT_REQUEST_TIMEOUT)
# http://www.apache.org/licenses/LICENSE-2.0
"""Module to define pythonsdk-test exceptions."""
-__author__ = ("Morgan Richomme <morgan.richomme@orange.com>")
+__author__ = "Morgan Richomme <morgan.richomme@orange.com>"
class OnapTestException(Exception):
def get_resource_location(relative_path: str):
+ """Get location of resource files.
+
+ Arguments:
+ relative_path (str): relative path of the resource
+
+ Returns:
+ full_path (Path): full path of the resource
+ """
return os.path.join(os.path.dirname(os.path.realpath(onaptests.__file__)),
relative_path)
[tox]
minversion = 3.2.0
-envlist = json,yaml,py,rst,md,pylama
+envlist = json,yaml,md,pylama
skipsdist = true
requires = pip >= 8
/bin/bash -c "coala --non-interactive --disable-caching --no-autoapply-warn md --files $(</tmp/.coalist_md) \ "
[testenv:pylama]
-deps = pylama
+deps =
+ pylama[all]
+ -rrequirements.txt
+skip_install = True
+setenv =
+ PYTHONPATH = {toxinidir}/src
+commands = pylama -o pylama.ini src/
+
+[testenv:validate]
+basepython = python3.11
+deps =
+ -rrequirements.txt
skip_install = True
-commands = pylama src/
+setenv =
+ PYTHONPATH = {toxinidir}/src
+commands = python run_test.py all true
Code Review / testsuite / pythonsdk-tests.git / commitdiff