blob: bd2932e8733eca8591fa7cb2570774cc2d041419 [file] [log] [blame]
#!/usr/bin/python
# Copyright 2012 Google Inc. All Rights Reserved.
"""This class defines the base class for automated Bruno test cases.
This file also defines some basic functions that is required for
most test cases. Users can define their own functions in the extended
class, the Run() method should be use as the start point of test case.
"""
__author__ = 'Lehan Meng (lmeng@google.com)'
import datetime
import json
import logging
import os
import re
import device
class TestCase(object):
"""Base class for all test cases.
This base class provides basic functions for the test case:
- configure file parsing
- logging
- test information creation
- etc.
Test case starts in the Run() method, test case specific code can be
placed in here.
"""
def __init__(self, ssh=None, **kwargs):
"""Initiate a test class instance.
Args:
ssh: ssh tunnel to device
kwargs['testID']: The assigned test ID on TCM
kwargs['title']: Title (keyword) of the test case
kwargs['start_time']: Start time of the test case (for statistic purpose)
kwargs['key_word']: the label of the test case
kwargs['description']: description if required
"""
self.test_info = {
'testID': '0',
'title': 'None',
'start_time': str(datetime.datetime.now()
.strftime('%Y-%m-%d %H:%M')),
'key_word': 'None',
'description': 'None',
'configFile': 'config.cfg'}
self.params = {}
for s in kwargs:
self.test_info[s] = kwargs[s]
self.device = device.Device()
self.log = logging.Logging(testID=self.test_info['testID'])
self.ParseJSONConfigFile()
#self.ParseConfig()
if ssh:
self.p_ssh = ssh
else:
self.p_ssh = None
self.Run()
@staticmethod
def FindTestCaseID(file_name):
"""Parse the configuration file for test case.
Find test ids in the configuration file.
Args:
file_name: the configure file to parse
Returns:
return the test case ids that needs to run the test
"""
test_id_list = []
cfg_file = open(file_name, 'r')
cfg_list = cfg_file.readlines()
for line in cfg_list:
s = line.strip().strip('\n')
m = re.match(r'testID\s*:\s*(\d*)\s*$', s)
if (m is not None) and not s.startswith('#'):
test_id_list.append(m.group(1))
cfg_file.close()
return test_id_list
@staticmethod
def FindTestCaseToExecute(file_name):
"""Parse the configuration file for test case.
Find test ids in the configuration file.
Args:
file_name: the configure file to parse
Returns:
return the test case ids that needs to run the test
"""
test_id_list = []
try:
cfg_file = open(file_name, 'r')
except IOError:
print 'Cannot open the input file, return'
return os.sys.exit(1)
data = json.load(cfg_file)
cfg_file.close()
for test_case in data.iterkeys():
for key in data[test_case].iterkeys():
if isinstance(data[test_case][key], list) and key == 'testID':
for i in range(0, len(data[test_case][key])):
test_id_list.append(data[test_case][key][i])
return test_id_list
def ConfigForThisTestCase(self, data):
"""Match if the configuration data is for current test case.
Args:
data: configuration data
Returns:
True: if the data is for current test case
False: if otherwise
"""
for key in data.iterkeys():
if key == 'testID':
if self.test_info['testID'] in data[key]:
return True
else:
return False
return False
def ParseJSONConfigFile(self):
"""Parse the JSON format configuration file.
The sturcture of a JSON file is as follows:
File format: parameter_names: string_values (or lists of values)
comment is defined in the comment tag (JSON does not support
comment).
'Test_Case_Description' : {
'comment': ['comments'],
'testID': ['11843140', '111111', '222222'],
'user' : 'root',
'addr' : '192.168.1.4',
'addr_ipv6' : 'None',
'pwd' : 'google',
'bruno_prompt' : 'gfibertv#',
'title' : 'Download',
'expected_version_id': '716006',
'expected_version': 'bruno-koala-5',
'downgrade_version': 'bruno-koala-4',
'downgrade_version_id': '728001',
'acs_url' : 'https://gfiber-acs-staging.appspot.com/cwmp',
'jump_server' : 'jmp.googlefiber.net',
'athena_user' : 'your_uname',
'athena_pwd' : 'your_pwd'
}
Returns:
return True if configure file parsed successfully
"""
try:
self.cfg_file = open(self.test_info['configFile'], 'r')
except IOError:
print 'Cannot open the input file, return'
return os.sys.exit(1)
data = json.load(self.cfg_file)
self.cfg_file.close()
test_id_matched = False
for test_case in data.iterkeys():
if not self.ConfigForThisTestCase(data[test_case]):
continue
for key in data[test_case].iterkeys():
if isinstance(data[test_case][key], (str, unicode)):
value = data[test_case][key]
if not value or value == 'None':
self.params[key] = None
else:
self.params[key] = value
elif isinstance(data[test_case][key], dict):
if (key == 'title' and
self.test_info['testID'] in data[test_case][key]):
self.params[key] = data[test_case][key][self.test_info['testID']]
self.test_info['key_word'] = self.params[key]
else:
self.params[key] = data[test_case][key]
elif isinstance(data[test_case][key], list):
if key == 'testID':
for i in range(0, len(data[test_case][key])):
test_id = data[test_case][key][i]
if str(self.test_info['testID']) == test_id:
test_id_matched = True
elif key == 'comment':
# do not process comment lines
continue
else:
self.params[key] = data[test_case][key]
else:
info = self.log.CreateErrorInfo(
'Low', 'Warning: Configuration Parameter type unknown '
'in file ' + self.test_info['configFile'] + ': '
+ test_case + ':' + key)
self.log.SendLine(self.test_info, info)
if not test_id_matched:
info = self.log.CreateErrorInfo(
'Low', 'Warning: No configuration found for this test '
'case in file: ' + self.test_info['configFile'])
self.log.SendLine(self.test_info, info)
os.sys.exit(1)
return True
def ParseConfig(self):
"""Parse the configuration file for test case.
The structure of the configuration file:
Each test case configuration begins with its testID in a sigle line,
the following lines have the format:
############# configuration for Test 11362 ########################
testID: 11362
parameter_1 = value_1
parameter_2 = value_2
parameter_3 = value_3
...
############# configuration for Test 11363 ########################
testID: 11363
parameter_1 = value_1
parameter_2 = value_2
parameter_3 = value_3
...
############# comment line start with '#' #########################
Returns:
reuturn 1 if succeed
"""
# dictionary for params
try:
self.cfg_file = open(self.test_info['configFile'], 'r')
except IOError:
print 'Cannot open the input file, return'
return os.sys.exit(1)
cfg_list = self.cfg_file.readlines()
test_id_matched = False
for line in cfg_list:
s = line.strip().strip('\n')
m = re.match('testID: ' + str(self.test_info['testID']), s)
if (m is not None) and not s.startswith('#'):
# test case matching succeed
test_id_matched = True
# get next line index:
index = cfg_list.index(line)+1
if index >= len(cfg_list):
# end of file reached
info = self.log.createErrorInfo(
'Low',
'Warning: No configuration found for this test case in file: '
+ self.test_info['configFile'])
self.log.sendLine(self.test_info, info)
s = cfg_list[index]
s = s.strip()
s = s.split('#', 2)[0]
m = re.match('testID: ', s)
while m is None:
# not reaching configuration for next test case,
# read and create current test case configuration data as dictionary
# format: parameter = value
if s is not '' and not s.startswith('#'):
name = s.split('=')[0].strip()
value = s.split('=')[1].strip()
if not value or value == 'None':
self.params[name] = None
else:
self.params[name] = value
index += 1
if index == len(cfg_list): break
s = cfg_list[index]
s = s.strip()
m = re.match('testID: ', s)
# all configuration parsed
self.cfg_file.close()
if not test_id_matched:
# end of file reached
info = self.log.CreateErrorInfo(
'Low', 'Warning: No configuration found for this test case in file: '
+ self.test_info['configFile'])
self.log.SendLine(self.test_info, info)
os.sys.exit(1)
return True
def AddConfigParams(self, par='par', value='value'):
"""Add new parameter to the params list.
Args:
par: parameter name
value: parameter value
"""
if par and value:
self.params[par.strip()] = value.strip()
def __del__(self):
pass
#################### define your own functions here #######################
def Run(self):
"""Starts to Run the test case."""
##### Add your code here -- BEGIN #######
print 'Test Started...'
print 'Test Completed...'
##### Add your code here -- END #######
def Destructor(self):
pass