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