-class TestCase(object):
- def __init__(self, name, project, criteria=-1, isRunnable=True):
- self.name = name
- self.project = project
- self.criteria = criteria
- self.isRunnable = isRunnable
-
- def getName(self):
- return self.name
-
- def getProject(self):
- return self.project
-
- def getCriteria(self):
- return self.criteria
-
- def setCriteria(self, criteria):
- self.criteria = criteria
-
- def setIsRunnable(self, isRunnable):
- self.isRunnable = isRunnable
-
- def checkRunnable(self, installer, scenario, config):
- # Re-use Functest declaration
- # Retrieve Functest configuration file functest_config.yaml
- is_runnable = True
- config_test = ""
- TEST_ENV = functest_yaml_config.get("test-dependencies")
-
- # print " *********************** "
- # print TEST_ENV
- # print " ---------------------- "
- # print "case = " + self.name
- # print "installer = " + installer
- # print "scenario = " + scenario
- # print "project = " + self.project
-
- # Retrieve test constraints
- case_name_formated = test_match_matrix[self.name]
-
- try:
- config_test = TEST_ENV[self.project][case_name_formated]
- except KeyError:
- # if not defined in dependencies => no dependencies
- config_test = TEST_ENV[case_name_formated]
- except Exception, e:
- print "Error [getTestEnv]:", e
-
- # Retrieve test execution param
- test_execution_context = {"installer": installer,
- "scenario": scenario}
- # By default we assume that all the tests are always runnable...
- # if test_env not empty => dependencies to be checked
- if config_test is not None and len(config_test) > 0:
- # possible criteria = ["installer", "scenario"]
- # consider test criteria from config file
- # compare towards CI env through CI en variable
- for criteria in config_test:
- if re.search(config_test[criteria],
- test_execution_context[criteria]) is None:
- # print "Test "+ test + " cannot be run on the environment"
- is_runnable = False
- # print is_runnable
- self.isRunnable = is_runnable
-
-
-def getApiResults(case, installer, scenario, version):
- case = case.getName()
- results = json.dumps([])
- # to remove proxy (to be removed at the end for local test only)
- # proxy_handler = urllib2.ProxyHandler({})
- # opener = urllib2.build_opener(proxy_handler)
- # urllib2.install_opener(opener)
- # url = "http://127.0.0.1:8000/results?case=" + case + \
- # "&period=30&installer=" + installer
- url = ("http://testresults.opnfv.org/testapi/results?case=" + case +
- "&period=" + str(PERIOD) + "&installer=" + installer +
- "&scenario=" + scenario + "&version=" + version)
- request = Request(url)
-
- try:
- response = urlopen(request)
- k = response.read()
- results = json.loads(k)
- except URLError, e:
- print 'No kittez. Got an error code:', e
-
- return results
-
-
-def getScenarios(case, installer, version):
-
- case = case.getName()
- url = "http://testresults.opnfv.org/testapi/results?case=" + case + \
- "&period=" + str(PERIOD) + "&installer=" + installer + \
- "&version=" + version
- request = Request(url)
-
- try:
- response = urlopen(request)
- k = response.read()
- results = json.loads(k)
- except URLError, e:
- print 'Got an error code:', e
-
- test_results = results['test_results']
-
- if test_results is not None:
- test_results.reverse()
-
- scenario_results = {}
-
- for r in test_results:
- # Retrieve all the scenarios per installer
- if not r['scenario'] in scenario_results.keys():
- scenario_results[r['scenario']] = []
- scenario_results[r['scenario']].append(r)
-
- return scenario_results
-
-
-def getScenarioStats(scenario_results):
- scenario_stats = {}
- for k, v in scenario_results.iteritems():
- scenario_stats[k] = len(v)
-
- return scenario_stats
-
-
-def getNbtestOk(results):
- nb_test_ok = 0
- for r in results:
- for k, v in r.iteritems():
- try:
- if "passed" in v:
- nb_test_ok += 1
- except:
- print "Cannot retrieve test status"
- return nb_test_ok
-
-
-def getResult(testCase, installer, scenario, version):
-
- # retrieve raw results
- results = getApiResults(testCase, installer, scenario, version)
- # let's concentrate on test results only
- test_results = results['test_results']
-
- # if results found, analyze them
- if test_results is not None:
- test_results.reverse()
-
- scenario_results = []
-
- # print " ---------------- "
- # print test_results
- # print " ---------------- "
- # print "nb of results:" + str(len(test_results))
-
- for r in test_results:
- # print r["creation_date"]
- # print r["criteria"]
- scenario_results.append({r["creation_date"]: r["criteria"]})
- # sort results
- scenario_results.sort()
- # 4 levels for the results
- # 3: 4+ consecutive runs passing the success criteria
- # 2: <4 successful consecutive runs but passing the criteria
- # 1: close to pass the success criteria
- # 0: 0% success, not passing
- test_result_indicator = 0
- nbTestOk = getNbtestOk(scenario_results)
- # print "Nb test OK:"+ str(nbTestOk)
- # check that we have at least 4 runs
- if nbTestOk < 1:
- test_result_indicator = 0
- elif nbTestOk < 2:
- test_result_indicator = 1
- else:
- # Test the last 4 run
- if (len(scenario_results) > 3):
- last4runResults = scenario_results[-4:]
- if getNbtestOk(last4runResults):
- test_result_indicator = 3
- else:
- test_result_indicator = 2
- else:
- test_result_indicator = 2
- print " >>>> Test indicator:" + str(test_result_indicator)
- return test_result_indicator
-
-# ******************************************************************************
-# ******************************************************************************
-# ******************************************************************************
-# ******************************************************************************
-# ******************************************************************************