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 from _datetime import timezone
  30
  31 def sendVesEvent(data):
  32     url = data['config']['vesEndpoint']['url']
  33     username = data['config']['vesEndpoint']['username']
  34     password = data['config']['vesEndpoint']['password']
  35     headers = {'content-type': 'application/json'}
  36     verify = data['config']['vesEndpoint']['verify']
  37     try:
  38       response = requests.post(url, json=data['body'], auth=(
  39           username, password), headers=headers, verify=verify)
  40     except requests.exceptions.Timeout:
  41       sys.exit('HTTP request failed, please check you internet connection.')
  42     except requests.exceptions.TooManyRedirects:
  43       sys.exit('HTTP request failed, please check your proxy settings.')
  44     except requests.exceptions.RequestException as e:
  45       # catastrophic error. bail.
  46       raise SystemExit(e)
  47
  48     if response.status_code >= 200 and response.status_code < 300:
  49       print(response)
  50     else:
  51       sys.exit('Reading VES "stndDefined" message template failed.')
  52
  53 def sendHttpGet(url):
  54     try:
  55       response = requests.get(url)
  56     except requests.exceptions.Timeout:
  57       sys.exit('HTTP request failed, please check you internet connection.')
  58     except requests.exceptions.TooManyRedirects:
  59       sys.exit('HTTP request failed, please check your proxy settings.')
  60     except requests.exceptions.RequestException as e:
  61       # catastrophic error. bail.
  62       raise SystemExit(e)
  63
  64     if response.status_code >= 200 and response.status_code < 300:
  65       return response.json()
  66     else:
  67       sys.exit('Reading VES "stndDefined" message template failed.')
  68
  69 def getInitData(domain, stndBody=''):
  70   currentTime = datetime.datetime.now(tz=timezone.utc)
  71   dir = os.path.dirname(os.path.realpath(__file__))
  72
  73   result = {}
  74   result['domain']= domain
  75   result['directory']= dir
  76   result['outdir']= dir + '/json/examples'
  77   result['fqdn']= socket.getfqdn()
  78   result['timestamp']= int(currentTime.timestamp()*1000000)
  79   result['eventTime']= currentTime.isoformat() + 'Z'
  80   result['interface']= "urn:ietf:params:xml:ns:yang:ietf-interfaces:interfaces/interface/name='O-RAN-SC-OAM'"
  81
  82   # Read config
  83   with open('config.yml', 'r') as stream:
  84     try:
  85         result['config']= yaml.safe_load(stream)
  86     except yaml.YAMLError as exc:
  87         print(exc)
  88
  89   # Read template body
  90   if domain == 'stndDefined':
  91     url = 'https://raw.githubusercontent.com/onap/testsuite/master/robot/assets/dcae/ves_stdnDefined_' + stndBody + '.json'
  92     result['body']= sendHttpGet(url)
  93   else:
  94     templateFileName = dir + '/json/templates/' + domain + '.json'
  95     with open(templateFileName) as f:
  96       result['body']= json.load(f)
  97
  98   Path(result["outdir"]).mkdir(parents=True, exist_ok=True)
  99   return result
 100
 101 def saveExample(data):
 102   if 'directory' in data and 'domain' in data and 'body' in data:
 103     name = data['domain']
 104     if 'pnfId' in data: name = '-'.join( (data['pnfId'], data['domain']) )
 105     outputFileName = data['directory'] + '/json/examples/' + name + '.json'
 106     with open(outputFileName, 'w') as f:
 107       json.dump(data['body'], f, indent=2, sort_keys=True)
 108   else:
 109     print("Example could not been saved:\n" + json.dump(data, f, indent=2, sort_keys=True))
 110