import datetime
import json
+import requests
import os
import socket
+import sys
import yaml
from pathlib import Path
-def getInitData(domain):
+def sendVesEvent(data):
+ url = data['config']['vesEndpoint']['url']
+ username = data['config']['vesEndpoint']['username']
+ password = data['config']['vesEndpoint']['password']
+ headers = {'content-type': 'application/json'}
+ verify = data['config']['vesEndpoint']['verify']
+ try:
+ response = requests.post(url, json=data['body'], auth=(
+ username, password), headers=headers, verify=verify)
+ except requests.exceptions.Timeout:
+ sys.exit('HTTP request failed, please check you internet connection.')
+ except requests.exceptions.TooManyRedirects:
+ sys.exit('HTTP request failed, please check your proxy settings.')
+ except requests.exceptions.RequestException as e:
+ # catastrophic error. bail.
+ raise SystemExit(e)
+
+ if response.status_code >= 200 and response.status_code < 300:
+ print(response)
+ else:
+ sys.exit('Reading VES "stndDefined" message template failed.')
+
+def sendHttpGet(url):
+ try:
+ response = requests.get(url)
+ except requests.exceptions.Timeout:
+ sys.exit('HTTP request failed, please check you internet connection.')
+ except requests.exceptions.TooManyRedirects:
+ sys.exit('HTTP request failed, please check your proxy settings.')
+ except requests.exceptions.RequestException as e:
+ # catastrophic error. bail.
+ raise SystemExit(e)
+
+ if response.status_code >= 200 and response.status_code < 300:
+ return response.json()
+ else:
+ sys.exit('Reading VES "stndDefined" message template failed.')
+
+def getInitData(domain, stndBody=''):
currentTime = datetime.datetime.utcnow()
dir = os.path.dirname(os.path.realpath(__file__))
result['fqdn']= socket.getfqdn()
result['timestamp']= int(currentTime.timestamp()*1000000)
result['eventTime']= currentTime.isoformat() + 'Z'
+ result['interface']= "urn:ietf:params:xml:ns:yang:ietf-interfaces:interfaces/interface/name='O-RAN-SC-OAM'"
# Read config
with open('config.yml', 'r') as stream:
print(exc)
# Read template body
- templateFileName = dir + '/json/templates/' + domain + '.json'
- with open(templateFileName) as f:
- result['body']= json.load(f)
-
+ if domain == 'stndDefined':
+ url = 'https://raw.githubusercontent.com/onap/testsuite/master/robot/assets/dcae/ves_stdnDefined_' + stndBody + '.json'
+ result['body']= sendHttpGet(url)
+ else:
+ templateFileName = dir + '/json/templates/' + domain + '.json'
+ with open(templateFileName) as f:
+ result['body']= json.load(f)
Path(result["outdir"]).mkdir(parents=True, exist_ok=True)
return result
def saveExample(data):
if 'directory' in data and 'domain' in data and 'body' in data:
- outputFileName = data['directory'] + '/json/examples/' + data['domain'] + '.json'
+ name = data['domain']
+ if 'pnfId' in data: name = '-'.join( (data['pnfId'], data['domain']) )
+ outputFileName = data['directory'] + '/json/examples/' + name + '.json'
with open(outputFileName, 'w') as f:
json.dump(data['body'], f, indent=2, sort_keys=True)
else: