solution/dev/client-scripts-ves-v7/globalVesEventEmitter.py

   1 #!/usr/bin/env python
   2 ################################################################################
   3 # Copyright 2021 highstreet technologies GmbH
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the 'License');
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #     https://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an 'AS IS' BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 #
  17
  18 ################################################################################
  19 # A selection of common methods
  20
  21 import datetime
  22 import json
  23 import requests
  24 import os
  25 import socket
  26 import sys
  27 import yaml
  28 from pathlib import Path
  29
  30 def sendVesEvent(data):
  31     url = data['config']['vesEndpoint']['url']
  32     username = data['config']['vesEndpoint']['username']
  33     password = data['config']['vesEndpoint']['password']
  34     headers = {'content-type': 'application/json'}
  35     verify = data['config']['vesEndpoint']['verify']
  36     try:
  37       response = requests.post(url, json=data['body'], auth=(
  38           username, password), headers=headers, verify=verify)
  39     except requests.exceptions.Timeout:
  40       sys.exit('HTTP request failed, please check you internet connection.')
  41     except requests.exceptions.TooManyRedirects:
  42       sys.exit('HTTP request failed, please check your proxy settings.')
  43     except requests.exceptions.RequestException as e:
  44       # catastrophic error. bail.
  45       raise SystemExit(e)
  46
  47     if response.status_code >= 200 and response.status_code < 300:
  48       print(response)
  49     else:
  50       sys.exit('Reading VES "stndDefined" message template failed.')
  51
  52 def sendHttpGet(url):
  53     try:
  54       response = requests.get(url)
  55     except requests.exceptions.Timeout:
  56       sys.exit('HTTP request failed, please check you internet connection.')
  57     except requests.exceptions.TooManyRedirects:
  58       sys.exit('HTTP request failed, please check your proxy settings.')
  59     except requests.exceptions.RequestException as e:
  60       # catastrophic error. bail.
  61       raise SystemExit(e)
  62
  63     if response.status_code >= 200 and response.status_code < 300:
  64       return response.json()
  65     else:
  66       sys.exit('Reading VES "stndDefined" message template failed.')
  67
  68 def getInitData(domain, stndBody=''):
  69   currentTime = datetime.datetime.utcnow()
  70   dir = os.path.dirname(os.path.realpath(__file__))
  71
  72   result = {}
  73   result['domain']= domain
  74   result['directory']= dir
  75   result['outdir']= dir + '/json/examples'
  76   result['fqdn']= socket.getfqdn()
  77   result['timestamp']= int(currentTime.timestamp()*1000000)
  78   result['eventTime']= currentTime.isoformat() + 'Z'
  79   result['interface']= "urn:ietf:params:xml:ns:yang:ietf-interfaces:interfaces/interface/name='O-RAN-SC-OAM'"
  80
  81   # Read config
  82   with open('config.yml', 'r') as stream:
  83     try:
  84         result['config']= yaml.safe_load(stream)
  85     except yaml.YAMLError as exc:
  86         print(exc)
  87
  88   # Read template body
  89   if domain == 'stndDefined':
  90     url = 'https://raw.githubusercontent.com/onap/testsuite/master/robot/assets/dcae/ves_stdnDefined_' + stndBody + '.json'
  91     result['body']= sendHttpGet(url)
  92   else:
  93     templateFileName = dir + '/json/templates/' + domain + '.json'
  94     with open(templateFileName) as f:
  95       result['body']= json.load(f)
  96
  97   Path(result["outdir"]).mkdir(parents=True, exist_ok=True)
  98   return result
  99
 100 def saveExample(data):
 101   if 'directory' in data and 'domain' in data and 'body' in data:
 102     name = data['domain']
 103     if 'pnfId' in data: name = '-'.join( (data['pnfId'], data['domain']) )
 104     outputFileName = data['directory'] + '/json/examples/' + name + '.json'
 105     with open(outputFileName, 'w') as f:
 106       json.dump(data['body'], f, indent=2, sort_keys=True)
 107   else:
 108     print("Example could not been saved:\n" + json.dump(data, f, indent=2, sort_keys=True))
 109