5132620618

pending completion

Build # 5132620618

Build Type

push

github

Committed by

ylobankov

Commit Message

Prettify messages about log files

The ideas behind the change:

* The log is sometimes from a tarantool instance, but a luatest based
  test may print logs from another process to stderr. Use more general
  term 'log' to don't confuse anyone.
* Replace term 'instance' with 'test-run server' due to the same
  reasons.
* Make zero size and no file situations more explicit.
* Catch and report 'no logfile property' situation (shouldn't occur, but
  was semi-handled by this code previously for some reason).
* Reduce code reusability for the sake of readability and to ease future
  modifications.

Run Details

758 of 1554 branches covered (48.78%)

Branch coverage included in aggregate %.

22 of 22 new or added lines in 1 file covered. (100.0%)

2929 of 4328 relevant lines covered (67.68%)

0.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.48

/lib/app_server.py

import errno
import glob
import os
import re
import shutil
import signal
import sys

from gevent.subprocess import Popen

from lib.colorer import color_stdout
from lib.colorer import color_log
from lib.colorer import qa_notice
from lib.options import Options
from lib.preprocessor import TestState
from lib.sampler import sampler
from lib.server import Server
from lib.server import DEFAULT_SNAPSHOT_NAME
from lib.tarantool_server import Test
from lib.tarantool_server import TarantoolServer
from lib.tarantool_server import TarantoolStartError
from lib.utils import format_process
from lib.utils import signame
from lib.utils import warn_unix_socket
from threading import Timer
from lib.test import TestRunGreenlet, TestExecutionError


def timeout_handler(server_process, test_timeout):
    color_stdout("Test timeout of %d secs reached\t" % test_timeout, schema='error')
    server_process.kill()


def run_server(execs, cwd, server, logfile, retval, test_id):
    os.putenv("LISTEN", server.listen_uri)
    with open(logfile, 'ab') as f:
        server.process = Popen(execs, stdout=sys.stdout, stderr=f, cwd=cwd)
    sampler.register_process(server.process.pid, test_id, server.name)
    test_timeout = Options().args.test_timeout
    timer = Timer(test_timeout, timeout_handler, (server.process, test_timeout))
    timer.start()
    retval['returncode'] = server.process.wait()
    timer.cancel()
    server.process = None


class AppTest(Test):
    def execute(self, server):
        super(AppTest, self).execute(server)
        ts = TestState(self.suite_ini, None, TarantoolServer,
                       self.run_params,
                       default_server_no_connect=server)
        self.inspector.set_parser(ts)

        execs = server.prepare_args()
        retval = dict()
        tarantool = TestRunGreenlet(run_server, execs, server.vardir, server,
                                    server.logfile, retval, self.id)
        self.current_test_greenlet = tarantool

        # Copy the snapshot right before starting the server.
        # Otherwise pretest_clean() would remove it.
        if server.snapshot_path:
            snapshot_dest = os.path.join(server.vardir, DEFAULT_SNAPSHOT_NAME)
            color_log("Copying snapshot {} to {}\n".format(
                server.snapshot_path, snapshot_dest))
            shutil.copy(server.snapshot_path, snapshot_dest)

        try:
            tarantool.start()
            tarantool.join()
        except TarantoolStartError:
            # A non-default server failed to start.
            raise TestExecutionError
        finally:
            self.teardown(server, ts)
        if retval.get('returncode', None) != 0:
            raise TestExecutionError

    def teardown(self, server, ts):
        # Stop any servers created by the test, except the
        # default one.
        #
        # See a comment in LuaTest.execute() for motivation of
        # SIGKILL usage.
        ts.stop_nondefault(signal=signal.SIGKILL)

        # When a supplementary (non-default) server fails, we
        # should not leave the process that executes an app test.
        # Let's kill it.
        #
        # Reuse AppServer.stop() code for convenience.
        server.stop(signal=signal.SIGKILL)


class AppServer(Server):
    """A dummy server implementation for application server tests"""
    def __new__(cls, ini=None, *args, **kwargs):
        cls = Server.get_mixed_class(cls, ini)
        return object.__new__(cls)

    def __init__(self, _ini=None, test_suite=None):
        ini = dict(vardir=None)
        ini.update({} if _ini is None else _ini)
        Server.__init__(self, ini, test_suite)
        self.testdir = os.path.abspath(os.curdir)
        self.vardir = ini['vardir']
        self.builddir = ini['builddir']
        self.lua_libs = ini['lua_libs']
        self.name = 'app_server'
        self.process = None
        self.localhost = '127.0.0.1'
        self.use_unix_sockets_iproto = ini['use_unix_sockets_iproto']

    @property
    def logfile(self):
        # remove suite name using basename
        test_name = os.path.basename(self.current_test.name)
        # add :conf_name if any
        if self.current_test.conf_name is not None:
            test_name += ':' + self.current_test.conf_name
        # add '.tarantool.log'
        file_name = test_name + '.tarantool.log'
        # put into vardir
        return os.path.join(self.vardir, file_name)

    def prepare_args(self, args=[]):
        # Disable stdout bufferization.
        cli_args = [self.binary, '-e', "io.stdout:setvbuf('no')"]

        # Disable schema upgrade if requested.
        if self.disable_schema_upgrade:
            cli_args.extend(['-e', self.DISABLE_AUTO_UPGRADE])

        # Add path to the script (the test).
        cli_args.extend([os.path.join(os.getcwd(), self.current_test.name)])

        # Add extra args if provided.
        cli_args.extend(args)

        return cli_args

    def deploy(self, vardir=None, silent=True, need_init=True):
        self.vardir = vardir
        if not os.access(self.vardir, os.F_OK):
            os.makedirs(self.vardir)
        if self.lua_libs:
            for i in self.lua_libs:
                source = os.path.join(self.testdir, i)
                try:
                    if os.path.isdir(source):
                        shutil.copytree(source,
                                        os.path.join(self.vardir,
                                                     os.path.basename(source)))
                    else:
                        shutil.copy(source, self.vardir)
                except IOError as e:
                    if (e.errno == errno.ENOENT):
                        continue
                    raise
        if self.use_unix_sockets_iproto:
            path = os.path.join(self.vardir, self.name + ".i")
            warn_unix_socket(path)
            self.listen_uri = path
        else:
            self.listen_uri = self.localhost + ':0'
        shutil.copy(os.path.join(self.TEST_RUN_DIR, 'test_run.lua'),
                    self.vardir)

        # Note: we don't know the instance name of the tarantool server, so
        # cannot check length of path of *.control unix socket created by it.
        # So for 'app' tests type we don't check *.control unix sockets paths.

    def stop(self, silent=True, signal=signal.SIGTERM):
        # FIXME: Extract common parts of AppServer.stop() and
        # TarantoolServer.stop() to an utility function.

        color_log('DEBUG: [app server] Stopping the server...\n',
                  schema='info')

        if not self.process:
            color_log(' | Nothing to do: the process does not exist\n',
                      schema='info')
            return

        if self.process.returncode:
            if self.process.returncode < 0:
                signaled_by = -self.process.returncode
                color_log(' | Nothing to do: the process was terminated by '
                          'signal {} ({})\n'.format(signaled_by,
                                                    signame(signaled_by)),
                          schema='info')
            else:
                color_log(' | Nothing to do: the process was exited with code '
                          '{}\n'.format(self.process.returncode),
                          schema='info')
            return

        color_log(' | Sending signal {0} ({1}) to {2}\n'.format(
                  signal, signame(signal),
                  format_process(self.process.pid)))
        try:
            self.process.send_signal(signal)
        except OSError:
            pass

        # Waiting for stopping the server. If the timeout
        # reached, send SIGKILL.
        timeout = 5

        def kill():
            qa_notice('The app server does not stop during {} '
                      'seconds after the {} ({}) signal.\n'
                      'Info: {}\n'
                      'Sending SIGKILL...'.format(
                          timeout, signal, signame(signal),
                          format_process(self.process.pid)))
            try:
                self.process.kill()
            except OSError:
                pass

        timer = Timer(timeout, kill)
        timer.start()
        self.process.wait()
        timer.cancel()

    @classmethod
    def find_exe(cls, builddir):
        cls.builddir = builddir
        cls.binary = TarantoolServer.binary
        cls.debug = bool(re.findall(r'^Target:.*-Debug$', str(cls.version()),
                                    re.M))

    @staticmethod
    def find_tests(test_suite, suite_path):
        def patterned(test_name, patterns):
            answer = []
            for i in patterns:
                if test_name.find(i) != -1:
                    answer.append(test_name)
            return answer

        def is_correct(run):
            return test_suite.args.conf is None or test_suite.args.conf == run

        test_suite.ini['suite'] = suite_path

        test_names = sorted(glob.glob(os.path.join(suite_path, "*.test.lua")))
        test_names = Server.exclude_tests(test_names, test_suite.args.exclude)
        test_names = sum(map((lambda x: patterned(x, test_suite.args.tests)),
                             test_names), [])
        tests = []

        for test_name in test_names:
            runs = test_suite.get_multirun_params(test_name)
            if runs:
                tests.extend([AppTest(
                    test_name,
                    test_suite.args,
                    test_suite.ini,
                    params=params,
                    conf_name=conf_name
                ) for conf_name, params in runs.items()
                    if is_correct(conf_name)])
            else:
                tests.append(AppTest(test_name,
                                     test_suite.args,
                                     test_suite.ini))

        test_suite.tests = tests

1	import errno	1✔
2	import glob	1✔
3	import os	1✔
4	import re	1✔
5	import shutil	1✔
6	import signal	1✔
7	import sys	1✔
8
9	from gevent.subprocess import Popen	1✔
10
11	from lib.colorer import color_stdout	1✔
12	from lib.colorer import color_log	1✔
13	from lib.colorer import qa_notice	1✔
14	from lib.options import Options	1✔
15	from lib.preprocessor import TestState	1✔
16	from lib.sampler import sampler	1✔
17	from lib.server import Server	1✔
18	from lib.server import DEFAULT_SNAPSHOT_NAME	1✔
19	from lib.tarantool_server import Test	1✔
20	from lib.tarantool_server import TarantoolServer	1✔
21	from lib.tarantool_server import TarantoolStartError	1✔
22	from lib.utils import format_process	1✔
23	from lib.utils import signame	1✔
24	from lib.utils import warn_unix_socket	1✔
25	from threading import Timer	1✔
26	from lib.test import TestRunGreenlet, TestExecutionError	1✔
27
28
29	def timeout_handler(server_process, test_timeout):	1✔
30	color_stdout("Test timeout of %d secs reached\t" % test_timeout, schema='error')	×
31	server_process.kill()	×
32
33
34	def run_server(execs, cwd, server, logfile, retval, test_id):	1✔
35	os.putenv("LISTEN", server.listen_uri)	1✔
36	with open(logfile, 'ab') as f:	1✔
37	server.process = Popen(execs, stdout=sys.stdout, stderr=f, cwd=cwd)	1✔
38	sampler.register_process(server.process.pid, test_id, server.name)	1✔
39	test_timeout = Options().args.test_timeout	1✔
40	timer = Timer(test_timeout, timeout_handler, (server.process, test_timeout))	1✔
41	timer.start()	1✔
42	retval['returncode'] = server.process.wait()	1✔
43	timer.cancel()	1✔
44	server.process = None	1✔
45
46
47	class AppTest(Test):	1✔
48	def execute(self, server):	1✔
49	super(AppTest, self).execute(server)	1✔
50	ts = TestState(self.suite_ini, None, TarantoolServer,	1✔
51	self.run_params,
52	default_server_no_connect=server)
53	self.inspector.set_parser(ts)	1✔
54
55	execs = server.prepare_args()	1✔
56	retval = dict()	1✔
57	tarantool = TestRunGreenlet(run_server, execs, server.vardir, server,	1✔
58	server.logfile, retval, self.id)
59	self.current_test_greenlet = tarantool	1✔
60
61	# Copy the snapshot right before starting the server.
62	# Otherwise pretest_clean() would remove it.
63	if server.snapshot_path:	1!
64	snapshot_dest = os.path.join(server.vardir, DEFAULT_SNAPSHOT_NAME)	×
65	color_log("Copying snapshot {} to {}\n".format(	×
66	server.snapshot_path, snapshot_dest))
67	shutil.copy(server.snapshot_path, snapshot_dest)	×
68
69	try:	1✔
70	tarantool.start()	1✔
71	tarantool.join()	1✔
72	except TarantoolStartError:	×
73	# A non-default server failed to start.
74	raise TestExecutionError	×
75	finally:
76	self.teardown(server, ts)	1!
77	if retval.get('returncode', None) != 0:	1!
78	raise TestExecutionError	×
79
80	def teardown(self, server, ts):	1✔
81	# Stop any servers created by the test, except the
82	# default one.
83	#
84	# See a comment in LuaTest.execute() for motivation of
85	# SIGKILL usage.
86	ts.stop_nondefault(signal=signal.SIGKILL)	1✔
87
88	# When a supplementary (non-default) server fails, we
89	# should not leave the process that executes an app test.
90	# Let's kill it.
91	#
92	# Reuse AppServer.stop() code for convenience.
93	server.stop(signal=signal.SIGKILL)	1✔
94
95
96	class AppServer(Server):	1✔
97	"""A dummy server implementation for application server tests"""
98	def __new__(cls, ini=None, args, *kwargs):	1✔
99	cls = Server.get_mixed_class(cls, ini)	1✔
100	return object.__new__(cls)	1✔
101
102	def __init__(self, _ini=None, test_suite=None):	1✔
103	ini = dict(vardir=None)	1✔
104	ini.update({} if _ini is None else _ini)	1✔
105	Server.__init__(self, ini, test_suite)	1✔
106	self.testdir = os.path.abspath(os.curdir)	1✔
107	self.vardir = ini['vardir']	1✔
108	self.builddir = ini['builddir']	1✔
109	self.lua_libs = ini['lua_libs']	1✔
110	self.name = 'app_server'	1✔
111	self.process = None	1✔
112	self.localhost = '127.0.0.1'	1✔
113	self.use_unix_sockets_iproto = ini['use_unix_sockets_iproto']	1✔
114
115	@property	1✔
116	def logfile(self):	1✔
117	# remove suite name using basename
118	test_name = os.path.basename(self.current_test.name)	1✔
119	# add :conf_name if any
120	if self.current_test.conf_name is not None:	1!
121	test_name += ':' + self.current_test.conf_name	×
122	# add '.tarantool.log'
123	file_name = test_name + '.tarantool.log'	1✔
124	# put into vardir
125	return os.path.join(self.vardir, file_name)	1✔
126
127	def prepare_args(self, args=[]):	1✔
128	# Disable stdout bufferization.
129	cli_args = [self.binary, '-e', "io.stdout:setvbuf('no')"]	1✔
130
131	# Disable schema upgrade if requested.
132	if self.disable_schema_upgrade:	1!
133	cli_args.extend(['-e', self.DISABLE_AUTO_UPGRADE])	×
134
135	# Add path to the script (the test).
136	cli_args.extend([os.path.join(os.getcwd(), self.current_test.name)])	1✔
137
138	# Add extra args if provided.
139	cli_args.extend(args)	1✔
140
141	return cli_args	1✔
142
143	def deploy(self, vardir=None, silent=True, need_init=True):	1✔
144	self.vardir = vardir	1✔
145	if not os.access(self.vardir, os.F_OK):	1✔
146	os.makedirs(self.vardir)	1✔
147	if self.lua_libs:	1!
148	for i in self.lua_libs:	1!
149	source = os.path.join(self.testdir, i)	×
150	try:	×
151	if os.path.isdir(source):	×
152	shutil.copytree(source,	×
153	os.path.join(self.vardir,
154	os.path.basename(source)))
155	else:
156	shutil.copy(source, self.vardir)	×
157	except IOError as e:	×
158	if (e.errno == errno.ENOENT):	×
159	continue	×
160	raise	×
161	if self.use_unix_sockets_iproto:	1!
162	path = os.path.join(self.vardir, self.name + ".i")	×
163	warn_unix_socket(path)	×
164	self.listen_uri = path	×
165	else:
166	self.listen_uri = self.localhost + ':0'	1✔
167	shutil.copy(os.path.join(self.TEST_RUN_DIR, 'test_run.lua'),	1✔
168	self.vardir)
169
170	# Note: we don't know the instance name of the tarantool server, so
171	# cannot check length of path of *.control unix socket created by it.
172	# So for 'app' tests type we don't check *.control unix sockets paths.
173
174	def stop(self, silent=True, signal=signal.SIGTERM):	1✔
175	# FIXME: Extract common parts of AppServer.stop() and
176	# TarantoolServer.stop() to an utility function.
177
178	color_log('DEBUG: [app server] Stopping the server...\n',	1✔
179	schema='info')
180
181	if not self.process:	1!
182	color_log(' \| Nothing to do: the process does not exist\n',	1✔
183	schema='info')
184	return	1✔
185
186	if self.process.returncode:	×
187	if self.process.returncode < 0:	×
188	signaled_by = -self.process.returncode	×
189	color_log(' \| Nothing to do: the process was terminated by '	×
190	'signal {} ({})\n'.format(signaled_by,
191	signame(signaled_by)),
192	schema='info')
193	else:
194	color_log(' \| Nothing to do: the process was exited with code '	×
195	'{}\n'.format(self.process.returncode),
196	schema='info')
197	return	×
198
199	color_log(' \| Sending signal {0} ({1}) to {2}\n'.format(	×
200	signal, signame(signal),
201	format_process(self.process.pid)))
202	try:	×
203	self.process.send_signal(signal)	×
204	except OSError:	×
205	pass	×
206
207	# Waiting for stopping the server. If the timeout
208	# reached, send SIGKILL.
209	timeout = 5	×
210
211	def kill():	×
212	qa_notice('The app server does not stop during {} '	×
213	'seconds after the {} ({}) signal.\n'
214	'Info: {}\n'
215	'Sending SIGKILL...'.format(
216	timeout, signal, signame(signal),
217	format_process(self.process.pid)))
218	try:	×
219	self.process.kill()	×
220	except OSError:	×
221	pass	×
222
223	timer = Timer(timeout, kill)	×
224	timer.start()	×
225	self.process.wait()	×
226	timer.cancel()	×
227
228	@classmethod	1✔
229	def find_exe(cls, builddir):	1✔
230	cls.builddir = builddir	1✔
231	cls.binary = TarantoolServer.binary	1✔
232	cls.debug = bool(re.findall(r'^Target:.*-Debug$', str(cls.version()),	1✔
233	re.M))
234
235	@staticmethod	1✔
236	def find_tests(test_suite, suite_path):	1✔
237	def patterned(test_name, patterns):	1✔
238	answer = []	1✔
239	for i in patterns:	1✔
240	if test_name.find(i) != -1:	1!
241	answer.append(test_name)	1✔
242	return answer	1✔
243
244	def is_correct(run):	1✔
245	return test_suite.args.conf is None or test_suite.args.conf == run	×
246
247	test_suite.ini['suite'] = suite_path	1✔
248
249	test_names = sorted(glob.glob(os.path.join(suite_path, "*.test.lua")))	1✔
250	test_names = Server.exclude_tests(test_names, test_suite.args.exclude)	1✔
251	test_names = sum(map((lambda x: patterned(x, test_suite.args.tests)),	1✔
252	test_names), [])
253	tests = []	1✔
254
255	for test_name in test_names:	1✔
256	runs = test_suite.get_multirun_params(test_name)	1✔
257	if runs:	1!
258	tests.extend([AppTest(	×
259	test_name,
260	test_suite.args,
261	test_suite.ini,
262	params=params,
263	conf_name=conf_name
264	) for conf_name, params in runs.items()
265	if is_correct(conf_name)])
266	else:
267	tests.append(AppTest(test_name,	1✔
268	test_suite.args,
269	test_suite.ini))
270
271	test_suite.tests = tests	1✔

tarantool / test-run / 5132620618

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous