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