aboutsummaryrefslogtreecommitdiffstats
path: root/src/onaptests/steps/loop/instantiate_loop.py
blob: 01eada539106b10d44e144d3f183787376aad244 (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
#!/usr/bin/python
# http://www.apache.org/licenses/LICENSE-2.0
"""Instantiation class."""
import logging
import logging.config

from onapsdk.clamp.loop_instance import LoopInstance
from onapsdk.configuration import settings
from onaptests.utils.exceptions import OnapTestException


class InstantiateLoop():
    """class instantiating a closed loop in clamp."""

    def __init__(self, template: str, loop_name: str, operational_policies: list):
        self.template=template
        self.loop_name=loop_name
        self.operational_policies=operational_policies

        self._logger: logging.Logger = logging.getLogger("")
        logging.config.dictConfig(settings.LOG_CONFIG)

    def add_policies(self, loop: LoopInstance) -> None:
        """add necessary wanted operational policies."""
        for policy in self.operational_policies:
            self._logger.info("******** ADD OPERATIONAL POLICY %s *******", policy["name"])
            added = loop.add_operational_policy(policy_type=policy["policy_type"],
                                                policy_version=policy["policy_version"])
            if not added:
                self._logger.error("an error occured while adding an operational policy")
            self._logger.info("ADD OPERATION SUCCESSFULY DONE")


    def configure_policies(self, loop: LoopInstance) -> None:
        """Configure all policies."""
        self._logger.info("******** UPDATE MICROSERVICE POLICY *******")
        loop._update_loop_details()
        loop.update_microservice_policy()
        self._logger.info("******** UPDATE OPERATIONAL POLICIES CONFIG *******")
        for policy in self.operational_policies:
            #loop.add_op_policy_config(loop.LoopInstance.__dict__[policy["config_function"]])
            #possible configurations for the moment
            if policy["name"] == "MinMax":
                loop.add_op_policy_config(loop.add_minmax_config)
            if policy["name"] == "Drools":
                loop.add_op_policy_config(loop.add_drools_conf)
            if policy["name"] == "FrequencyLimiter":
                loop.add_op_policy_config(loop.add_frequency_limiter)
        self._logger.info("Policies are well configured")

    def deploy(self, loop: LoopInstance) -> None:
        """Deploy closed loop."""
        self._logger.info("******** SUBMIT POLICIES TO PE *******")
        submit = loop.act_on_loop_policy(loop.submit)
        self._logger.info("******** CHECK POLICIES SUBMITION *******")
        if submit :
            self._logger.info("Policies successfully submited to PE")

        else:
            self._logger.error("An error occured while submitting the loop instance")
            raise OnapTestException.PolicyException
        self._logger.info("******** DEPLOY LOOP INSTANCE *******")
        deploy = loop.deploy_microservice_to_dcae()
        if deploy:
            self._logger.info("Loop instance %s successfully deployed on DCAE !!", self.loop_name)
        else:
            self._logger.error("An error occured while deploying the loop instance")
            raise OnapTestException.DcaeException

    def instantiate_loop(self):
        """Instantiate the control loop."""
        loop = LoopInstance(template=self.template,
                            name=self.loop_name,
                            details={})
        details = loop.create()
        if details:
            self._logger.info("Loop instance %s successfully created !!", self.loop_name)
        else:
            self._logger.error("An error occured while creating the loop instance")

        self.add_policies(loop=loop)
        self.configure_policies(loop=loop)
        self.deploy(loop=loop)

        loop.details = loop._update_loop_details()
        return loop