3 # ============LICENSE_START=======================================================
4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
5 # ================================================================================
6 # Copyright (C) 2022 AT&T Intellectual Property. All rights
8 # ================================================================================
9 # Licensed under the Apache License, Version 2.0 (the "License");
10 # you may not use this file except in compliance with the License.
11 # You may obtain a copy of the License at
13 # http://www.apache.org/licenses/LICENSE-2.0
15 # Unless required by applicable law or agreed to in writing, software
16 # distributed under the License is distributed on an "AS IS" BASIS,
17 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
18 # See the License for the specific language governing permissions and
19 # limitations under the License.
20 # ============LICENSE_END============================================
21 # ===================================================================
24 """Onap Policy Clamp Tosca Template module."""
26 from typing import Dict
27 from onapsdk.clamp.clamp_element import Clamp
28 from onapsdk.exceptions import RequestError
30 class ClampToscaTemplate(Clamp):
31 """Onap Policy Clamp Tosca Template class."""
33 header = {"Accept": "application/json", "Content-Type": "application/json"}
35 def __init__(self, basic_auth: Dict[str, str]) -> None:
37 Initialize loop instance object.
40 basic_auth : basic auth
44 self.basic_auth = basic_auth
46 def get_template_instance(self) -> dict:
48 Get tosca template instance.
51 the tosca template instance
53 url = f"{self.base_url()}/acm/getToscaInstantiation"
54 template_instance = self.send_message_json('GET',
55 'Get tosca template instance',
57 basic_auth=self.basic_auth)
59 return template_instance
61 def upload_commission(self, tosca_template) -> dict:
63 Upload Tosca to commissioning.
66 tosca_template: the tosca template yaml
68 the response of the uploading action
71 url = f"{self.base_url()}/acm/commissionToscaTemplate"
72 response = self.send_message_json('POST',
73 'Upload Tosca to commissioning',
77 basic_auth=self.basic_auth)
80 def create_instance(self, tosca_instance_properties) -> dict:
82 Create Tosca instance.
85 tosca_instance_properties (str): the tosca template properties
87 the response of the creation action
89 url = f"{self.base_url()}/acm/postToscaInstanceProperties"
90 response = self.send_message_json('POST',
91 'Create Tosca instance',
93 data=tosca_instance_properties,
95 basic_auth=self.basic_auth)
98 def get_template_instance_status(self, name, version) -> dict:
100 Get tosca template instance status.
103 name (str): the name of the template instance
104 version (str): the version of the template instance
106 the template instance
108 url = f"{self.base_url()}/acm/getInstantiationOrderState?name={name}&version={version}"
109 template_instance = self.send_message_json('GET',
110 'Get tosca template instance',
112 basic_auth=self.basic_auth)
114 return template_instance
116 def change_instance_status(self, new_status, name, version) -> str:
118 Update tosca instance status.
121 new_status (str): the new instance status
122 name (str): the new instance name
123 version (str): the new instance version
125 the updated template instance
127 body = '{"orderedState":"' + new_status + '","automationCompositionIdentifierList":[{"name":"' + name + '","version":"' + version + '"}]}'
128 url = f"{self.base_url()}/acm/putToscaInstantiationStateChange"
130 response = self.send_message_json('PUT',
131 'Update tosca instance status',
135 basic_auth=self.basic_auth)
137 self._logger.error("Change Instance Status request returned failed. Will query the instance status to double check whether the request is successful or not.")
139 # There's a bug in Clamp code, sometimes it returned 500, but actually the status has been changed successfully
140 # Thus we verify the status to determine whether it was successful or not
142 response = self.get_template_instance()
143 return response["automationCompositionList"][0]["orderedState"]
145 def verify_instance_status(self, new_status):
147 Verify whether the instance changed to the new status.
150 new_status : the new status of the instance
152 the boolean value indicating whether status changed successfully
154 response = self.get_template_instance()
155 if response["automationCompositionList"][0]["state"] == new_status:
159 def delete_template_instance(self, name: str, version: str) -> dict:
161 Delete the tosca instance.
164 name (str): the instance name.
165 version (str): the instance version.
167 the response of the deletion action
169 url = f"{self.base_url()}/acm/deleteToscaInstanceProperties?name={name}&version={version}"
170 response = self.send_message_json('DELETE',
171 'Delete the tosca instance',
174 basic_auth=self.basic_auth)
177 def decommission_template(self, name: str, version: str) -> dict:
179 Decommission the tosca template.
182 name (str): the tosca template name.
183 version (str): the tosca template version.
185 the response of the decommission action
187 url = f"{self.base_url()}/acm/decommissionToscaTemplate?name={name}&version={version}"
188 response = self.send_message_json('DELETE',
189 'Decommission the tosca template',
192 basic_auth=self.basic_auth)