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