xref: /unit/test/unit/applications/proto.py (revision 1971:3410f9d2a662)
11596Szelenkov@nginx.comimport os
21019Szelenkov@nginx.comimport re
31019Szelenkov@nginx.comimport time
41477Szelenkov@nginx.com
51019Szelenkov@nginx.comfrom unit.control import TestControl
6*1971Szelenkov@nginx.comfrom unit.log import Log
71730Szelenkov@nginx.comfrom unit.option import option
81019Szelenkov@nginx.com
91019Szelenkov@nginx.com
101019Szelenkov@nginx.comclass TestApplicationProto(TestControl):
111670Smax.romanov@nginx.com    application_type = None
121670Smax.romanov@nginx.com
131019Szelenkov@nginx.com    def sec_epoch(self):
141019Szelenkov@nginx.com        return time.mktime(time.gmtime())
151019Szelenkov@nginx.com
161019Szelenkov@nginx.com    def date_to_sec_epoch(self, date, template='%a, %d %b %Y %H:%M:%S %Z'):
171019Szelenkov@nginx.com        return time.mktime(time.strptime(date, template))
181019Szelenkov@nginx.com
191850Smax.romanov@nginx.com    def findall(self, pattern, name='unit.log'):
201850Smax.romanov@nginx.com        with Log.open(name) as f:
211850Smax.romanov@nginx.com            return re.findall(pattern, f.read())
221850Smax.romanov@nginx.com
231027Szelenkov@nginx.com    def search_in_log(self, pattern, name='unit.log'):
241850Smax.romanov@nginx.com        with Log.open(name) as f:
251019Szelenkov@nginx.com            return re.search(pattern, f.read())
261028Szelenkov@nginx.com
271804Szelenkov@nginx.com    def wait_for_record(self, pattern, name='unit.log', wait=150):
281850Smax.romanov@nginx.com        with Log.open(name) as f:
291850Smax.romanov@nginx.com            for i in range(wait):
301850Smax.romanov@nginx.com                found = re.search(pattern, f.read())
311028Szelenkov@nginx.com
321850Smax.romanov@nginx.com                if found is not None:
331850Smax.romanov@nginx.com                    break
341028Szelenkov@nginx.com
351850Smax.romanov@nginx.com                time.sleep(0.1)
361028Szelenkov@nginx.com
371028Szelenkov@nginx.com        return found
381050Szelenkov@nginx.com
391670Smax.romanov@nginx.com    def get_application_type(self):
401596Szelenkov@nginx.com        current_test = (
411596Szelenkov@nginx.com            os.environ.get('PYTEST_CURRENT_TEST').split(':')[-1].split(' ')[0]
421596Szelenkov@nginx.com        )
431596Szelenkov@nginx.com
441670Smax.romanov@nginx.com        return option.generated_tests.get(current_test, self.application_type)
451596Szelenkov@nginx.com
461303St.nateldemoura@f5.com    def _load_conf(self, conf, **kwargs):
471303St.nateldemoura@f5.com        if 'applications' in conf:
481303St.nateldemoura@f5.com            for app in conf['applications'].keys():
491303St.nateldemoura@f5.com                app_conf = conf['applications'][app]
501303St.nateldemoura@f5.com
511912So.canty@f5.com                for key in [
521912So.canty@f5.com                    'user',
531912So.canty@f5.com                    'group',
541912So.canty@f5.com                    'isolation',
551912So.canty@f5.com                    'processes',
561912So.canty@f5.com                    'threads',
571912So.canty@f5.com                ]:
581912So.canty@f5.com                    if key in kwargs:
591912So.canty@f5.com                        app_conf[key] = kwargs[key]
601303St.nateldemoura@f5.com
611596Szelenkov@nginx.com        assert 'success' in self.conf(conf), 'load application configuration'
62