smo-install/test/pythonsdk/src/oransdk/policy/clamp.py

   1 #!/usr/bin/env python3
   2 ###
   3 # ============LICENSE_START=======================================================
   4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
   5 # ================================================================================
   6 # Copyright (C) 2022 AT&T Intellectual Property. All rights
   7 #                             reserved.
   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
  12 #
  13 # http://www.apache.org/licenses/LICENSE-2.0
  14 #
  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 # ===================================================================
  22 #
  23 ###
  24 """Onap Policy Clamp Tosca Template module."""
  25 import time
  26 from typing import Dict
  27 from onapsdk.clamp.clamp_element import Clamp
  28 from onapsdk.exceptions import RequestError
  29
  30 class ClampToscaTemplate(Clamp):
  31     """Onap Policy Clamp Tosca Template class."""
  32
  33     header = {"Accept": "application/json", "Content-Type": "application/json"}
  34
  35     def __init__(self, basic_auth: Dict[str, str]) -> None:
  36         """
  37         Initialize loop instance object.
  38
  39         Args:
  40             basic_auth : basic auth
  41
  42         """
  43         super().__init__()
  44         self.basic_auth = basic_auth
  45
  46     def get_template_instance(self) -> dict:
  47         """
  48         Get tosca template instance.
  49
  50         Returns:
  51             the tosca template instance
  52         """
  53         url = f"{self.base_url()}/acm/getToscaInstantiation"
  54         template_instance = self.send_message_json('GET',
  55                                                    'Get tosca template instance',
  56                                                    url,
  57                                                    basic_auth=self.basic_auth)
  58
  59         return template_instance
  60
  61     def upload_commission(self, tosca_template) -> dict:
  62         """
  63         Upload Tosca to commissioning.
  64
  65         Args:
  66             tosca_template: the tosca template yaml
  67         Returns:
  68             the response of the uploading action
  69
  70         """
  71         url = f"{self.base_url()}/acm/commissionToscaTemplate"
  72         response = self.send_message_json('POST',
  73                                           'Upload Tosca to commissioning',
  74                                           url,
  75                                           data=tosca_template,
  76                                           headers=self.header,
  77                                           basic_auth=self.basic_auth)
  78         return response
  79
  80     def create_instance(self, tosca_instance_properties) -> dict:
  81         """
  82         Create Tosca instance.
  83
  84         Args:
  85             tosca_instance_properties (str): the tosca template properties
  86         Returns:
  87             the response of the creation action
  88         """
  89         url = f"{self.base_url()}/acm/postToscaInstanceProperties"
  90         response = self.send_message_json('POST',
  91                                           'Create Tosca instance',
  92                                           url,
  93                                           data=tosca_instance_properties,
  94                                           headers=self.header,
  95                                           basic_auth=self.basic_auth)
  96         return response
  97
  98     def get_template_instance_status(self, name, version) -> dict:
  99         """
 100         Get tosca template instance status.
 101
 102         Args:
 103             name (str): the name of the template instance
 104             version (str): the version of the template instance
 105         Returns:
 106             the template instance
 107         """
 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',
 111                                                    url,
 112                                                    basic_auth=self.basic_auth)
 113
 114         return template_instance
 115
 116     def change_instance_status(self, new_status, name, version) -> str:
 117         """
 118         Update tosca instance status.
 119
 120         Args:
 121             new_status (str): the new instance status
 122             name (str): the new instance name
 123             version (str): the new instance version
 124         Returns:
 125             the updated template instance
 126         """
 127         body = '{"orderedState":"' + new_status + '","controlLoopIdentifierList":[{"name":"' + name + '","version":"' + version + '"}]}'
 128         url = f"{self.base_url()}/acm/putToscaInstantiationStateChange"
 129         try:
 130             response = self.send_message_json('PUT',
 131                                               'Update tosca instance status',
 132                                               url,
 133                                               data=body,
 134                                               headers=self.header,
 135                                               basic_auth=self.basic_auth)
 136         except RequestError:
 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.")
 138
 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
 141         time.sleep(2)
 142         response = self.get_template_instance()
 143         return response["automationCompositionList"][0]["orderedState"]
 144
 145     def verify_instance_status(self, new_status):
 146         """
 147         Verify whether the instance changed to the new status.
 148
 149         Args:
 150             new_status : the new status of the instance
 151         Returns:
 152             the boolean value indicating whether status changed successfully
 153         """
 154         response = self.get_template_instance()
 155         if response["automationCompositionList"][0]["state"] == new_status:
 156             return True
 157         return False
 158
 159     def delete_template_instance(self, name: str, version: str) -> dict:
 160         """
 161         Delete the tosca instance.
 162
 163         Args:
 164             name (str): the instance name.
 165             version (str): the instance version.
 166         Returns:
 167             the response of the deletion action
 168         """
 169         url = f"{self.base_url()}/acm/deleteToscaInstanceProperties?name={name}&version={version}"
 170         response = self.send_message_json('DELETE',
 171                                           'Delete the tosca instance',
 172                                           url,
 173                                           headers=self.header,
 174                                           basic_auth=self.basic_auth)
 175         return response
 176
 177     def decommission_template(self, name: str, version: str) -> dict:
 178         """
 179         Decommission the tosca template.
 180
 181         Args:
 182             name (str): the tosca template name.
 183             version (str): the tosca template version.
 184         Returns:
 185             the response of the decommission action
 186         """
 187         url = f"{self.base_url()}/acm/decommissionToscaTemplate?name={name}&version={version}"
 188         response = self.send_message_json('DELETE',
 189                                           'Decommission the tosca template',
 190                                           url,
 191                                           headers=self.header,
 192                                           basic_auth=self.basic_auth)
 193         return response