1import os 2import re 3import time 4 5from unit.control import TestControl 6from conftest import option 7 8 9class TestApplicationProto(TestControl): 10 def sec_epoch(self): 11 return time.mktime(time.gmtime()) 12 13 def date_to_sec_epoch(self, date, template='%a, %d %b %Y %H:%M:%S %Z'): 14 return time.mktime(time.strptime(date, template)) 15 16 def search_in_log(self, pattern, name='unit.log'): 17 with open(self.temp_dir + '/' + name, 'r', errors='ignore') as f: 18 return re.search(pattern, f.read()) 19 20 def wait_for_record(self, pattern, name='unit.log'): 21 for i in range(50): 22 found = self.search_in_log(pattern, name) 23 24 if found is not None: 25 break 26 27 time.sleep(0.1) 28 29 return found 30 31 def get_appication_type(self): 32 current_test = ( 33 os.environ.get('PYTEST_CURRENT_TEST').split(':')[-1].split(' ')[0] 34 ) 35 36 if current_test in option.generated_tests: 37 return option.generated_tests[current_test] 38 39 return None 40 41 def _load_conf(self, conf, **kwargs): 42 if 'applications' in conf: 43 for app in conf['applications'].keys(): 44 app_conf = conf['applications'][app] 45 if 'user' in kwargs: 46 app_conf['user'] = kwargs['user'] 47 48 if 'group' in kwargs: 49 app_conf['group'] = kwargs['group'] 50 51 if 'isolation' in kwargs: 52 app_conf['isolation'] = kwargs['isolation'] 53 54 assert 'success' in self.conf(conf), 'load application configuration' 55