#
###
"""Onap Policy Clamp Tosca Template module."""
-
+import time
from typing import Dict
from onapsdk.clamp.clamp_element import Clamp
+from onapsdk.exceptions import RequestError
class ClampToscaTemplate(Clamp):
"""Onap Policy Clamp Tosca Template class."""
Returns:
the tosca template instance
"""
- url = f"{self.base_url()}/toscaControlLoop/getToscaInstantiation"
+ url = f"{self.base_url()}/acm/getToscaInstantiation"
template_instance = self.send_message_json('GET',
'Get tosca template instance',
url,
the response of the uploading action
"""
- url = f"{self.base_url()}/toscaControlLoop/commissionToscaTemplate"
+ url = f"{self.base_url()}/acm/commissionToscaTemplate"
response = self.send_message_json('POST',
'Upload Tosca to commissioning',
url,
Returns:
the response of the creation action
"""
- url = f"{self.base_url()}/toscaControlLoop/postToscaInstanceProperties"
+ url = f"{self.base_url()}/acm/postToscaInstanceProperties"
response = self.send_message_json('POST',
'Create Tosca instance',
url,
Returns:
the template instance
"""
- url = f"{self.base_url()}/toscaControlLoop/getInstantiationOrderState?name={name}&version={version}"
+ url = f"{self.base_url()}/acm/getInstantiationOrderState?name={name}&version={version}"
template_instance = self.send_message_json('GET',
'Get tosca template instance',
url,
return template_instance
- def change_instance_status(self, new_status, name, version) -> dict:
+ def change_instance_status(self, new_status, name, version) -> str:
"""
Update tosca instance status.
the updated template instance
"""
body = '{"orderedState":"' + new_status + '","controlLoopIdentifierList":[{"name":"' + name + '","version":"' + version + '"}]}'
- url = f"{self.base_url()}/toscaControlLoop/putToscaInstantiationStateChange"
- response = self.send_message_json('PUT',
- 'Update tosca instance status',
- url,
- data=body,
- headers=self.header,
- basic_auth=self.basic_auth)
- return response
+ url = f"{self.base_url()}/acm/putToscaInstantiationStateChange"
+ try:
+ response = self.send_message_json('PUT',
+ 'Update tosca instance status',
+ url,
+ data=body,
+ headers=self.header,
+ basic_auth=self.basic_auth)
+ except RequestError:
+ self._logger.error("Change Instance Status request returned failed. Will query the instance status to double check whether the request is successful or not.")
+
+ # There's a bug in Clamp code, sometimes it returned 500, but actually the status has been changed successfully
+ # Thus we verify the status to determine whether it was successful or not
+ time.sleep(2)
+ response = self.get_template_instance()
+ return response["automationCompositionList"][0]["orderedState"]
+
+ def verify_instance_status(self, new_status):
+ """
+ Verify whether the instance changed to the new status.
+
+ Args:
+ new_status : the new status of the instance
+ Returns:
+ the boolean value indicating whether status changed successfully
+ """
+ response = self.get_template_instance()
+ if response["automationCompositionList"][0]["state"] == new_status:
+ return True
+ return False
def delete_template_instance(self, name: str, version: str) -> dict:
"""
Returns:
the response of the deletion action
"""
- url = f"{self.base_url()}/toscaControlLoop/deleteToscaInstanceProperties?name={name}&version={version}"
+ url = f"{self.base_url()}/acm/deleteToscaInstanceProperties?name={name}&version={version}"
response = self.send_message_json('DELETE',
'Delete the tosca instance',
url,
Returns:
the response of the decommission action
"""
- url = f"{self.base_url()}/toscaControlLoop/decommissionToscaTemplate?name={name}&version={version}"
+ url = f"{self.base_url()}/acm/decommissionToscaTemplate?name={name}&version={version}"
response = self.send_message_json('DELETE',
'Decommission the tosca template',
url,