# http://www.apache.org/licenses/LICENSE-2.0
#
+import yaml
+import os
+
CERT_PATH = './cert/'
TESTCASE_PATH = './testcase/'
SCENARIO_NAMING_FMT = 'certification_%s'
-import yaml
-import os
-
curr_path = os.path.dirname(os.path.abspath(__file__))
with open(os.path.join(curr_path, 'dovetail_config.yml')) as f:
dovetail_config = yaml.safe_load(f)
import utils.dovetail_logger as dt_logger
import utils.dovetail_utils as dt_utils
-from conf.dovetail_config import *
-
+from conf.dovetail_config import dovetail_config
logger = dt_logger.Logger('container.py').getLogger()
+
class Container:
container_list = {}
- has_pull_latest_image = {'yardstick':False, 'functest':False}
+ has_pull_latest_image = {'yardstick': False, 'functest': False}
def __init__(cls):
pass
@classmethod
def get_docker_image(cls, type):
- return '%s:%s' % (dovetail_config[type]['image_name'], dovetail_config[type]['docker_tag'])
+ return '%s:%s' % (dovetail_config[type]['image_name'],
+ dovetail_config[type]['docker_tag'])
@classmethod
def create(cls, type):
- #sshkey="-v /root/.ssh/id_rsa:/root/.ssh/id_rsa "
+ # sshkey="-v /root/.ssh/id_rsa:/root/.ssh/id_rsa "
docker_image = cls.get_docker_image(type)
envs = dovetail_config[type]['envs']
opts = dovetail_config[type]['opts']
sshkey = ''
- result_volume = ' -v %s:%s ' % (dovetail_config['result_dir'],dovetail_config[type]['result']['dir'])
- cmd = 'sudo docker run %s %s %s %s %s /bin/bash' % (opts, envs, sshkey, result_volume, docker_image)
- dt_utils.exec_cmd(cmd,logger)
- ret, container_id=dt_utils.exec_cmd("sudo docker ps | grep "+ docker_image + " | awk '{print $1}' | head -1",logger)
+ result_volume = ' -v %s:%s ' % (dovetail_config['result_dir'],
+ dovetail_config[type]['result']['dir'])
+ cmd = 'sudo docker run %s %s %s %s %s /bin/bash' % \
+ (opts, envs, sshkey, result_volume, docker_image)
+ dt_utils.exec_cmd(cmd, logger)
+ ret, container_id = \
+ dt_utils.exec_cmd("sudo docker ps | grep " + docker_image +
+ " | awk '{print $1}' | head -1", logger)
cls.container_list[type] = container_id
return container_id
@classmethod
def pull_image(cls, type):
docker_image = cls.get_docker_image(type)
- if cls.has_pull_latest_image[type] == True:
+ if cls.has_pull_latest_image[type] is True:
logger.debug('%s is already the newest version.' % (docker_image))
else:
cmd = 'sudo docker pull %s' % (docker_image)
- dt_utils.exec_cmd(cmd,logger)
+ dt_utils.exec_cmd(cmd, logger)
cls.has_pull_latest_image[type] = True
@classmethod
def clean(cls, container_id):
cmd1 = 'sudo docker stop %s' % (container_id)
- dt_utils.exec_cmd(cmd1,logger)
+ dt_utils.exec_cmd(cmd1, logger)
cmd2 = 'sudo docker rm %s' % (container_id)
- dt_utils.exec_cmd(cmd2,logger)
+ dt_utils.exec_cmd(cmd2, logger)
@classmethod
def exec_cmd(cls, container_id, sub_cmd, exit_on_error=False):
cmd = 'sudo docker exec %s /bin/bash -c "%s"' % (container_id, sub_cmd)
- dt_utils.exec_cmd(cmd,logger,exit_on_error)
+ dt_utils.exec_cmd(cmd, logger, exit_on_error)
import utils.dovetail_logger as dt_logger
import utils.dovetail_utils as dt_utils
+from conf.dovetail_config import dovetail_config
logger = dt_logger.Logger('parser.py').getLogger()
-from conf.dovetail_config import *
class Parser:
'''preprocess configuration files'''
for arg in dovetail_config['parameters']:
path = eval(arg['path'])
logger.debug('name: %s, eval path: %s ' % (arg['name'], path))
- kwargs[arg['name']] = dt_utils.get_obj_by_path(testcase.testcase,path)
+ kwargs[arg['name']] = \
+ dt_utils.get_obj_by_path(testcase.testcase, path)
- logger.debug('kwargs: %s' % kwargs)
+ logger.debug('kwargs: %s' % kwargs)
cmd_lines = template.render(**kwargs)
except Exception as e:
logger.error('failed to parse cmd %s, exception:%s' % (cmd, e))
# http://www.apache.org/licenses/LICENSE-2.0
#
-import os
-
import utils.dovetail_logger as dt_logger
import utils.dovetail_utils as dt_utils
cmd = "sudo pip install click pyyaml jinja2"
dt_utils.exec_cmd(cmd, logger)
-
import json
import urllib2
import re
+import os
import utils.dovetail_logger as dt_logger
-import utils.dovetail_utils as dt_utils
-from conf.dovetail_config import *
-from testcase import *
+from conf.dovetail_config import dovetail_config
+from testcase import Testcase
logger = dt_logger.Logger('report.py').getLogger()
+
def get_pass_str(passed):
if passed:
return 'PASS'
else:
return 'FAIL'
+
class Report:
- results = {'functest':{},'yardstick':{}}
+ results = {'functest': {}, 'yardstick': {}}
@classmethod
def check_result(cls, testcase, db_result):
report = ''
report += '\n\
-+=============================================================================+\n\
-| report | \n\
-+-----------------------------------------------------------------------------+\n'
++==========================================================================+\n\
+| report |\n\
++--------------------------------------------------------------------------+\n'
report += '|scenario: %s\n' % scenario_yaml['name']
for testcase_name in scenario_yaml['testcase_list']:
testcase = Testcase.get(testcase_name)
- report += '| [testcase]: %s\t\t\t\t[%s]\n' % (testcase_name, get_pass_str(testcase.passed()))
+ report += '| [testcase]: %s\t\t\t\t[%s]\n' % \
+ (testcase_name, get_pass_str(testcase.passed()))
report += '| |-objective: %s\n' % testcase.objective()
if testcase.sub_testcase() is not None:
for subtest in testcase.sub_testcase():
- report += '| |-%s \t\t [%s]\n' % (subtest, get_pass_str(testcase.sub_testcase_passed(subtest)))
- report += '+-----------------------------------------------------------------------------+\n'
+ report += '| |-%s \t\t [%s]\n' % \
+ (subtest,
+ get_pass_str(testcase.sub_testcase_passed(subtest)))
+ report += '+-----------------------------------------------------'
+ report += '---------------------+\n'
logger.info(report)
cls.save(report)
return report
- #save to disk as default
+ # save to disk as default
@classmethod
def save(cls, report):
report_file_path = dovetail_config['report_file']
try:
- with open(os.path.join(dovetail_config['result_dir'], report_file_path),'w') as report_file:
+ with open(os.path.join(dovetail_config['result_dir'],
+ report_file_path), 'w') as report_file:
report_file.write(report)
logger.info('save report to %s' % report_file_path)
- except Exception as e:
+ except Exception:
logger.error('Failed to save: %s' % report_file_path)
@classmethod
logger.debug('testcase: %s -> result acquired' % script_testcase)
else:
retry = testcase.increase_retry()
- logger.debug('testcase: %s -> result acquired retry:%d' % (script_testcase, retry))
+ logger.debug('testcase: %s -> result acquired retry:%d' %
+ (script_testcase, retry))
return result
+
class CrawlerFactory:
@classmethod
return None
+
class FunctestCrawler:
def __init__(self):
return self.crawl_from_url(testcase)
def crawl_from_file(self, testcase=None):
- file_path = os.path.join(dovetail_config['result_dir'],dovetail_config[self.type]['result']['file_path'])
+ file_path = \
+ os.path.join(dovetail_config['result_dir'],
+ dovetail_config[self.type]['result']['file_path'])
if not os.path.exists(file_path):
logger.info('result file not found: %s' % file_path)
return None
if failed_num != 0:
criteria = 'FAIL'
- match = re.findall('Ran: (\d*) tests in (\d*)\.\d* sec.', output)
+ match = re.findall('Ran: (\d*) tests in (\d*)\.\d* sec.', output)
num_tests, dur_sec_int = match[0]
- json_results = {'criteria':criteria,'details':{"timestart": '', "duration": int(dur_sec_int),
+ json_results = {'criteria': criteria, 'details': {"timestart": '',
+ "duration": int(dur_sec_int),
"tests": int(num_tests), "failures": failed_num,
"errors": error_logs}}
logger.debug('Results: %s' % str(json_results))
return json_results
except Exception as e:
- logger.error('Cannot read content from the file: %s, exception: %s' % (file_path, e))
+ logger.error('Cannot read content from the file: %s, exception: %s'
+ % (file_path, e))
return None
def crawl_from_url(self, testcase=None):
data = json.load(urllib2.urlopen(url))
return data['results'][0]
except Exception as e:
- logger.error("Cannot read content from the url: %s, exception: %s" % (url, e))
+ logger.error("Cannot read content from the url: %s, exception: %s"
+ % (url, e))
return None
+
class YardstickCrawler:
def __init__(self):
return self.crawl_from_url(testcase)
def crawl_from_file(self, testcase=None):
- file_path = os.path.join(dovetail_config['result_dir'], testcase+'.out')
+ file_path = os.path.join(dovetail_config['result_dir'],
+ testcase+'.out')
if not os.path.exists(file_path):
logger.info('result file not found: %s' % file_path)
return None
try:
with open(file_path, 'r') as myfile:
- output = myfile.read()
+ myfile.read()
criteria = 'PASS'
- json_results = {'criteria':criteria}
+ json_results = {'criteria': criteria}
logger.debug('Results: %s' % str(json_results))
return json_results
except Exception as e:
- logger.error('Cannot read content from the file: %s, exception: %s' % (file_path, e))
+ logger.error('Cannot read content from the file: %s, exception: %s'
+ % (file_path, e))
return None
def crawl_from_url(self, testcase=None):
return None
+
class CheckerFactory:
@classmethod
- def create(cls,type):
+ def create(cls, type):
if type == 'functest':
return FunctestChecker()
return None
+
class ResultChecker:
def check(cls):
return 'PASS'
+
class FunctestChecker:
def check(cls, testcase, db_result):
if not db_result:
for sub_testcase in testcase.sub_testcase():
- testcase.sub_testcase_passed(sub_testcase,False)
+ testcase.sub_testcase_passed(sub_testcase, False)
return
testcase.passed(db_result['criteria'] == 'PASS')
if testcase.sub_testcase() is None:
return
- if testcase.testcase['passed'] == True:
+ if testcase.testcase['passed'] is True:
for sub_testcase in testcase.sub_testcase():
testcase.sub_testcase_passed(sub_testcase, True)
return
testcase.passed(all_passed)
+
class YardstickChecker:
def check(cls, testcase, result):
import click
-import yaml
-import os
-import time
import utils.dovetail_logger as dt_logger
-import utils.dovetail_utils as dt_utils
from container import Container
-from testcase import *
-from report import *
-from conf.dovetail_config import *
+from testcase import Testcase
+from testcase import Scenario
+from report import Report
+from conf.dovetail_config import SCENARIO_NAMING_FMT
logger = dt_logger.Logger('run.py').getLogger()
+
def load_scenario(scenario):
Scenario.load()
return Scenario.get(SCENARIO_NAMING_FMT % scenario)
+
def load_testcase():
Testcase.load()
+
def run_test(scenario):
for testcase_name in scenario['testcase_list']:
logger.info('>>[testcase]: %s' % (testcase_name))
logger.debug('container id:%s' % container_id)
if not Testcase.prepared(testcase.script_type()):
- cmds = Testcase.pre_condition(testcase.script_type())['cmds']
+ cmds = \
+ Testcase.pre_condition_cls(testcase.script_type())['cmds']
if cmds:
for cmd in cmds:
Container.exec_cmd(container_id, cmd)
- Testcase.prepared(testcase.script_type(),True)
+ Testcase.prepared(testcase.script_type(), True)
if not testcase.prepare_cmd():
logger.error('failed to prepare testcase:%s' % testcase.name())
for cmd in testcase.cmds:
Container.exec_cmd(container_id, cmd)
- #testcase.post_condition()
+ # testcase.post_condition()
Container.clean(container_id)
db_result = Report.get_result(testcase)
Report.check_result(testcase, db_result)
+
@click.command()
@click.option('--scenario', default='basic', help='certification scenario')
def main(scenario):
# http://www.apache.org/licenses/LICENSE-2.0
#
-import jinja2
+import os
+import yaml
import utils.dovetail_logger as dt_logger
-import utils.dovetail_utils as dt_utils
-from parser import *
+from parser import Parser
+from conf.dovetail_config import CERT_PATH
+from conf.dovetail_config import TESTCASE_PATH
+from conf.dovetail_config import dovetail_config
logger = dt_logger.Logger('testcase.py').getLogger()
-from conf.dovetail_config import *
class Testcase:
self.testcase['passed'] = False
self.cmds = []
self.sub_testcase_status = {}
- Testcase.update_script_testcase(self.script_type(), self.script_testcase())
+ Testcase.update_script_testcase(self.script_type(),
+ self.script_testcase())
def prepare_cmd(self):
for cmd in dovetail_config[self.script_type()]['testcase']['cmds']:
- cmd_lines = Parser.parse_cmd(cmd,self)
+ cmd_lines = Parser.parse_cmd(cmd, self)
if not cmd_lines:
return False
self.cmds.append(cmd_lines)
return self.testcase['scripts']['testcase']
def exceed_max_retry_times(self):
- #logger.debug('retry times:%d' % self.testcase['retry'])
- return Testcase._exceed_max_retry_times(self.script_type(), self.script_testcase())
+ # logger.debug('retry times:%d' % self.testcase['retry'])
+ return Testcase._exceed_max_retry_times(self.script_type(),
+ self.script_testcase())
def increase_retry(self):
- #self.testcase['retry'] = self.testcase['retry'] + 1
- #return self.testcase['retry']
- return Testcase._increase_retry(self.script_type(), self.script_testcase())
+ # self.testcase['retry'] = self.testcase['retry'] + 1
+ # return self.testcase['retry']
+ return Testcase._increase_retry(self.script_type(),
+ self.script_testcase())
- def passed(self, passed = None):
+ def passed(self, passed=None):
if passed is not None:
self.testcase['passed'] = passed
return self.testcase['passed']
def script_result_acquired(self, acquired=None):
- return Testcase._result_acquired(self.script_type(), self.script_testcase(), acquired)
+ return Testcase._result_acquired(self.script_type(),
+ self.script_testcase(), acquired)
def pre_condition(self):
- return Testcase.pre_condition(self.script_type())
+ return Testcase.pre_condition_cls(self.script_type())
def post_condition(self):
- return Testcase.post_condition(self.script_type())
+ return Testcase.post_condition_cls(self.script_type())
+ # testcase in upstream testing project
+ script_testcase_list = {'functest': {}, 'yardstick': {}}
- #testcase in upstream testing project
- script_testcase_list = {'functest':{}, 'yardstick':{}}
-
- #testcase in dovetail
+ # testcase in dovetail
testcase_list = {}
@classmethod
return cls.script_testcase_list[script_type]['cleaned']
@classmethod
- def pre_condition(cls, script_type):
+ def pre_condition_cls(cls, script_type):
return dovetail_config[script_type]['pre_condition']
- def post_condition(cls, script_type):
+ def post_condition_cls(cls, script_type):
return dovetail_config[script_type]['post_condition']
-
@classmethod
- def update_script_testcase(cls,script_type, script_testcase):
+ def update_script_testcase(cls, script_type, script_testcase):
if script_testcase not in cls.script_testcase_list[script_type]:
- cls.script_testcase_list[script_type][script_testcase] = {'retry':0, 'acquired':False}
+ cls.script_testcase_list[script_type][script_testcase] = \
+ {'retry': 0, 'acquired': False}
cls.script_testcase_list[script_type]['prepared'] = False
cls.script_testcase_list[script_type]['cleaned'] = False
@classmethod
- def _exceed_max_retry_times(cls, script_type, script_testcase ):
- return cls.script_testcase_list[script_type][script_testcase]['retry'] > 1
+ def _exceed_max_retry_times(cls, script_type, script_testcase):
+ retry = cls.script_testcase_list[script_type][script_testcase]['retry']
+ return retry > 1
@classmethod
def _increase_retry(cls, script_type, script_testcase):
return cls.script_testcase_list[script_type][script_testcase]['retry']
@classmethod
- def _result_acquired(cls, script_type, script_testcase, acquired=None):
+ def _result_acquired(cls, script_type, testcase, acquired=None):
if acquired is not None:
- cls.script_testcase_list[script_type][script_testcase]['acquired'] = acquired
- return cls.script_testcase_list[script_type][script_testcase]['acquired']
+ cls.script_testcase_list[script_type][testcase]['acquired'] = \
+ acquired
+ return cls.script_testcase_list[script_type][testcase]['acquired']
@classmethod
def load(cls):
for testcase_file in files:
with open(os.path.join(root, testcase_file)) as f:
testcase_yaml = yaml.safe_load(f)
- cls.testcase_list[testcase_yaml.keys()[0]] = Testcase(testcase_yaml)
- logger.debug( cls.testcase_list )
+ cls.testcase_list[testcase_yaml.keys()[0]] = \
+ Testcase(testcase_yaml)
+ logger.debug(cls.testcase_list)
@classmethod
def get(cls, testcase_name):
return None
scenario_list = {}
+
@classmethod
def load(cls):
for root, dirs, files in os.walk(CERT_PATH):
if scenario_name in cls.scenario_list:
return cls.scenario_list[scenario_name]
return None
-
import logging
import os
+
class Logger:
def __init__(self, logger_name):
def getLogger(self):
return self.logger
-
import sys
import subprocess
+from collections import Mapping, Set, Sequence
+
-def exec_cmd(cmd, logger=None,
- exit_on_error=True,
- info=False,
- error_msg="",
- verbose=True):
+def exec_cmd(cmd, logger=None, exit_on_error=True, info=False,
+ error_msg="", verbose=True):
if not error_msg:
error_msg = ("The command '%s' failed." % cmd)
msg_exec = ("Executing command: '%s'" % cmd)
return returncode, output[0].strip()
-#walkthrough the object, yield path and value
-from collections import Mapping, Set, Sequence
+# walkthrough the object, yield path and value
# dual python 2/3 compatability, inspired by the "six" library
string_types = (str, unicode) if str is bytes else (str, bytes)
-iteritems = lambda mapping: getattr(mapping, 'iteritems', mapping.items)()
+# iteritems = lambda mapping: getattr(mapping, 'iteritems', mapping.items)()
+
+
+def iteritems(mapping):
+ return getattr(mapping, 'iteritems', mapping.items)()
+
def objwalk(obj, path=(), memo=None):
if memo is None:
iterator = None
if isinstance(obj, Mapping):
iterator = iteritems
- elif isinstance(obj, (Sequence, Set)) and not isinstance(obj, string_types):
+ elif isinstance(obj, (Sequence, Set)) and not isinstance(obj,
+ string_types):
iterator = enumerate
if iterator:
if id(obj) not in memo:
else:
yield path, obj
-def get_obj_by_path(obj,dst_path):
+
+def get_obj_by_path(obj, dst_path):
for path, obj in objwalk(obj):
if path == dst_path:
return obj
-
Code Review / dovetail.git / commitdiff