aboutsummaryrefslogtreecommitdiffstats
path: root/src/onaptests/steps/loop
diff options
context:
space:
mode:
authorMichal Jagiello <michal.jagiello@t-mobile.pl>2023-07-17 13:30:55 +0000
committerMichal Jagiello <michal.jagiello@t-mobile.pl>2023-07-17 13:50:59 +0000
commit939f1933c8635a18de428c696c97cc5d8600abff (patch)
treea58cb9342f1a9d2cec80fc6a792eae0daae7bc49 /src/onaptests/steps/loop
parentb1e1a5fa99e29b3ffc2bc48d11b163e28cbb7744 (diff)
Add pylama into checks
Checks code with pylama Issue-ID: INT-2222 Signed-off-by: Michal Jagiello <michal.jagiello@t-mobile.pl> Change-Id: If275ef90f21e3049abe72bcf373473b40846005c
Diffstat (limited to 'src/onaptests/steps/loop')
-rw-r--r--src/onaptests/steps/loop/clamp.py15
-rw-r--r--src/onaptests/steps/loop/instantiate_loop.py13
2 files changed, 13 insertions, 15 deletions
diff --git a/src/onaptests/steps/loop/clamp.py b/src/onaptests/steps/loop/clamp.py
index c614857..502a7f0 100644
--- a/src/onaptests/steps/loop/clamp.py
+++ b/src/onaptests/steps/loop/clamp.py
@@ -80,13 +80,12 @@ class ClampStep(YamlTemplateBaseStep):
else:
return self.parent.service_name
-
def check(self, operational_policies: list, is_template: bool = False) -> str:
"""Check CLAMP requirements to create a loop."""
self._logger.info("Check operational policy")
for policy in operational_policies:
exist = Clamp.check_policies(policy_name=policy["name"],
- req_policies=30)# 30 required policy
+ req_policies=30) # 30 required policy
self._logger.info("Operational policy found.")
if not exist:
raise ValueError("Couldn't load the policy %s", policy)
@@ -109,14 +108,14 @@ class ClampStep(YamlTemplateBaseStep):
def loop_counter(self, action: str) -> None:
""" Count number of loop instances."""
- if action == "plus":
+ if action == "plus":
self.count += 1
- if action == "minus":
+ if action == "minus":
self.count -= 1
@YamlTemplateBaseStep.store_state
def execute(self):
- super().execute() # TODO work only the 1st time, not if already onboarded
+ super().execute() # TODO work only the 1st time, not if already onboarded
# Before instantiating, be sure that the service has been distributed
service = Service(self.service_name)
@@ -128,8 +127,8 @@ class ClampStep(YamlTemplateBaseStep):
distribution_completed = service.distributed
if distribution_completed is True:
self._logger.info(
- "Service Distribution for %s is sucessfully finished",
- service.name)
+ "Service Distribution for %s is sucessfully finished",
+ service.name)
break
self._logger.info(
"Service Distribution for %s ongoing, Wait for 60 s",
@@ -139,7 +138,7 @@ class ClampStep(YamlTemplateBaseStep):
if distribution_completed is False:
self._logger.error(
- "Service Distribution for %s failed !!",service.name)
+ "Service Distribution for %s failed !!", service.name)
raise onap_test_exceptions.ServiceDistributionException
# time to wait for template load in CLAMP
diff --git a/src/onaptests/steps/loop/instantiate_loop.py b/src/onaptests/steps/loop/instantiate_loop.py
index 07586ac..fab70ad 100644
--- a/src/onaptests/steps/loop/instantiate_loop.py
+++ b/src/onaptests/steps/loop/instantiate_loop.py
@@ -13,9 +13,9 @@ 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.template = template
+ self.loop_name = loop_name
+ self.operational_policies = operational_policies
self._logger: logging.Logger = logging.getLogger("")
logging.config.dictConfig(settings.LOG_CONFIG)
@@ -30,7 +30,6 @@ class InstantiateLoop():
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 *******")
@@ -38,8 +37,8 @@ class InstantiateLoop():
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
+ # 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":
@@ -53,7 +52,7 @@ class InstantiateLoop():
self._logger.info("******** SUBMIT POLICIES TO PE *******")
submit = loop.act_on_loop_policy(loop.submit)
self._logger.info("******** CHECK POLICIES SUBMITION *******")
- if submit :
+ if submit:
self._logger.info("Policies successfully submited to PE")
else: