tests/dcaegen2/testcases/resources/robot_library/DcaeLibrary.py

   1 '''
   2 Created on Aug 18, 2017
   3
   4 @author: sw6830
   5 '''
   6 from robot.api import logger
   7 import uuid
   8 import time
   9 import datetime
  10 import json
  11 import os
  12 import platform
  13 import subprocess
  14 import paramiko
  15
  16
  17 class DcaeLibrary(object):
  18
  19     def __init__(self):
  20         pass
  21
  22     @staticmethod
  23     def override_collector_properties(properties_path):
  24         global client
  25         if 'Windows' in platform.system():
  26             try:
  27                 DcaeLibrary.change_properties_for_windows_platform_system(properties_path)
  28             finally:
  29                 client.close()
  30             return
  31         DcaeLibrary.change_properties_for_non_windows_platform_system(properties_path)
  32         return
  33
  34     @staticmethod
  35     def change_properties_for_non_windows_platform_system(properties_path):
  36         ws = os.environ['WORKSPACE']
  37         script2run = ws + '/tests/dcaegen2/testcases/resources/override_collector_properties.sh'
  38         logger.info("Running script: " + script2run)
  39         logger.console("Running script: " + script2run)
  40         subprocess.call([script2run, properties_path])
  41         time.sleep(5)
  42
  43     @staticmethod
  44     def change_properties_for_windows_platform_system(properties_path):
  45         global client
  46         client = paramiko.SSHClient()
  47         client.load_system_host_keys()
  48         client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
  49         client.connect(os.environ['CSIT_IP'], port=22, username=os.environ['CSIT_USER'], password=os.environ['CSIT_PD'])
  50         stdin, stdout, stderr = client.exec_command(
  51             '%{WORKSPACE}' + '/tests/dcaegen2/testcases/resources/override_collector_properties.sh', properties_path)
  52         logger.console(stdout.read())
  53
  54     @staticmethod
  55     def is_json_empty(resp):
  56         logger.info("Enter is_json_empty: resp.text: " + resp.text)
  57         if resp.text is None or len(resp.text) < 2:
  58             return 'True'
  59         return 'False'
  60
  61     @staticmethod
  62     def generate_uuid():
  63         """generate a uuid"""
  64         return uuid.uuid4()
  65
  66     @staticmethod
  67     def get_json_value_list(jsonstr, keyval):
  68         logger.info("Enter Get_Json_Key_Value_List")
  69         if jsonstr is None or len(jsonstr) < 2:
  70             logger.info("No Json data found")
  71             return []
  72         try:
  73             return DcaeLibrary.extract_list_of_items_from_json_string(jsonstr, keyval)
  74         except Exception as e:
  75             logger.info("Json data parsing fails")
  76             print str(e)
  77             return []
  78
  79     @staticmethod
  80     def extract_list_of_items_from_json_string(jsonstr, keyval):
  81         data = json.loads(jsonstr)
  82         nodelist = []
  83         for item in data:
  84             nodelist.append(item[keyval])
  85         return nodelist
  86
  87     @staticmethod
  88     def generate_millitimestamp_uuid():
  89         """generate a millisecond timestamp uuid"""
  90         then = datetime.datetime.now()
  91         return int(time.mktime(then.timetuple())*1e3 + then.microsecond/1e3)
  92
  93     @staticmethod
  94     def test():
  95         import json
  96         from pprint import pprint
  97
  98         with open('robot/assets/dcae/ves_volte_single_fault_event.json') as data_file:
  99             data = json.load(data_file)
 100
 101         data['event']['commonEventHeader']['version'] = '5.0'
 102         pprint(data)
 103
 104
 105 if __name__ == '__main__':
 106
 107     lib = DcaeLibrary()
 108     lib.enable_vesc_https_auth()
 109
 110     ret = lib.setup_dmaap_server()
 111     print ret
 112     time.sleep(100000)