smo-install/test/pythonsdk/src/oransdk/enrichmentservice/enrichmentservice.py

   1 #!/usr/bin/env python3
   2 ###
   3 # ============LICENSE_START=======================================================
   4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
   5 # ================================================================================
   6 # Copyright (C) 2021-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 """Oran Enrichment Service module."""
  25
  26 from onapsdk.onap_service import OnapService
  27 from oransdk.configuration import settings
  28
  29 class EnrichmentService(OnapService):
  30     """Enrichment Service library."""
  31
  32     base_url = settings.EMS_URL
  33     header = {"Content-Type": "application/json"}
  34
  35     @classmethod
  36     def check_status(cls) -> str:
  37         """
  38         Get the status of the EnrichmentService component.
  39
  40         Returns:
  41             the status of the EnrichmentService component
  42
  43         """
  44         url = f"{cls.base_url}/status"
  45         status = cls.send_message('GET',
  46                                   'Get EMS status',
  47                                   url)
  48         return status
  49
  50     @classmethod
  51     def get_eitypes(cls) -> str:
  52         """
  53         Get all the EiTypes.
  54
  55         Returns:
  56             the list of EiTypes
  57
  58         """
  59         url = f"{cls.base_url}/data-producer/v1/info-types"
  60         eitypes = cls.send_message('GET',
  61                                    'Get all the EiTypes',
  62                                    url,
  63                                    headers=cls.header)
  64         return eitypes
  65
  66     @classmethod
  67     def get_eitype_individual(cls, eitype_name) -> str:
  68         """
  69         Get individual EiType.
  70
  71         Args:
  72            eitype_name: the EiType name
  73
  74         Returns:
  75             the details of the EiType
  76
  77         """
  78         url = f"{cls.base_url}/data-producer/v1/info-types/{eitype_name}"
  79         eitype = cls.send_message('GET',
  80                                   'Get individual EiType',
  81                                   url,
  82                                   headers=cls.header)
  83         return eitype
  84
  85     @classmethod
  86     def get_eiproducers(cls) -> str:
  87         """
  88         Get all the EiProducers.
  89
  90         Returns:
  91             the list of EiProducers
  92
  93         """
  94         url = f"{cls.base_url}/data-producer/v1/info-producers"
  95         eitypes = cls.send_message('GET',
  96                                    'Get all the EiProducers',
  97                                    url,
  98                                    headers=cls.header)
  99         return eitypes
 100
 101     @classmethod
 102     def get_eiproducer_individual(cls, producer) -> str:
 103         """
 104         Get individual EiProducer.
 105
 106         Args:
 107            type: the EiProducer name
 108
 109         Returns:
 110             the details of the EiProducer
 111
 112         """
 113         url = f"{cls.base_url}/data-producer/v1/info-producers/{producer}"
 114         eiproducer = cls.send_message('GET',
 115                                       'Get individual EiProducer',
 116                                       url,
 117                                       headers=cls.header)
 118         return eiproducer
 119
 120     @classmethod
 121     def get_eiproducer_status(cls, producer) -> str:
 122         """
 123         Get the status of EiProducer.
 124
 125         Args:
 126            type: the EiProducer name
 127
 128         Returns:
 129             the status of the EiProducer
 130
 131         """
 132         url = f"{cls.base_url}/data-producer/v1/info-producers/{producer}/status"
 133         status = cls.send_message('GET',
 134                                   'Get the status of EiProducer',
 135                                   url,
 136                                   headers=cls.header)
 137         return status
 138
 139     @classmethod
 140     def get_eijobs(cls) -> str:
 141         """
 142         Get all the EiJobs.
 143
 144         Returns:
 145             the list of EiJobs
 146
 147         """
 148         url = f"{cls.base_url}/A1-EI/v1/eijobs"
 149         eijobs = cls.send_message('GET',
 150                                   'Get all the EiJobs',
 151                                   url,
 152                                   headers=cls.header)
 153         return eijobs
 154
 155     @classmethod
 156     def get_eijob_individual(cls, job) -> str:
 157         """
 158         Get individual EiJob.
 159
 160         Args:
 161            type: the EiJob name
 162
 163         Returns:
 164             the details of the EiJob
 165
 166         """
 167         url = f"{cls.base_url}/A1-EI/v1/eijobs/{job}"
 168         eijob = cls.send_message('GET',
 169                                  'Get individual EiJob',
 170                                  url,
 171                                  headers=cls.header)
 172         return eijob
 173
 174     @classmethod
 175     def create_eitype(cls,
 176                       type_name,
 177                       type_data) -> None:
 178         """
 179         Create EiType.
 180
 181         Args:
 182            type: the EiType name
 183            type_data: the EiType data to create, in binary format
 184
 185         """
 186         url = f"{cls.base_url}/data-producer/v1/info-types/{type_name}"
 187         cls.send_message('PUT',
 188                          'Create EiType',
 189                          url,
 190                          data=type_data,
 191                          headers=cls.header)
 192
 193     @classmethod
 194     def create_eiproducer(cls,
 195                           producer,
 196                           producer_data) -> None:
 197         """
 198         Create EiProducer.
 199
 200         Args:
 201            producer: the EiProducer name
 202            producer_data: the EiProducer data to create, in binary format
 203
 204         """
 205         url = f"{cls.base_url}/data-producer/v1/info-producers/{producer}"
 206         cls.send_message('PUT',
 207                          'Create EiProducer',
 208                          url,
 209                          data=producer_data,
 210                          headers=cls.header)
 211
 212     @classmethod
 213     def create_eijob(cls,
 214                      job,
 215                      job_data) -> None:
 216         """
 217         Create EiJob.
 218
 219         Args:
 220            job: the EiJob name
 221            job_data: the EiJob data to create, in binary format
 222
 223         """
 224         url = f"{cls.base_url}/A1-EI/v1/eijobs/{job}"
 225         cls.send_message('PUT',
 226                          'Create EiJob',
 227                          url,
 228                          data=job_data,
 229                          headers=cls.header)