aboutsummaryrefslogtreecommitdiffstats
path: root/src/onaptests/scenario/scenario_base.py
blob: ca6d898db2b4bc01704f77871c1db38290744000 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
import logging
import time

from onapsdk.configuration import settings
from onapsdk.exceptions import SDKException, SettingsError
from xtesting.core import testcase

from onaptests.steps.base import BaseStep, YamlTemplateBaseStep
from onaptests.utils.exceptions import (OnapTestException,
                                        TestConfigurationException)


class ScenarioBase(testcase.TestCase):
    """Scenario base class."""

    __logger = logging.getLogger()

    def __init__(self, case_name_override, **kwargs):
        """Init base scenario."""
        if "case_name" not in kwargs:
            kwargs["case_name"] = case_name_override
        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_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, **kwargs):
        """Run scenario and cleanup resources afterwards"""
        self.start_time = time.time()
        try:
            for test_phase in (self.test.execute, self.test.cleanup):
                phase_name = test_phase.__name__
                try:
                    if (phase_name == "cleanup" and settings.CLEANUP_FLAG and
                            settings.CLEANUP_ACTIVITY_TIMER > 0):
                        time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
                    self.__logger.info("%s %s Phase Started",
                                       self.scenario_name, phase_name.title())
                    test_phase()
                    self.result += 50
                except OnapTestException as exc:
                    self.__logger.exception("%s on %s", exc.error_message, phase_name)
                except SDKException:
                    self.__logger.exception("SDK Exception on %s", phase_name)
                except Exception as e:
                    self.__logger.exception("General Exception on %s", phase_name)
                    if self.general_exception:
                        e = ExceptionGroup("General Exceptions", [self.general_exception, e])  # noqa
                    self.general_exception = e
        finally:
            self.stop_time = time.time()
            self.__logger.info(f"{self.scenario_name} Execution {self.result}% Completed")
        if self.general_exception:
            raise self.general_exception

    def clean(self):
        """Clean Additional resources if needed."""
        self.__logger.info("Generate %s Test report", self.scenario_name)
        self.test.reports_collection.generate_report()

    def validate(self):
        """Validate implementation of the scenario."""

        self._validate_service_details()
        self.test.validate_step_implementation()
        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."""
        super().__init__(cleanup=cleanup)

    @BaseStep.store_state
    def execute(self) -> None:
        super().execute()


class YamlTemplateBaseScenarioStep(YamlTemplateBaseStep, BaseScenarioStep):
    """Main scenario yaml template step that has no own execution method."""

    def __init__(self, cleanup=False):
        """Initialize YamlTemplateBaseScenarioStep step."""
        super().__init__(cleanup=cleanup)