aboutsummaryrefslogtreecommitdiffstats
path: root/src/onaptests/steps/loop/instantiate_loop.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/onaptests/steps/loop/instantiate_loop.py')
-rw-r--r--src/onaptests/steps/loop/instantiate_loop.py87
1 files changed, 0 insertions, 87 deletions
diff --git a/src/onaptests/steps/loop/instantiate_loop.py b/src/onaptests/steps/loop/instantiate_loop.py
deleted file mode 100644
index 9aeefca..0000000
--- a/src/onaptests/steps/loop/instantiate_loop.py
+++ /dev/null
@@ -1,87 +0,0 @@
-#!/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
-
-import onaptests.utils.exceptions as onap_test_exceptions
-
-
-# pylint: disable=protected-access
-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 onap_test_exceptions.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 onap_test_exceptions.DcaeException
-
- def instantiate_loop(self):
- """Instantiate the control loop."""
- loop = LoopInstance(template=self.template,
- name=self.loop_name,
- 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")
-
- self.add_policies(loop=loop)
- self.configure_policies(loop=loop)
- self.deploy(loop=loop)
-
- loop.details = loop._update_loop_details()
- return loop