utils/test/reporting/functest/reportingUtils.py

   1 #!/usr/bin/python
   2 #
   3 # This program and the accompanying materials
   4 # are made available under the terms of the Apache License, Version 2.0
   5 # which accompanies this distribution, and is available at
   6 #
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 from urllib2 import Request, urlopen, URLError
  10 import logging
  11 import json
  12 import reportingConf as conf
  13
  14
  15 def getLogger(module):
  16     logFormatter = logging.Formatter("%(asctime)s [" +
  17                                      module +
  18                                      "] [%(levelname)-5.5s]  %(message)s")
  19     logger = logging.getLogger()
  20
  21     fileHandler = logging.FileHandler("{0}/{1}".format('.', conf.LOG_FILE))
  22     fileHandler.setFormatter(logFormatter)
  23     logger.addHandler(fileHandler)
  24
  25     consoleHandler = logging.StreamHandler()
  26     consoleHandler.setFormatter(logFormatter)
  27     logger.addHandler(consoleHandler)
  28     logger.setLevel(conf.LOG_LEVEL)
  29     return logger
  30
  31
  32 def getApiResults(case, installer, scenario, version):
  33     results = json.dumps([])
  34     # to remove proxy (to be removed at the end for local test only)
  35     # proxy_handler = urllib2.ProxyHandler({})
  36     # opener = urllib2.build_opener(proxy_handler)
  37     # urllib2.install_opener(opener)
  38     # url = "http://127.0.0.1:8000/results?case=" + case + \
  39     #       "&period=30&installer=" + installer
  40     url = (conf.URL_BASE + "?case=" + case +
  41            "&period=" + str(conf.PERIOD) + "&installer=" + installer +
  42            "&scenario=" + scenario + "&version=" + version +
  43            "&last=" + str(conf.NB_TESTS))
  44     request = Request(url)
  45
  46     try:
  47         response = urlopen(request)
  48         k = response.read()
  49         results = json.loads(k)
  50     except URLError, e:
  51         print 'No kittez. Got an error code:', e
  52
  53     return results
  54
  55
  56 def getScenarios(case, installer, version):
  57
  58     case = case.getName()
  59     url = (conf.URL_BASE + "?case=" + case +
  60            "&period=" + str(conf.PERIOD) + "&installer=" + installer +
  61            "&version=" + version)
  62     request = Request(url)
  63
  64     try:
  65         response = urlopen(request)
  66         k = response.read()
  67         results = json.loads(k)
  68         test_results = results['results']
  69     except URLError, e:
  70         print 'Got an error code:', e
  71
  72     if test_results is not None:
  73         test_results.reverse()
  74
  75         scenario_results = {}
  76
  77         for r in test_results:
  78             # Retrieve all the scenarios per installer
  79             if not r['scenario'] in scenario_results.keys():
  80                 scenario_results[r['scenario']] = []
  81             scenario_results[r['scenario']].append(r)
  82
  83     return scenario_results
  84
  85
  86 def getScenarioStats(scenario_results):
  87     scenario_stats = {}
  88     for k, v in scenario_results.iteritems():
  89         scenario_stats[k] = len(v)
  90
  91     return scenario_stats
  92
  93
  94 def getNbtestOk(results):
  95     nb_test_ok = 0
  96     for r in results:
  97         for k, v in r.iteritems():
  98             try:
  99                 if "PASS" in v:
 100                     nb_test_ok += 1
 101             except:
 102                 print "Cannot retrieve test status"
 103     return nb_test_ok
 104
 105
 106 def getResult(testCase, installer, scenario, version):
 107
 108     # retrieve raw results
 109     results = getApiResults(testCase, installer, scenario, version)
 110     # let's concentrate on test results only
 111     test_results = results['results']
 112
 113     # if results found, analyze them
 114     if test_results is not None:
 115         test_results.reverse()
 116
 117         scenario_results = []
 118
 119         # print " ---------------- "
 120         # print test_results
 121         # print " ---------------- "
 122         # print "nb of results:" + str(len(test_results))
 123
 124         for r in test_results:
 125             # print r["start_date"]
 126             # print r["criteria"]
 127             scenario_results.append({r["start_date"]: r["criteria"]})
 128         # sort results
 129         scenario_results.sort()
 130         # 4 levels for the results
 131         # 3: 4+ consecutive runs passing the success criteria
 132         # 2: <4 successful consecutive runs but passing the criteria
 133         # 1: close to pass the success criteria
 134         # 0: 0% success, not passing
 135         # -1: no run available
 136         test_result_indicator = 0
 137         nbTestOk = getNbtestOk(scenario_results)
 138
 139         # print "Nb test OK (last 10 days):"+ str(nbTestOk)
 140         # check that we have at least 4 runs
 141         if len(scenario_results) < 1:
 142             # No results available
 143             test_result_indicator = -1
 144         elif nbTestOk < 1:
 145             test_result_indicator = 0
 146         elif nbTestOk < 2:
 147             test_result_indicator = 1
 148         else:
 149             # Test the last 4 run
 150             if (len(scenario_results) > 3):
 151                 last4runResults = scenario_results[-4:]
 152                 nbTestOkLast4 = getNbtestOk(last4runResults)
 153                 # print "Nb test OK (last 4 run):"+ str(nbTestOkLast4)
 154                 if nbTestOkLast4 > 3:
 155                     test_result_indicator = 3
 156                 else:
 157                     test_result_indicator = 2
 158             else:
 159                 test_result_indicator = 2
 160     return test_result_indicator