solution/dev/ves-test-collector/client-scripts-ves-v7/sendVesNotification.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 #     http://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 # Send a VES event for domain 'notification'
  20
  21 # importing the datetime, json, requests, os socket and yaml library
  22 import datetime
  23 import json
  24 import requests
  25 import os
  26 import socket
  27 import yaml
  28 from pathlib import Path
  29
  30 # Globals
  31 dir = os.path.dirname(os.path.realpath(__file__))
  32 domain = "notification"
  33 fqdn = socket.getfqdn()
  34
  35 # time formats
  36 currentTime = datetime.datetime.utcnow()
  37 timestamp = int(currentTime.timestamp()*1000000)
  38 eventTime = currentTime.isoformat() + "Z"
  39
  40 # Create output path
  41 Path(dir + "/json/examples").mkdir(parents=True, exist_ok=True)
  42
  43 # Read settings
  44 with open("config.yml", 'r') as stream:
  45     try:
  46         cfg = yaml.safe_load(stream)
  47     except yaml.YAMLError as exc:
  48         print(exc)
  49
  50 print("################################################################################")
  51 print("# send SDN-Controller " + domain)
  52
  53 # Read template body
  54 templateFileName = dir + "/json/templates/" + domain + ".json"
  55 with open(templateFileName) as f:
  56     data = json.load(f)
  57
  58 data["event"]["commonEventHeader"]["domain"] = domain
  59 data["event"]["commonEventHeader"]["eventId"] = fqdn + "_" + eventTime
  60 data["event"]["commonEventHeader"]["eventName"] = domain + \
  61     "_" + cfg["settings"]["eventType"]
  62 data["event"]["commonEventHeader"]["eventType"] = cfg["settings"]["eventType"]
  63 data["event"]["commonEventHeader"]["sequence"] = cfg["settings"]["sequence"]
  64 data["event"]["commonEventHeader"]["reportingEntityName"] = fqdn
  65 data["event"]["commonEventHeader"]["sourceName"] = fqdn
  66 data["event"]["commonEventHeader"]["startEpochMicrosec"] = timestamp
  67 data["event"]["commonEventHeader"]["lastEpochMicrosec"] = timestamp
  68 data["event"]["commonEventHeader"]["nfNamingCode"] = "SDN-Controller"
  69 data["event"]["commonEventHeader"]["nfVendorName"] = "O-RAN-SC OAM"
  70
  71 data["event"]["heartbeatFields"]["additionalFields"]["eventTime"] = eventTime
  72
  73 # save example body
  74 outputFileName = dir + "/json/examples/" + domain + ".json"
  75 with open(outputFileName, 'w') as f:
  76     json.dump(data, f, indent=2, sort_keys=True)
  77
  78 # Send VES Event
  79 url = cfg["vesEndpoint"]["url"]
  80 username = cfg["vesEndpoint"]["username"]
  81 password = cfg["vesEndpoint"]["password"]
  82 verify = cfg["vesEndpoint"]["verify"]
  83 response = requests.post(url, json=data, auth=(username, password), verify=verify)
  84 print(response)