aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--pylama.ini28
-rw-r--r--run_test.py89
-rw-r--r--src/onaptests/configuration/basic_clamp_settings.py6
-rw-r--r--src/onaptests/configuration/basic_cnf_macro_settings.py10
-rw-r--r--src/onaptests/configuration/basic_cnf_yaml_settings.py10
-rw-r--r--src/onaptests/configuration/basic_cps_settings.py8
-rw-r--r--src/onaptests/configuration/basic_network_nomulticloud_settings.py30
-rw-r--r--src/onaptests/configuration/basic_onboard_settings.py16
-rw-r--r--src/onaptests/configuration/basic_sdnc_settings.py2
-rw-r--r--src/onaptests/configuration/basic_vm_macro_settings.py33
-rw-r--r--src/onaptests/configuration/basic_vm_macro_stability_settings.py3
-rw-r--r--src/onaptests/configuration/basic_vm_multicloud_yaml_settings.py4
-rw-r--r--src/onaptests/configuration/basic_vm_settings.py30
-rw-r--r--src/onaptests/configuration/cba_enrichment_settings.py2
-rw-r--r--src/onaptests/configuration/cds_resource_resolution_settings.py2
-rw-r--r--src/onaptests/configuration/clearwater_ims_nomulticloud_settings.py25
-rw-r--r--src/onaptests/configuration/multi_vnf_ubuntu_settings.py26
-rw-r--r--src/onaptests/configuration/pnf_macro_settings.py1
-rw-r--r--src/onaptests/configuration/settings.py4
-rw-r--r--src/onaptests/configuration/status_settings.py3
-rw-r--r--src/onaptests/scenario/basic_cnf_macro.py9
-rw-r--r--src/onaptests/scenario/basic_cps.py6
-rw-r--r--src/onaptests/scenario/basic_vm_macro.py9
-rw-r--r--src/onaptests/scenario/multi_vnf_macro.py13
-rw-r--r--src/onaptests/scenario/pnf_macro.py7
-rw-r--r--src/onaptests/scenario/scenario_base.py39
-rw-r--r--src/onaptests/steps/base.py53
-rw-r--r--src/onaptests/steps/cloud/check_status.py62
-rw-r--r--src/onaptests/steps/cloud/complex_create.py2
-rw-r--r--src/onaptests/steps/cloud/customer_create.py2
-rw-r--r--src/onaptests/steps/cloud/expose_service_node_port.py29
-rw-r--r--src/onaptests/steps/cloud/k8s_connectivity_info_create.py9
-rw-r--r--src/onaptests/steps/cloud/register_cloud.py2
-rw-r--r--src/onaptests/steps/cloud/resources.py6
-rw-r--r--src/onaptests/steps/instantiate/k8s_profile_create.py9
-rw-r--r--src/onaptests/steps/instantiate/sdnc_service.py12
-rw-r--r--src/onaptests/steps/instantiate/service_ala_carte.py99
-rw-r--r--src/onaptests/steps/instantiate/service_macro.py15
-rw-r--r--src/onaptests/steps/instantiate/vf_module_ala_carte.py10
-rw-r--r--src/onaptests/steps/instantiate/vl_ala_carte.py12
-rw-r--r--src/onaptests/steps/instantiate/vnf_ala_carte.py10
-rw-r--r--src/onaptests/steps/loop/clamp.py22
-rw-r--r--src/onaptests/steps/loop/instantiate_loop.py5
-rw-r--r--src/onaptests/steps/onboard/cds.py2
-rw-r--r--src/onaptests/steps/onboard/clamp.py18
-rw-r--r--src/onaptests/steps/onboard/cps.py6
-rw-r--r--src/onaptests/steps/onboard/service.py7
-rw-r--r--src/onaptests/steps/onboard/vf.py11
-rw-r--r--src/onaptests/steps/onboard/vsp.py12
-rw-r--r--src/onaptests/steps/reports_collection.py15
-rw-r--r--src/onaptests/steps/simulator/cds_mockserver.py5
-rw-r--r--src/onaptests/steps/simulator/pnf_simulator_cnf/pnf_register.py18
-rw-r--r--src/onaptests/steps/wrapper/helm_charts.py3
-rw-r--r--src/onaptests/steps/wrapper/start.py4
-rw-r--r--src/onaptests/utils/exceptions.py2
-rw-r--r--src/onaptests/utils/resources.py8
-rw-r--r--tox.ini19
57 files changed, 520 insertions, 384 deletions
diff --git a/pylama.ini b/pylama.ini
new file mode 100644
index 0000000..b0b7bfa
--- /dev/null
+++ b/pylama.ini
@@ -0,0 +1,28 @@
+[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
diff --git a/run_test.py b/run_test.py
index f25820e..a7b407d 100644
--- a/run_test.py
+++ b/run_test.py
@@ -4,8 +4,21 @@ import logging.config
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()
@@ -23,37 +36,57 @@ def get_entrypoints():
}
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.
@@ -78,12 +111,26 @@ def main(argv):
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:])
diff --git a/src/onaptests/configuration/basic_clamp_settings.py b/src/onaptests/configuration/basic_clamp_settings.py
index ac99278..7563fb6 100644
--- a/src/onaptests/configuration/basic_clamp_settings.py
+++ b/src/onaptests/configuration/basic_clamp_settings.py
@@ -3,15 +3,13 @@ from onaptests.utils.resources import get_resource_location
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"
@@ -44,7 +42,7 @@ CONFIGURATION_PATH = get_resource_location("configuration/")
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
diff --git a/src/onaptests/configuration/basic_cnf_macro_settings.py b/src/onaptests/configuration/basic_cnf_macro_settings.py
index 56bbd81..60682ca 100644
--- a/src/onaptests/configuration/basic_cnf_macro_settings.py
+++ b/src/onaptests/configuration/basic_cnf_macro_settings.py
@@ -7,10 +7,10 @@ from onaptests.utils.resources import get_resource_location
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
@@ -68,11 +68,11 @@ SERVICE_YAML_TEMPLATE = Path(get_resource_location(
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())}"
diff --git a/src/onaptests/configuration/basic_cnf_yaml_settings.py b/src/onaptests/configuration/basic_cnf_yaml_settings.py
index 4eb380c..34a1d4d 100644
--- a/src/onaptests/configuration/basic_cnf_yaml_settings.py
+++ b/src/onaptests/configuration/basic_cnf_yaml_settings.py
@@ -4,10 +4,10 @@ from onaptests.utils.resources import get_resource_location
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
@@ -21,11 +21,11 @@ SERVICE_YAML_TEMPLATE = get_resource_location("templates/vnf-services/basic_cnf-
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
diff --git a/src/onaptests/configuration/basic_cps_settings.py b/src/onaptests/configuration/basic_cps_settings.py
index 690499f..cbf9d4c 100644
--- a/src/onaptests/configuration/basic_cps_settings.py
+++ b/src/onaptests/configuration/basic_cps_settings.py
@@ -1,10 +1,10 @@
-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(
@@ -63,7 +63,7 @@ SCHEMA_SET_NAME = "basic-cps-test-schema-set"
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"
@@ -73,4 +73,4 @@ DB_PRIMARY_HOST = "cps-core-pg-primary"
DB_PORT = 5432
DB_LOGIN = "login"
DB_PASSWORD = "password"
-CHECK_POSTGRESQL = False \ No newline at end of file
+CHECK_POSTGRESQL = False
diff --git a/src/onaptests/configuration/basic_network_nomulticloud_settings.py b/src/onaptests/configuration/basic_network_nomulticloud_settings.py
index ed2cf41..7cecb51 100644
--- a/src/onaptests/configuration/basic_network_nomulticloud_settings.py
+++ b/src/onaptests/configuration/basic_network_nomulticloud_settings.py
@@ -3,14 +3,12 @@ import openstack
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)
@@ -22,11 +20,11 @@ SERVICE_YAML_TEMPLATE = get_resource_location("templates/vnf-services/basic_netw
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
@@ -54,15 +52,15 @@ SERVICE_INSTANCE_NAME = "basicnw_service_instance"
# The cloud Part
# Assuming a cloud.yaml is available, use the openstack client
# to retrieve cloud info and avoid data duplication
-TEST_CLOUD = os.getenv('OS_TEST_CLOUD')
-TEST_CLOUD = os.getenv('OS_TEST_CLOUD')
-cloud = openstack.connect(cloud=TEST_CLOUD)
-VIM_USERNAME = cloud.config.auth.get('username', 'Fill me')
-VIM_PASSWORD = cloud.config.auth.get('password', 'Fill me')
-VIM_SERVICE_URL = cloud.config.auth.get('auth_url', 'Fill me')
-TENANT_ID = cloud.config.auth.get('project_id', 'Fill me')
-TENANT_NAME = cloud.config.auth.get('project_name', 'Fill me')
-CLOUD_REGION_ID = cloud.config.auth.get('region_name', 'RegionOne')
-CLOUD_DOMAIN = cloud.config.auth.get('project_domain_name', 'Default')
+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
diff --git a/src/onaptests/configuration/basic_onboard_settings.py b/src/onaptests/configuration/basic_onboard_settings.py
index c4a1a85..e757035 100644
--- a/src/onaptests/configuration/basic_onboard_settings.py
+++ b/src/onaptests/configuration/basic_onboard_settings.py
@@ -8,8 +8,6 @@ import onaptests.utils.exceptions as onap_test_exceptions
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
@@ -30,16 +28,12 @@ def generate_service_config_yaml_file():
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
@@ -54,11 +48,11 @@ generate_service_config_yaml_file()
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
diff --git a/src/onaptests/configuration/basic_sdnc_settings.py b/src/onaptests/configuration/basic_sdnc_settings.py
index d364616..8fedf32 100644
--- a/src/onaptests/configuration/basic_sdnc_settings.py
+++ b/src/onaptests/configuration/basic_sdnc_settings.py
@@ -4,7 +4,7 @@ CLEANUP_FLAG = True
SERVICE_NAME = "Basic SDNC test"
-SERVICE_COMPONENTS = "SDNC"
+SERVICE_DETAILS = "Test basic functionality of SDNC controller"
SERVICE_ID = "pythonsdk-tests-service-01"
diff --git a/src/onaptests/configuration/basic_vm_macro_settings.py b/src/onaptests/configuration/basic_vm_macro_settings.py
index 1aeea26..ce1a607 100644
--- a/src/onaptests/configuration/basic_vm_macro_settings.py
+++ b/src/onaptests/configuration/basic_vm_macro_settings.py
@@ -1,13 +1,17 @@
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
@@ -32,15 +36,16 @@ COMPLEX_DATA_CENTER_CODE = "1234-5"
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"
@@ -52,11 +57,11 @@ SERVICE_YAML_TEMPLATE = Path(get_resource_location(
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())}"
diff --git a/src/onaptests/configuration/basic_vm_macro_stability_settings.py b/src/onaptests/configuration/basic_vm_macro_stability_settings.py
index c128ae4..58e4395 100644
--- a/src/onaptests/configuration/basic_vm_macro_stability_settings.py
+++ b/src/onaptests/configuration/basic_vm_macro_stability_settings.py
@@ -1,9 +1,8 @@
-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"))
diff --git a/src/onaptests/configuration/basic_vm_multicloud_yaml_settings.py b/src/onaptests/configuration/basic_vm_multicloud_yaml_settings.py
index 838a5ec..1cb1aec 100644
--- a/src/onaptests/configuration/basic_vm_multicloud_yaml_settings.py
+++ b/src/onaptests/configuration/basic_vm_multicloud_yaml_settings.py
@@ -1,10 +1,10 @@
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
diff --git a/src/onaptests/configuration/basic_vm_settings.py b/src/onaptests/configuration/basic_vm_settings.py
index 8066625..f26bbd8 100644
--- a/src/onaptests/configuration/basic_vm_settings.py
+++ b/src/onaptests/configuration/basic_vm_settings.py
@@ -3,14 +3,12 @@ import openstack
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
@@ -23,11 +21,11 @@ SERVICE_YAML_TEMPLATE = get_resource_location("templates/vnf-services/basic_vm-s
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
@@ -58,15 +56,15 @@ SERVICE_INSTANCE_NAME = "basic_vm_service_instance"
# The cloud Part
# Assuming a cloud.yaml is available, use the openstack client
# to retrieve cloud info and avoid data duplication
-TEST_CLOUD = os.getenv('OS_TEST_CLOUD')
-TEST_CLOUD = os.getenv('OS_TEST_CLOUD')
-cloud = openstack.connect(cloud=TEST_CLOUD)
-VIM_USERNAME = cloud.config.auth.get('username', 'Fill me')
-VIM_PASSWORD = cloud.config.auth.get('password', 'Fill me')
-VIM_SERVICE_URL = cloud.config.auth.get('auth_url', 'Fill me')
-TENANT_ID = cloud.config.auth.get('project_id', 'Fill me')
-TENANT_NAME = cloud.config.auth.get('project_name', 'Fill me')
-CLOUD_REGION_ID = cloud.config.auth.get('region_name', 'RegionOne')
-CLOUD_DOMAIN = cloud.config.auth.get('project_domain_name', 'Default')
+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
diff --git a/src/onaptests/configuration/cba_enrichment_settings.py b/src/onaptests/configuration/cba_enrichment_settings.py
index 847e15e..534ebbe 100644
--- a/src/onaptests/configuration/cba_enrichment_settings.py
+++ b/src/onaptests/configuration/cba_enrichment_settings.py
@@ -5,6 +5,8 @@ from .settings import * # noqa
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"))
diff --git a/src/onaptests/configuration/cds_resource_resolution_settings.py b/src/onaptests/configuration/cds_resource_resolution_settings.py
index 0ab6ebc..f7573a6 100644
--- a/src/onaptests/configuration/cds_resource_resolution_settings.py
+++ b/src/onaptests/configuration/cds_resource_resolution_settings.py
@@ -6,6 +6,7 @@ from .settings import * # noqa
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"
@@ -82,4 +83,3 @@ CDS_WORKFLOW_EXPECTED_OUTPUT = {
}
}
}
-CDS_NODE_PORT = 30449
diff --git a/src/onaptests/configuration/clearwater_ims_nomulticloud_settings.py b/src/onaptests/configuration/clearwater_ims_nomulticloud_settings.py
index 0965e9d..07c966c 100644
--- a/src/onaptests/configuration/clearwater_ims_nomulticloud_settings.py
+++ b/src/onaptests/configuration/clearwater_ims_nomulticloud_settings.py
@@ -2,11 +2,11 @@ import os
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
@@ -24,7 +24,7 @@ SERVICE_YAML_TEMPLATE = get_resource_location("templates/vnf-services/clearwater
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:
@@ -51,14 +51,15 @@ SERVICE_INSTANCE_NAME = "clearwater-ims_service_instance"
# The cloud Part
# Assuming a cloud.yaml is available, use the openstack client
# to retrieve cloud info and avoid data duplication
-TEST_CLOUD = os.getenv('OS_TEST_CLOUD')
-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
diff --git a/src/onaptests/configuration/multi_vnf_ubuntu_settings.py b/src/onaptests/configuration/multi_vnf_ubuntu_settings.py
index 1b9d4a2..a23d72e 100644
--- a/src/onaptests/configuration/multi_vnf_ubuntu_settings.py
+++ b/src/onaptests/configuration/multi_vnf_ubuntu_settings.py
@@ -4,16 +4,19 @@ from pathlib import Path
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(
@@ -25,7 +28,7 @@ def generate_service_config_yaml_file(filename):
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)
@@ -50,15 +53,16 @@ COMPLEX_DATA_CENTER_CODE = "nlt"
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"
diff --git a/src/onaptests/configuration/pnf_macro_settings.py b/src/onaptests/configuration/pnf_macro_settings.py
index d332763..53d22e7 100644
--- a/src/onaptests/configuration/pnf_macro_settings.py
+++ b/src/onaptests/configuration/pnf_macro_settings.py
@@ -10,6 +10,7 @@ USE_MULTICLOUD = False
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"))
diff --git a/src/onaptests/configuration/settings.py b/src/onaptests/configuration/settings.py
index e43a15e..399f3f7 100644
--- a/src/onaptests/configuration/settings.py
+++ b/src/onaptests/configuration/settings.py
@@ -1,4 +1,4 @@
-"""Specific settings module.""" # pylint: disable=bad-whitespace
+"""Specific settings module."""
######################
# #
@@ -56,5 +56,7 @@ ORCHESTRATION_REQUEST_TIMEOUT = 60.0 * 15 # 15 minutes in seconds
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
diff --git a/src/onaptests/configuration/status_settings.py b/src/onaptests/configuration/status_settings.py
index 7bcb6d0..13fb8cd 100644
--- a/src/onaptests/configuration/status_settings.py
+++ b/src/onaptests/configuration/status_settings.py
@@ -1,9 +1,8 @@
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
diff --git a/src/onaptests/scenario/basic_cnf_macro.py b/src/onaptests/scenario/basic_cnf_macro.py
index 1fecd95..71b6863 100644
--- a/src/onaptests/scenario/basic_cnf_macro.py
+++ b/src/onaptests/scenario/basic_cnf_macro.py
@@ -1,13 +1,16 @@
"""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.
@@ -57,7 +60,7 @@ class BasicCnfMacroStep(YamlTemplateBaseScenarioStep):
"""
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
diff --git a/src/onaptests/scenario/basic_cps.py b/src/onaptests/scenario/basic_cps.py
index adfefb5..f6189da 100644
--- a/src/onaptests/scenario/basic_cps.py
+++ b/src/onaptests/scenario/basic_cps.py
@@ -1,12 +1,14 @@
#!/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.
diff --git a/src/onaptests/scenario/basic_vm_macro.py b/src/onaptests/scenario/basic_vm_macro.py
index ec60e78..0045761 100644
--- a/src/onaptests/scenario/basic_vm_macro.py
+++ b/src/onaptests/scenario/basic_vm_macro.py
@@ -1,13 +1,16 @@
"""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.
@@ -57,7 +60,7 @@ class BasicVmMacroStep(YamlTemplateBaseScenarioStep):
"""
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
diff --git a/src/onaptests/scenario/multi_vnf_macro.py b/src/onaptests/scenario/multi_vnf_macro.py
index fc00cec..47b1830 100644
--- a/src/onaptests/scenario/multi_vnf_macro.py
+++ b/src/onaptests/scenario/multi_vnf_macro.py
@@ -1,13 +1,16 @@
"""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.
@@ -58,15 +61,15 @@ class MultiVnfUbuntuMacroStep(YamlTemplateBaseScenarioStep):
"""
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
diff --git a/src/onaptests/scenario/pnf_macro.py b/src/onaptests/scenario/pnf_macro.py
index f8afad5..88d42d1 100644
--- a/src/onaptests/scenario/pnf_macro.py
+++ b/src/onaptests/scenario/pnf_macro.py
@@ -1,8 +1,9 @@
"""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 \
@@ -66,7 +67,7 @@ class PnfMacroScenarioStep(YamlTemplateBaseScenarioStep):
"""
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
diff --git a/src/onaptests/scenario/scenario_base.py b/src/onaptests/scenario/scenario_base.py
index 179bea8..7069afb 100644
--- a/src/onaptests/scenario/scenario_base.py
+++ b/src/onaptests/scenario/scenario_base.py
@@ -1,11 +1,13 @@
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):
@@ -17,18 +19,20 @@ 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:
@@ -62,11 +66,29 @@ class ScenarioBase(testcase.TestCase):
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."""
@@ -78,6 +100,7 @@ class BaseScenarioStep(BaseStep):
class YamlTemplateBaseScenarioStep(YamlTemplateBaseStep, BaseScenarioStep):
+ """Main scenario yaml template step that has no own execution method."""
def __init__(self, cleanup=False):
"""Initialize YamlTemplateBaseScenarioStep step."""
diff --git a/src/onaptests/steps/base.py b/src/onaptests/steps/base.py
index 4fca3e7..918bc6d 100644
--- a/src/onaptests/steps/base.py
+++ b/src/onaptests/steps/base.py
@@ -16,14 +16,16 @@ from onaptests.utils.exceptions import (OnapTestException,
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
@@ -117,7 +119,7 @@ class StoreStateHandler(ABC):
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("")
@@ -159,7 +161,7 @@ class BaseStep(StoreStateHandler, ABC):
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:
@@ -228,7 +230,7 @@ class BaseStep(StoreStateHandler, ABC):
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
@@ -294,6 +296,16 @@ class BaseStep(StoreStateHandler, ABC):
"""
+ 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}"
@@ -332,8 +344,7 @@ class BaseStep(StoreStateHandler, ABC):
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")
@@ -364,8 +375,7 @@ class BaseStep(StoreStateHandler, ABC):
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.
@@ -391,14 +401,29 @@ class BaseStep(StoreStateHandler, ABC):
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."""
diff --git a/src/onaptests/steps/cloud/check_status.py b/src/onaptests/steps/cloud/check_status.py
index cbefa18..b3fab03 100644
--- a/src/onaptests/steps/cloud/check_status.py
+++ b/src/onaptests/steps/cloud/check_status.py
@@ -24,6 +24,7 @@ from .resources import (ConfigMap, Container, DaemonSet, Deployment, Ingress,
class CheckK8sResourcesStep(BaseStep):
+ """Base step for check of k8s resources in the selected namespace."""
__logger = logging.getLogger(__name__)
@@ -110,6 +111,7 @@ class CheckK8sResourcesStep(BaseStep):
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."""
@@ -129,6 +131,7 @@ class CheckBasicK8sResourcesStep(CheckK8sResourcesStep):
class CheckK8sConfigMapsStep(CheckBasicK8sResourcesStep):
+ """Check of k8s configmap in the selected namespace."""
def __init__(self, namespace: str):
"""Init CheckK8sConfigMapsStep."""
@@ -140,6 +143,7 @@ class CheckK8sConfigMapsStep(CheckBasicK8sResourcesStep):
class CheckK8sSecretsStep(CheckBasicK8sResourcesStep):
+ """Check of k8s secrets in the selected namespace."""
def __init__(self, namespace: str):
"""Init CheckK8sSecretsStep."""
@@ -151,6 +155,7 @@ class CheckK8sSecretsStep(CheckBasicK8sResourcesStep):
class CheckK8sIngressesStep(CheckBasicK8sResourcesStep):
+ """Check of k8s ingress in the selected namespace."""
def __init__(self, namespace: str):
"""Init CheckK8sIngressesStep."""
@@ -162,6 +167,7 @@ class CheckK8sIngressesStep(CheckBasicK8sResourcesStep):
class CheckK8sPvcsStep(CheckK8sResourcesStep):
+ """Check of k8s pvcs in the selected namespace."""
def __init__(self, namespace: str):
"""Init CheckK8sPvcsStep."""
@@ -194,6 +200,7 @@ class CheckK8sPvcsStep(CheckK8sResourcesStep):
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."""
@@ -231,6 +238,7 @@ class CheckK8sResourcesUsingPodsStep(CheckK8sResourcesStep):
class CheckK8sJobsStep(CheckK8sResourcesUsingPodsStep):
+ """Check of k8s jobs in the selected namespace."""
__logger = logging.getLogger(__name__)
@@ -273,12 +281,13 @@ class CheckK8sJobsStep(CheckK8sResourcesUsingPodsStep):
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__)
@@ -400,7 +409,7 @@ class CheckK8sPodsStep(CheckK8sResourcesUsingPodsStep):
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)
@@ -412,7 +421,7 @@ class CheckK8sPodsStep(CheckK8sResourcesUsingPodsStep):
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):
@@ -430,9 +439,7 @@ class CheckK8sPodsStep(CheckK8sResourcesUsingPodsStep):
)
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
@@ -463,7 +470,7 @@ class CheckK8sPodsStep(CheckK8sResourcesUsingPodsStep):
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,
@@ -472,16 +479,16 @@ class CheckK8sPodsStep(CheckK8sResourcesUsingPodsStep):
"{}/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(
@@ -499,7 +506,7 @@ class CheckK8sPodsStep(CheckK8sResourcesUsingPodsStep):
"{}/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",
@@ -512,7 +519,7 @@ class CheckK8sPodsStep(CheckK8sResourcesUsingPodsStep):
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)
@@ -522,6 +529,7 @@ class CheckK8sPodsStep(CheckK8sResourcesUsingPodsStep):
class CheckK8sServicesStep(CheckK8sResourcesUsingPodsStep):
+ """Check of k8s services in the selected namespace."""
def __init__(self, namespace: str, pods):
"""Init CheckK8sServicesStep."""
@@ -547,6 +555,7 @@ class CheckK8sServicesStep(CheckK8sResourcesUsingPodsStep):
class CheckK8sDeploymentsStep(CheckK8sResourcesUsingPodsStep):
+ """Check of k8s deployments in the selected namespace."""
def __init__(self, namespace: str, pods):
"""Init CheckK8sDeploymentsStep."""
@@ -585,6 +594,7 @@ class CheckK8sDeploymentsStep(CheckK8sResourcesUsingPodsStep):
class CheckK8sReplicaSetsStep(CheckK8sResourcesUsingPodsStep):
+ """Check of k8s replicasets in the selected namespace."""
def __init__(self, namespace: str, pods):
"""Init CheckK8sReplicaSetsStep."""
@@ -625,6 +635,7 @@ class CheckK8sReplicaSetsStep(CheckK8sResourcesUsingPodsStep):
class CheckK8sStatefulSetsStep(CheckK8sResourcesUsingPodsStep):
+ """Check of k8s statefulsets in the selected namespace."""
def __init__(self, namespace: str, pods):
"""Init CheckK8sStatefulSetsStep."""
@@ -665,6 +676,7 @@ class CheckK8sStatefulSetsStep(CheckK8sResourcesUsingPodsStep):
class CheckK8sDaemonSetsStep(CheckK8sResourcesUsingPodsStep):
+ """Check of k8s daemonsets in the selected namespace."""
def __init__(self, namespace: str, pods):
"""Init CheckK8sDaemonSetsStep."""
@@ -694,7 +706,7 @@ class CheckK8sDaemonSetsStep(CheckK8sResourcesUsingPodsStep):
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)
@@ -716,6 +728,24 @@ class CheckNamespaceStatusStep(CheckK8sResourcesStep):
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)
@@ -820,10 +850,12 @@ class CheckNamespaceStatusStep(CheckK8sResourcesStep):
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))
diff --git a/src/onaptests/steps/cloud/complex_create.py b/src/onaptests/steps/cloud/complex_create.py
index 96d8e7f..7afad16 100644
--- a/src/onaptests/steps/cloud/complex_create.py
+++ b/src/onaptests/steps/cloud/complex_create.py
@@ -38,4 +38,4 @@ class ComplexCreateStep(BaseStep):
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.")
diff --git a/src/onaptests/steps/cloud/customer_create.py b/src/onaptests/steps/cloud/customer_create.py
index 96d192a..7bffb1a 100644
--- a/src/onaptests/steps/cloud/customer_create.py
+++ b/src/onaptests/steps/cloud/customer_create.py
@@ -33,4 +33,4 @@ class CustomerCreateStep(BaseStep):
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.")
diff --git a/src/onaptests/steps/cloud/expose_service_node_port.py b/src/onaptests/steps/cloud/expose_service_node_port.py
index 1f43a65..4f6eecc 100644
--- a/src/onaptests/steps/cloud/expose_service_node_port.py
+++ b/src/onaptests/steps/cloud/expose_service_node_port.py
@@ -22,10 +22,6 @@ class ExposeServiceNodePortStep(BaseStep):
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
@@ -53,9 +49,9 @@ class ExposeServiceNodePortStep(BaseStep):
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:
@@ -68,6 +64,10 @@ class ExposeServiceNodePortStep(BaseStep):
"""
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(
@@ -77,15 +77,16 @@ class ExposeServiceNodePortStep(BaseStep):
"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.
@@ -109,12 +110,12 @@ class ExposeServiceNodePortStep(BaseStep):
}
]
)
- 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()
diff --git a/src/onaptests/steps/cloud/k8s_connectivity_info_create.py b/src/onaptests/steps/cloud/k8s_connectivity_info_create.py
index c22df47..410ac47 100644
--- a/src/onaptests/steps/cloud/k8s_connectivity_info_create.py
+++ b/src/onaptests/steps/cloud/k8s_connectivity_info_create.py
@@ -42,7 +42,7 @@ class K8SConnectivityInfoStep(BaseStep):
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'))
@@ -56,9 +56,10 @@ class K8SConnectivityInfoStep(BaseStep):
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:
diff --git a/src/onaptests/steps/cloud/register_cloud.py b/src/onaptests/steps/cloud/register_cloud.py
index fbfb6f3..5035f86 100644
--- a/src/onaptests/steps/cloud/register_cloud.py
+++ b/src/onaptests/steps/cloud/register_cloud.py
@@ -5,9 +5,9 @@ from uuid import uuid4
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):
diff --git a/src/onaptests/steps/cloud/resources.py b/src/onaptests/steps/cloud/resources.py
index f00b60e..775d574 100644
--- a/src/onaptests/steps/cloud/resources.py
+++ b/src/onaptests/steps/cloud/resources.py
@@ -23,7 +23,6 @@ class K8sResource():
def specific_k8s_init(self):
"""Do the specific part for k8s resource when k8s object is present."""
- pass
def __repr__(self):
return self.name
@@ -32,10 +31,9 @@ class K8sResource():
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):
diff --git a/src/onaptests/steps/instantiate/k8s_profile_create.py b/src/onaptests/steps/instantiate/k8s_profile_create.py
index d3798e2..4be5684 100644
--- a/src/onaptests/steps/instantiate/k8s_profile_create.py
+++ b/src/onaptests/steps/instantiate/k8s_profile_create.py
@@ -48,7 +48,7 @@ class K8SProfileStep(BaseStep):
"""
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
@@ -152,7 +152,8 @@ class K8SProfileStep(BaseStep):
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:
@@ -179,7 +180,7 @@ class K8SProfileStep(BaseStep):
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()
diff --git a/src/onaptests/steps/instantiate/sdnc_service.py b/src/onaptests/steps/instantiate/sdnc_service.py
index 1c2437f..851902a 100644
--- a/src/onaptests/steps/instantiate/sdnc_service.py
+++ b/src/onaptests/steps/instantiate/sdnc_service.py
@@ -1,3 +1,5 @@
+import logging
+
from onapsdk.configuration import settings
from onapsdk.exceptions import APIError
from onapsdk.sdnc import VfModulePreload
@@ -59,7 +61,7 @@ class ServiceCreateStep(BaseSdncStep):
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:
@@ -107,8 +109,8 @@ class UpdateSdncService(BaseSdncStep):
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):
@@ -153,6 +155,8 @@ class GetSdncPreloadStep(BaseSdncStep):
Get preload information from SDNC over GR-API.
"""
+ __logger = logging.getLogger(__name__)
+
def __init__(self):
"""Initialize step.
@@ -180,7 +184,7 @@ class GetSdncPreloadStep(BaseSdncStep):
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):
diff --git a/src/onaptests/steps/instantiate/service_ala_carte.py b/src/onaptests/steps/instantiate/service_ala_carte.py
index 528215d..4a0a4c1 100644
--- a/src/onaptests/steps/instantiate/service_ala_carte.py
+++ b/src/onaptests/steps/instantiate/service_ala_carte.py
@@ -3,7 +3,6 @@ from uuid import uuid4
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
@@ -11,82 +10,9 @@ from onapsdk.sdc.service import Service
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):
@@ -128,7 +54,7 @@ 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
@@ -198,9 +124,9 @@ class YamlTemplateServiceAlaCarteInstantiateStep(YamlTemplateBaseStep):
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:
@@ -220,15 +146,14 @@ class YamlTemplateServiceAlaCarteInstantiateStep(YamlTemplateBaseStep):
)
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:
@@ -244,9 +169,9 @@ class YamlTemplateServiceAlaCarteInstantiateStep(YamlTemplateBaseStep):
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:
diff --git a/src/onaptests/steps/instantiate/service_macro.py b/src/onaptests/steps/instantiate/service_macro.py
index 4211cbf..a513d1e 100644
--- a/src/onaptests/steps/instantiate/service_macro.py
+++ b/src/onaptests/steps/instantiate/service_macro.py
@@ -77,7 +77,7 @@ class YamlTemplateServiceMacroInstantiateStep(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
@@ -94,8 +94,9 @@ class YamlTemplateServiceMacroInstantiateStep(YamlTemplateBaseStep):
"""
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
@@ -215,9 +216,9 @@ class YamlTemplateServiceMacroInstantiateStep(YamlTemplateBaseStep):
)
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
@@ -239,9 +240,9 @@ class YamlTemplateServiceMacroInstantiateStep(YamlTemplateBaseStep):
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:
diff --git a/src/onaptests/steps/instantiate/vf_module_ala_carte.py b/src/onaptests/steps/instantiate/vf_module_ala_carte.py
index 1682536..e148e4c 100644
--- a/src/onaptests/steps/instantiate/vf_module_ala_carte.py
+++ b/src/onaptests/steps/instantiate/vf_module_ala_carte.py
@@ -54,7 +54,7 @@ class YamlTemplateVfModuleAlaCarteInstantiateStep(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
@@ -135,9 +135,9 @@ class YamlTemplateVfModuleAlaCarteInstantiateStep(YamlTemplateBaseStep):
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:
@@ -163,7 +163,7 @@ class YamlTemplateVfModuleAlaCarteInstantiateStep(YamlTemplateBaseStep):
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()
diff --git a/src/onaptests/steps/instantiate/vl_ala_carte.py b/src/onaptests/steps/instantiate/vl_ala_carte.py
index 7c2f4d9..72dcb2f 100644
--- a/src/onaptests/steps/instantiate/vl_ala_carte.py
+++ b/src/onaptests/steps/instantiate/vl_ala_carte.py
@@ -48,7 +48,7 @@ class YamlTemplateVlAlaCarteInstantiateStep(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
@@ -89,7 +89,7 @@ class YamlTemplateVlAlaCarteInstantiateStep(YamlTemplateBaseStep):
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(
@@ -128,9 +128,9 @@ class YamlTemplateVlAlaCarteInstantiateStep(YamlTemplateBaseStep):
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:
@@ -148,7 +148,7 @@ class YamlTemplateVlAlaCarteInstantiateStep(YamlTemplateBaseStep):
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()
diff --git a/src/onaptests/steps/instantiate/vnf_ala_carte.py b/src/onaptests/steps/instantiate/vnf_ala_carte.py
index 2aa5e5e..e45f71e 100644
--- a/src/onaptests/steps/instantiate/vnf_ala_carte.py
+++ b/src/onaptests/steps/instantiate/vnf_ala_carte.py
@@ -46,7 +46,7 @@ class YamlTemplateVnfAlaCarteInstantiateStep(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
@@ -106,9 +106,9 @@ class YamlTemplateVnfAlaCarteInstantiateStep(YamlTemplateBaseStep):
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:
@@ -127,7 +127,7 @@ class YamlTemplateVnfAlaCarteInstantiateStep(YamlTemplateBaseStep):
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()
diff --git a/src/onaptests/steps/loop/clamp.py b/src/onaptests/steps/loop/clamp.py
index dc00eb1..68a2dde 100644
--- a/src/onaptests/steps/loop/clamp.py
+++ b/src/onaptests/steps/loop/clamp.py
@@ -6,20 +6,19 @@
#
# 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):
@@ -56,7 +55,7 @@ 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
@@ -73,13 +72,14 @@ class ClampStep(YamlTemplateBaseStep):
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."""
@@ -116,9 +116,9 @@ class ClampStep(YamlTemplateBaseStep):
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:
diff --git a/src/onaptests/steps/loop/instantiate_loop.py b/src/onaptests/steps/loop/instantiate_loop.py
index fab70ad..3d34731 100644
--- a/src/onaptests/steps/loop/instantiate_loop.py
+++ b/src/onaptests/steps/loop/instantiate_loop.py
@@ -9,6 +9,7 @@ from onapsdk.configuration import settings
import onaptests.utils.exceptions as onap_test_exceptions
+# pylint: disable=protected-access
class InstantiateLoop():
"""class instantiating a closed loop in clamp."""
@@ -71,8 +72,8 @@ class InstantiateLoop():
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")
diff --git a/src/onaptests/steps/onboard/cds.py b/src/onaptests/steps/onboard/cds.py
index e489145..b9cc458 100644
--- a/src/onaptests/steps/onboard/cds.py
+++ b/src/onaptests/steps/onboard/cds.py
@@ -129,7 +129,7 @@ class CbaPublishStep(CDSBaseStep):
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:
diff --git a/src/onaptests/steps/onboard/clamp.py b/src/onaptests/steps/onboard/clamp.py
index 22783c6..c2fe191 100644
--- a/src/onaptests/steps/onboard/clamp.py
+++ b/src/onaptests/steps/onboard/clamp.py
@@ -49,7 +49,7 @@ class OnboardClampStep(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
@@ -79,14 +79,14 @@ class OnboardClampStep(YamlTemplateBaseStep):
# 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")
diff --git a/src/onaptests/steps/onboard/cps.py b/src/onaptests/steps/onboard/cps.py
index f5820c7..04471c2 100644
--- a/src/onaptests/steps/onboard/cps.py
+++ b/src/onaptests/steps/onboard/cps.py
@@ -254,6 +254,8 @@ class CheckPostgressDataBaseConnectionStep(CpsBaseStep):
def __init__(self) -> None:
"""Initialize step."""
super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.login = None
+ self.password = None
@property
def description(self) -> str:
@@ -261,6 +263,8 @@ class CheckPostgressDataBaseConnectionStep(CpsBaseStep):
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:
@@ -286,6 +290,8 @@ class CheckPostgressDataBaseConnectionStep(CpsBaseStep):
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 = {
diff --git a/src/onaptests/steps/onboard/service.py b/src/onaptests/steps/onboard/service.py
index bcf153f..d2391b3 100644
--- a/src/onaptests/steps/onboard/service.py
+++ b/src/onaptests/steps/onboard/service.py
@@ -147,9 +147,9 @@ class YamlTemplateServiceOnboardStep(YamlTemplateBaseStep):
"""
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
@@ -166,7 +166,8 @@ class YamlTemplateServiceOnboardStep(YamlTemplateBaseStep):
"""
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
diff --git a/src/onaptests/steps/onboard/vf.py b/src/onaptests/steps/onboard/vf.py
index 3fc7443..0c68835 100644
--- a/src/onaptests/steps/onboard/vf.py
+++ b/src/onaptests/steps/onboard/vf.py
@@ -104,8 +104,7 @@ class YamlTemplateVfOnboardStep(YamlTemplateBaseStep):
"""
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:
@@ -128,10 +127,10 @@ class YamlTemplateVfOnboardStep(YamlTemplateBaseStep):
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():
diff --git a/src/onaptests/steps/onboard/vsp.py b/src/onaptests/steps/onboard/vsp.py
index 06e0fa2..36e78df 100644
--- a/src/onaptests/steps/onboard/vsp.py
+++ b/src/onaptests/steps/onboard/vsp.py
@@ -48,10 +48,11 @@ class VspOnboardStep(BaseStep):
"""
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):
@@ -103,8 +104,7 @@ class YamlTemplateVspOnboardStep(YamlTemplateBaseStep):
"""
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:
diff --git a/src/onaptests/steps/reports_collection.py b/src/onaptests/steps/reports_collection.py
index 99a6ef2..dfdc09a 100644
--- a/src/onaptests/steps/reports_collection.py
+++ b/src/onaptests/steps/reports_collection.py
@@ -28,9 +28,10 @@ class Report:
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.
@@ -65,16 +66,18 @@ class ReportsCollection:
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']),
@@ -104,5 +107,5 @@ class ReportsCollection:
]
}
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)
diff --git a/src/onaptests/steps/simulator/cds_mockserver.py b/src/onaptests/steps/simulator/cds_mockserver.py
index 75d0767..8773248 100644
--- a/src/onaptests/steps/simulator/cds_mockserver.py
+++ b/src/onaptests/steps/simulator/cds_mockserver.py
@@ -54,9 +54,10 @@ class CdsMockserverCnfConfigureStep(BaseStep):
"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
diff --git a/src/onaptests/steps/simulator/pnf_simulator_cnf/pnf_register.py b/src/onaptests/steps/simulator/pnf_simulator_cnf/pnf_register.py
index 5adba2c..0ecc2ca 100644
--- a/src/onaptests/steps/simulator/pnf_simulator_cnf/pnf_register.py
+++ b/src/onaptests/steps/simulator/pnf_simulator_cnf/pnf_register.py
@@ -5,13 +5,14 @@ import time
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):
@@ -64,9 +65,9 @@ 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.
@@ -92,9 +93,9 @@ class PnfSimulatorCnfRegisterStep(BaseStep):
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:
@@ -131,7 +132,8 @@ class PnfSimulatorCnfRegisterStep(BaseStep):
}
}
}
- }
+ },
+ timeout=settings.DEFAULT_REQUEST_TIMEOUT
)
response.raise_for_status()
registered_successfully = True
diff --git a/src/onaptests/steps/wrapper/helm_charts.py b/src/onaptests/steps/wrapper/helm_charts.py
index f745150..874e1ff 100644
--- a/src/onaptests/steps/wrapper/helm_charts.py
+++ b/src/onaptests/steps/wrapper/helm_charts.py
@@ -29,7 +29,7 @@ class HelmChartStep(BaseStep):
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."
@@ -80,6 +80,7 @@ class HelmChartStep(BaseStep):
msg = "Error during helm release installation."
raise EnvironmentPreparationException(msg) from err
+ @BaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Uninstall helm release."""
try:
diff --git a/src/onaptests/steps/wrapper/start.py b/src/onaptests/steps/wrapper/start.py
index 8b950da..e08f108 100644
--- a/src/onaptests/steps/wrapper/start.py
+++ b/src/onaptests/steps/wrapper/start.py
@@ -1,6 +1,7 @@
"""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
@@ -62,4 +63,5 @@ class SimulatorStartStep(BaseStep):
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)
diff --git a/src/onaptests/utils/exceptions.py b/src/onaptests/utils/exceptions.py
index 978e628..aeae9d1 100644
--- a/src/onaptests/utils/exceptions.py
+++ b/src/onaptests/utils/exceptions.py
@@ -6,7 +6,7 @@
# 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):
diff --git a/src/onaptests/utils/resources.py b/src/onaptests/utils/resources.py
index 46c6c6e..71e56d6 100644
--- a/src/onaptests/utils/resources.py
+++ b/src/onaptests/utils/resources.py
@@ -5,6 +5,14 @@ import onaptests
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)
diff --git a/tox.ini b/tox.ini
index 78bdbb8..2ce589d 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,6 +1,6 @@
[tox]
minversion = 3.2.0
-envlist = json,yaml,py,rst,md,pylama
+envlist = json,yaml,md,pylama
skipsdist = true
requires = pip >= 8
@@ -57,6 +57,19 @@ commands =
/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