#
import functools
import json
+import logging
import os
import re
import shutil
import subprocess
import sys
import time
-import urllib2
from datetime import datetime as dt
import dns.resolver
import requests
+from six.moves import urllib
import yaml
from git import Repo
+from functest.utils import constants
from functest.utils import decorators
-import functest.utils.functest_logger as ft_logger
-logger = ft_logger.Logger("functest_utils").getLogger()
+logger = logging.getLogger(__name__)
# ----------------------------------------------------------
Check if there is access to the internet
"""
try:
- urllib2.urlopen(url, timeout=5)
+ urllib.request.urlopen(url, timeout=5)
return True
- except urllib2.URLError:
+ except urllib.error.URLError:
return False
name = url.rsplit('/')[-1]
dest = dest_path + "/" + name
try:
- response = urllib2.urlopen(url)
- except (urllib2.HTTPError, urllib2.URLError):
+ response = urllib.request.urlopen(url)
+ except (urllib.error.HTTPError, urllib.error.URLError):
return False
with open(dest, 'wb') as f:
@decorators.can_dump_request_to_file
def push_results_to_db(project, case_name,
- start_date, stop_date, criteria, details):
+ start_date, stop_date, result, details):
"""
POST results to the Result target DB
"""
# Retrieve params from CI and conf
- url = get_db_url() + "/results"
+ url = get_db_url()
try:
installer = os.environ['INSTALLER_TYPE']
params = {"project_name": project, "case_name": case_name,
"pod_name": pod_name, "installer": installer,
- "version": version, "scenario": scenario, "criteria": criteria,
+ "version": version, "scenario": scenario, "criteria": result,
"build_tag": build_tag, "start_date": test_start,
"stop_date": test_stop, "details": details}
error = None
headers = {'Content-Type': 'application/json'}
try:
- r = requests.post(url, data=json.dumps(params), headers=headers)
+ r = requests.post(url, data=json.dumps(params, sort_keys=True),
+ headers=headers)
logger.debug(r)
r.raise_for_status()
except requests.RequestException as exc:
'pod': pod_name,
'v': version,
's': scenario,
- 'c': criteria,
+ 'c': result,
't': build_tag,
'd': details,
'error': e
f.write(line)
else:
line = line.replace('\n', '')
- print line
+ print(line)
sys.stdout.flush()
if output_file:
f.close()
def get_functest_config(parameter):
- yaml_ = os.environ["CONFIG_FUNCTEST_YAML"]
+ yaml_ = constants.CONST.__getattribute__('CONFIG_FUNCTEST_YAML')
return get_parameter_from_yaml(parameter, yaml_)
-def check_success_rate(case_name, success_rate):
- success_rate = float(success_rate)
- criteria = get_criteria_by_test(case_name)
-
- def get_criteria_value(op):
- return float(criteria.split(op)[1].rstrip('%'))
-
- status = 'FAIL'
- ops = ['==', '>=']
- for op in ops:
- if op in criteria:
- c_value = get_criteria_value(op)
- if eval("%s %s %s" % (success_rate, op, c_value)):
- status = 'PASS'
- break
-
- return status
-
-
def merge_dicts(dict1, dict2):
for k in set(dict1.keys()).union(dict2.keys()):
if k in dict1 and k in dict2:
def get_functest_yaml():
- with open(os.environ["CONFIG_FUNCTEST_YAML"]) as f:
+ with open(constants.CONST.__getattribute__('CONFIG_FUNCTEST_YAML')) as f:
functest_yaml = yaml.safe_load(f)
f.close()
return functest_yaml