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