xref: /unit/test/unit/applications/proto.py (revision 1730)
11596Szelenkov@nginx.comimport os
21019Szelenkov@nginx.comimport re
31019Szelenkov@nginx.comimport time
41477Szelenkov@nginx.com
51019Szelenkov@nginx.comfrom unit.control import TestControl
6*1730Szelenkov@nginx.comfrom unit.option import option
71019Szelenkov@nginx.com
81019Szelenkov@nginx.com
91019Szelenkov@nginx.comclass TestApplicationProto(TestControl):
101670Smax.romanov@nginx.com    application_type = None
111670Smax.romanov@nginx.com
121019Szelenkov@nginx.com    def sec_epoch(self):
131019Szelenkov@nginx.com        return time.mktime(time.gmtime())
141019Szelenkov@nginx.com
151019Szelenkov@nginx.com    def date_to_sec_epoch(self, date, template='%a, %d %b %Y %H:%M:%S %Z'):
161019Szelenkov@nginx.com        return time.mktime(time.strptime(date, template))
171019Szelenkov@nginx.com
181027Szelenkov@nginx.com    def search_in_log(self, pattern, name='unit.log'):
191654Szelenkov@nginx.com        with open(option.temp_dir + '/' + name, 'r', errors='ignore') as f:
201019Szelenkov@nginx.com            return re.search(pattern, f.read())
211028Szelenkov@nginx.com
221028Szelenkov@nginx.com    def wait_for_record(self, pattern, name='unit.log'):
231028Szelenkov@nginx.com        for i in range(50):
241028Szelenkov@nginx.com            found = self.search_in_log(pattern, name)
251028Szelenkov@nginx.com
261028Szelenkov@nginx.com            if found is not None:
271028Szelenkov@nginx.com                break
281028Szelenkov@nginx.com
291028Szelenkov@nginx.com            time.sleep(0.1)
301028Szelenkov@nginx.com
311028Szelenkov@nginx.com        return found
321050Szelenkov@nginx.com
331670Smax.romanov@nginx.com    def get_application_type(self):
341596Szelenkov@nginx.com        current_test = (
351596Szelenkov@nginx.com            os.environ.get('PYTEST_CURRENT_TEST').split(':')[-1].split(' ')[0]
361596Szelenkov@nginx.com        )
371596Szelenkov@nginx.com
381670Smax.romanov@nginx.com        return option.generated_tests.get(current_test, self.application_type)
391596Szelenkov@nginx.com
401303St.nateldemoura@f5.com    def _load_conf(self, conf, **kwargs):
411303St.nateldemoura@f5.com        if 'applications' in conf:
421303St.nateldemoura@f5.com            for app in conf['applications'].keys():
431303St.nateldemoura@f5.com                app_conf = conf['applications'][app]
441303St.nateldemoura@f5.com                if 'user' in kwargs:
451303St.nateldemoura@f5.com                    app_conf['user'] = kwargs['user']
461303St.nateldemoura@f5.com
471303St.nateldemoura@f5.com                if 'group' in kwargs:
481303St.nateldemoura@f5.com                    app_conf['group'] = kwargs['group']
491303St.nateldemoura@f5.com
501303St.nateldemoura@f5.com                if 'isolation' in kwargs:
511303St.nateldemoura@f5.com                    app_conf['isolation'] = kwargs['isolation']
521303St.nateldemoura@f5.com
531596Szelenkov@nginx.com        assert 'success' in self.conf(conf), 'load application configuration'
54