xref: /unit/test/unit/applications/proto.py (revision 1730:44912af5b3e6)
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