| #!/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 |
