11499903942

Committed 24 Oct 2024 01:05PM UTC coverage: 62.656%. Remained the same

Build # 11499903942

Build Type

push

github

Committed by

ylobankov

Commit Message

Follow-up fix for parsing non-utf8 chars (part 2)

This patch is similar to commit 52d3e4f7c90f ("Follow-up fix for parsing
non-utf8 chars"), but fixes the `--verbose` output, e.g.:

```
$ ./test-run.py --verbose some_test

[...]
[001] Worker "001_xxx-luatest" received the following error; stopping...
[001] Traceback (most recent call last):
[001]   File "./tarantool/test-run/lib/worker.py", line 357, in run_task
[001]     short_status, duration = self.suite.run_test(
[001]                              ^^^^^^^^^^^^^^^^^^^^
[001]   File "./tarantool/test-run/lib/test_suite.py", line 271, in run_test
[001]     short_status = test.run(server)
[001]                    ^^^^^^^^^^^^^^^^
[001]   File "./tarantool/test-run/lib/test.py", line 232, in run
[001]     color_stdout(f.read(), schema='log')
[001]                  ^^^^^^^^
[001]   File "<frozen codecs>", line 322, in decode
[001] UnicodeDecodeError: 'utf-8' codec can't decode byte 0xa3 in position 115111: invalid start byte
[001]
[001] Exception: 'utf-8' codec can't decode byte 0xa3 in position 115111: invalid start byte
```

Run Details

772 of 1586 branches covered (48.68%)

Branch coverage included in aggregate %.

0 of 1 new or added line in 1 file covered. (0.0%)

1 existing line in 1 file now uncovered.

2988 of 4415 relevant lines covered (67.68%)

0.68 hits per line

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

50.73

/lib/test.py

import filecmp
import gevent
import os
import pprint
import re
import shutil
import sys
import traceback
from functools import partial

from lib import Options
from lib.colorer import color_stdout
from lib.utils import assert_bytes
from lib.utils import non_empty_valgrind_logs
from lib.utils import print_tail_n
from lib.utils import print_unidiff as utils_print_unidiff
from lib.utils import safe_makedirs
from lib.utils import str_to_bytes
from lib import pytap13


class TestExecutionError(OSError):
    """To be raised when a test execution fails"""
    pass


class TestRunGreenlet(gevent.Greenlet):
    def __init__(self, green_callable, *args, **kwargs):
        self.callable = green_callable
        self.callable_args = args
        self.callable_kwargs = kwargs
        super(TestRunGreenlet, self).__init__()

    def _run(self, *args, **kwargs):
        self.callable(*self.callable_args, **self.callable_kwargs)

    def __repr__(self):
        return "<TestRunGreenlet at {0} info='{1}'>".format(
            hex(id(self)), getattr(self, "info", None))


class FilteredStream:
    """Helper class to filter .result file output"""
    def __init__(self, filename):
        self.stream = open(filename, "wb+")
        self.filters = []
        self.inspector = None

    def write_bytes(self, fragment):
        """ The same as ``write()``, but accepts ``<bytes>`` as
            input.
        """
        assert_bytes(fragment)
        skipped = False
        for line in fragment.splitlines(True):
            original_len = len(line.strip())
            for pattern, replacement in self.filters:
                line = re.sub(pattern, replacement, line)
                # don't write lines that are completely filtered out:
                skipped = original_len and not line.strip()
                if skipped:
                    break
            if not skipped:
                self.stream.write(line)

    def write(self, fragment):
        """ Apply all filters, then write result to the underlying
            stream.

            Do line-oriented filtering: the fragment doesn't have
            to represent just one line.

            Accepts ``<str>`` as input, just like the standard
            ``sys.stdout.write()``.
        """
        self.write_bytes(str_to_bytes(fragment))

    def push_filter(self, pattern, replacement):
        self.filters.append([str_to_bytes(pattern), str_to_bytes(replacement)])

    def pop_filter(self):
        self.filters.pop()

    def clear_all_filters(self):
        self.filters = []

    def close(self):
        self.clear_all_filters()
        self.stream.close()

    def flush(self):
        self.stream.flush()

    def fileno(self):
        """ May be used for direct writting. Discards any filters.
        """
        return self.stream.fileno()


def get_filename_by_test(postfix, test_name):
    """For <..>/<name>_test.* or <..>/<name>.test.* return <name> + postfix

    Examples:
        postfix='.result', test_name='foo/bar.test.lua' => return 'bar.result'
        postfix='.reject', test_name='bar_test.lua' => return 'bar.reject'
    """
    rg = re.compile(r'[._]test.*')
    return os.path.basename(rg.sub(postfix, test_name))


get_reject = partial(get_filename_by_test, '.reject')
get_result = partial(get_filename_by_test, '.result')
get_skipcond = partial(get_filename_by_test, '.skipcond')


class Test(object):
    """An individual test file. A test object can run itself
    and remembers completion state of the run.

    If file <test_name>.skipcond is exists it will be executed before
    test and if it sets self.skip to True value the test will be skipped.
    """

    def __init__(self, name, args, suite_ini, params={}, conf_name=None):
        """Initialize test properties: path to test file, path to
        temporary result file, path to the client program, test status."""
        self.name = name
        self.args = args
        self.suite_ini = suite_ini
        self.result = os.path.join(suite_ini['suite'], get_result(name))
        self.skip_cond = os.path.join(suite_ini['suite'], get_skipcond(name))
        self.tmp_result = os.path.join(suite_ini['vardir'], get_result(name))
        self.var_suite_path = os.path.join(suite_ini['vardir'], 'rejects',
                                           suite_ini['suite'])
        self.reject = os.path.join(self.var_suite_path, get_reject(name))
        self.is_executed = False
        self.is_executed_ok = None
        self.is_equal_result = None
        self.is_valgrind_clean = True
        self.is_terminated = False
        self.run_params = params
        self.conf_name = conf_name

        # filled in execute() when a greenlet runs
        self.current_test_greenlet = None

        # prevent double/triple reporting
        self.is_crash_reported = False

    @property
    def id(self):
        return self.name, self.conf_name

    def passed(self):
        """Return true if this test was run successfully."""
        return (self.is_executed and
                self.is_executed_ok and
                self.is_equal_result)

    def execute(self, server):
        # Note: don't forget to set 'server.current_test = self' in
        # inherited classes. Crash reporting relying on that.
        server.current_test = self
        # All the test runs must be isolated between each other on each worker.
        server.pretest_clean()

    def run(self, server):
        """ Execute the test assuming it's a python program.  If the test
            aborts, print its output to stdout, and raise an exception. Else,
            comprare result and reject files.  If there is a difference, print
            it to stdout.

            Returns short status of the test as a string: 'skip', 'pass',
            'new', 'updated' or 'fail'.
            There is also one possible value for short_status, 'disabled',
            but it returned in the caller, TestSuite.run_test().
        """

        # Note: test was created before certain worker become known, so we need
        # to update temporary result directory here as it depends on 'vardir'.
        self.tmp_result = os.path.join(self.suite_ini['vardir'],
                                       os.path.basename(self.result))

        diagnostics = "unknown"
        save_stdout = sys.stdout
        try:
            self.skip = False
            if os.path.exists(self.skip_cond):
                sys.stdout = FilteredStream(self.tmp_result)
                stdout_fileno = sys.stdout.stream.fileno()
                new_globals = dict(locals(), **server.__dict__)
                with open(self.skip_cond, 'r') as f:
                    code = compile(f.read(), self.skip_cond, 'exec')
                    exec(code, new_globals)
                sys.stdout.close()
                sys.stdout = save_stdout
            if not self.skip:
                sys.stdout = FilteredStream(self.tmp_result)
                stdout_fileno = sys.stdout.stream.fileno()
                self.execute(server)
                sys.stdout.flush()
            self.is_executed_ok = True
        except TestExecutionError:
            self.is_executed_ok = False
        except Exception as e:
            if e.__class__.__name__ == 'TarantoolStartError':
                # worker should stop
                raise
            color_stdout('\nTest.run() received the following error:\n'
                         '{0}\n'.format(traceback.format_exc()),
                         schema='error')
            diagnostics = str(e)
        finally:
            if sys.stdout and sys.stdout != save_stdout:
                sys.stdout.close()
            sys.stdout = save_stdout
        self.is_executed = True
        sys.stdout.flush()

        is_tap = False
        if not self.skip:
            if not os.path.exists(self.tmp_result):
                self.is_executed_ok = False
                self.is_equal_result = False
            elif self.is_executed_ok and os.path.isfile(self.result):
                self.is_equal_result = filecmp.cmp(self.result,
                                                   self.tmp_result)
            elif self.is_executed_ok:
                if Options().args.is_verbose:
                    color_stdout('\n')
                    with open(self.tmp_result, 'r', encoding='utf-8',
                              errors='replace') as f:
                        color_stdout(f.read(), schema='log')
                is_tap, is_ok, is_skip = self.check_tap_output()
                self.is_equal_result = is_ok
                self.skip = is_skip
        else:
            self.is_equal_result = 1

        if self.args.valgrind:
            non_empty_logs = non_empty_valgrind_logs(
                server.current_valgrind_logs(for_test=True))
            self.is_valgrind_clean = not bool(non_empty_logs)

        short_status = None

        if self.skip:
            short_status = 'skip'
            color_stdout("[ skip ]\n", schema='test_skip')
            if os.path.exists(self.tmp_result):
                os.remove(self.tmp_result)
        elif (self.is_executed_ok and
              self.is_equal_result and
              self.is_valgrind_clean):
            short_status = 'pass'
            color_stdout("[ pass ]\n", schema='test_pass')
            if os.path.exists(self.tmp_result):
                os.remove(self.tmp_result)
        elif (self.is_executed_ok and
              not self.is_equal_result and
              not os.path.isfile(self.result) and
              not is_tap and
              Options().args.update_result):
            shutil.copy(self.tmp_result, self.result)
            short_status = 'new'
            color_stdout("[ new ]\n", schema='test_new')
        elif (self.is_executed_ok and
              not self.is_equal_result and
              os.path.isfile(self.result) and
              not is_tap and
              Options().args.update_result):
            shutil.copy(self.tmp_result, self.result)
            short_status = 'updated'
            color_stdout("[ updated ]\n", schema='test_new')
        else:
            has_result = os.path.exists(self.tmp_result)
            if has_result:
                safe_makedirs(self.var_suite_path)
                shutil.copy(self.tmp_result, self.reject)
            short_status = 'fail'
            color_stdout("[ fail ]\n", schema='test_fail')

            where = ""
            if not self.is_crash_reported and not has_result:
                color_stdout('\nCannot open %s\n' % self.tmp_result,
                             schema='error')
            elif not self.is_crash_reported and not self.is_executed_ok:
                self.print_diagnostics(self.reject,
                                       "Test failed! Output from reject file "
                                       "{0}:\n".format(self.reject))
                server.print_log(15)
                where = ": test execution aborted, reason " \
                        "'{0}'".format(diagnostics)
            elif not self.is_crash_reported and not self.is_equal_result:
                self.print_unidiff()
                server.print_log(15)
                where = ": wrong test output"
            elif not self.is_crash_reported and not self.is_valgrind_clean:
                os.remove(self.reject)
                for log_file in non_empty_logs:
                    self.print_diagnostics(log_file,
                                           "Test failed! Output from log file "
                                           "{0}:\n".format(log_file))
                where = ": there were warnings in the valgrind log file(s)"
        return short_status

    def print_diagnostics(self, log_file, message):
        """Print whole lines of client program output leading to test
        failure. Used to diagnose a failure of the client program"""

        color_stdout(message, schema='error')
        print_tail_n(log_file)

    def print_unidiff(self):
        """Print a unified diff between .test and .result files. Used
        to establish the cause of a failure when .test differs
        from .result."""

        color_stdout("\nTest failed! Result content mismatch:\n",
                     schema='error')
        utils_print_unidiff(self.result, self.reject)

    def tap_parse_print_yaml(self, yml):
        if 'expected' in yml and 'got' in yml:
            color_stdout('Expected: %s\n' % yml['expected'], schema='error')
            color_stdout('Got:      %s\n' % yml['got'], schema='error')
            del yml['expected']
            del yml['got']
        if 'trace' in yml:
            color_stdout('Traceback:\n', schema='error')
            for fr in yml['trace']:
                fname = fr.get('name', '')
                if fname:
                    fname = " function '%s'" % fname
                line = '[%-4s]%s at <%s:%d>\n' % (
                    fr['what'], fname, fr['filename'], fr['line']
                )
                color_stdout(line, schema='error')
            del yml['trace']
        if 'filename' in yml:
            del yml['filename']
        if 'line' in yml:
            del yml['line']
        yaml_str = pprint.pformat(yml)
        color_stdout('\n', schema='error')
        if len(yml):
            for line in yaml_str.splitlines():
                color_stdout(line + '\n', schema='error')
            color_stdout('\n', schema='error')

    def check_tap_output(self):
        """ Returns is_tap, is_ok, is_skip """
        try:
            with open(self.tmp_result, 'r', encoding='utf-8', errors='replace') as f:
                content = f.read()
            tap = pytap13.TAP13()
            tap.parse(content)
        except (ValueError, UnicodeDecodeError) as e:
            color_stdout('\nTAP13 parse failed (%s).\n' % str(e),
                         schema='error')
            color_stdout('\nNo result file (%s) found.\n' % self.result,
                         schema='error')
            if not Options().args.update_result:
                msg = 'Run the test with --update-result option to write the new result file.\n'
                color_stdout(msg, schema='error')
            self.is_crash_reported = True
            return False, False, False

        is_ok = True
        is_skip = False
        num_skipped_tests = 0
        for test_case in tap.tests:
            if test_case.directive == "SKIP":
                num_skipped_tests += 1
            if test_case.result == 'ok':
                continue
            if is_ok:
                color_stdout('\n')
            color_stdout('%s %s %s # %s %s\n' % (
                test_case.result,
                test_case.id or '',
                test_case.description or '-',
                test_case.directive or '',
                test_case.comment or ''), schema='error')
            if test_case.yaml:
                self.tap_parse_print_yaml(test_case.yaml)
            is_ok = False
        if not is_ok:
            color_stdout('Rejected result file: %s\n' % self.reject,
                         schema='test_var')
            self.is_crash_reported = True
        if num_skipped_tests == len(tap.tests):
            is_skip = True

        return True, is_ok, is_skip

1	import filecmp	1✔
2	import gevent	1✔
3	import os	1✔
4	import pprint	1✔
5	import re	1✔
6	import shutil	1✔
7	import sys	1✔
8	import traceback	1✔
9	from functools import partial	1✔
10
11	from lib import Options	1✔
12	from lib.colorer import color_stdout	1✔
13	from lib.utils import assert_bytes	1✔
14	from lib.utils import non_empty_valgrind_logs	1✔
15	from lib.utils import print_tail_n	1✔
16	from lib.utils import print_unidiff as utils_print_unidiff	1✔
17	from lib.utils import safe_makedirs	1✔
18	from lib.utils import str_to_bytes	1✔
19	from lib import pytap13	1✔
20
21
22	class TestExecutionError(OSError):	1✔
23	"""To be raised when a test execution fails"""
24	pass	1✔
25
26
27	class TestRunGreenlet(gevent.Greenlet):	1✔
28	def __init__(self, green_callable, args, *kwargs):	1✔
29	self.callable = green_callable	1✔
30	self.callable_args = args	1✔
31	self.callable_kwargs = kwargs	1✔
32	super(TestRunGreenlet, self).__init__()	1✔
33
34	def _run(self, args, *kwargs):	1✔
35	self.callable(self.callable_args, *self.callable_kwargs)	1✔
36
37	def __repr__(self):	1✔
38	return "<TestRunGreenlet at {0} info='{1}'>".format(	×
39	hex(id(self)), getattr(self, "info", None))
40
41
42	class FilteredStream:	1✔
43	"""Helper class to filter .result file output"""
44	def __init__(self, filename):	1✔
45	self.stream = open(filename, "wb+")	1✔
46	self.filters = []	1✔
47	self.inspector = None	1✔
48
49	def write_bytes(self, fragment):	1✔
50	""" The same as ``write()``, but accepts ``<bytes>`` as
51	input.
52	"""
53	assert_bytes(fragment)	1✔
54	skipped = False	1✔
55	for line in fragment.splitlines(True):	1✔
56	original_len = len(line.strip())	1✔
57	for pattern, replacement in self.filters:	1✔
58	line = re.sub(pattern, replacement, line)	1✔
59	# don't write lines that are completely filtered out:
60	skipped = original_len and not line.strip()	1✔
61	if skipped:	1!
62	break	×
63	if not skipped:	1!
64	self.stream.write(line)	1✔
65
66	def write(self, fragment):	1✔
67	""" Apply all filters, then write result to the underlying
68	stream.
69
70	Do line-oriented filtering: the fragment doesn't have
71	to represent just one line.
72
73	Accepts ``<str>`` as input, just like the standard
74	``sys.stdout.write()``.
75	"""
76	self.write_bytes(str_to_bytes(fragment))	1✔
77
78	def push_filter(self, pattern, replacement):	1✔
79	self.filters.append([str_to_bytes(pattern), str_to_bytes(replacement)])	1✔
80
81	def pop_filter(self):	1✔
82	self.filters.pop()	×
83
84	def clear_all_filters(self):	1✔
85	self.filters = []	1✔
86
87	def close(self):	1✔
88	self.clear_all_filters()	1✔
89	self.stream.close()	1✔
90
91	def flush(self):	1✔
92	self.stream.flush()	1✔
93
94	def fileno(self):	1✔
95	""" May be used for direct writting. Discards any filters.
96	"""
97	return self.stream.fileno()	1✔
98
99
100	def get_filename_by_test(postfix, test_name):	1✔
101	"""For <..>/<name>_test.* or <..>/<name>.test.* return <name> + postfix
102
103	Examples:
104	postfix='.result', test_name='foo/bar.test.lua' => return 'bar.result'
105	postfix='.reject', test_name='bar_test.lua' => return 'bar.reject'
106	"""
107	rg = re.compile(r'[._]test.*')	1✔
108	return os.path.basename(rg.sub(postfix, test_name))	1✔
109
110
111	get_reject = partial(get_filename_by_test, '.reject')	1✔
112	get_result = partial(get_filename_by_test, '.result')	1✔
113	get_skipcond = partial(get_filename_by_test, '.skipcond')	1✔
114
115
116	class Test(object):	1✔
117	"""An individual test file. A test object can run itself
118	and remembers completion state of the run.
119
120	If file <test_name>.skipcond is exists it will be executed before
121	test and if it sets self.skip to True value the test will be skipped.
122	"""
123
124	def __init__(self, name, args, suite_ini, params={}, conf_name=None):	1✔
125	"""Initialize test properties: path to test file, path to
126	temporary result file, path to the client program, test status."""
127	self.name = name	1✔
128	self.args = args	1✔
129	self.suite_ini = suite_ini	1✔
130	self.result = os.path.join(suite_ini['suite'], get_result(name))	1✔
131	self.skip_cond = os.path.join(suite_ini['suite'], get_skipcond(name))	1✔
132	self.tmp_result = os.path.join(suite_ini['vardir'], get_result(name))	1✔
133	self.var_suite_path = os.path.join(suite_ini['vardir'], 'rejects',	1✔
134	suite_ini['suite'])
135	self.reject = os.path.join(self.var_suite_path, get_reject(name))	1✔
136	self.is_executed = False	1✔
137	self.is_executed_ok = None	1✔
138	self.is_equal_result = None	1✔
139	self.is_valgrind_clean = True	1✔
140	self.is_terminated = False	1✔
141	self.run_params = params	1✔
142	self.conf_name = conf_name	1✔
143
144	# filled in execute() when a greenlet runs
145	self.current_test_greenlet = None	1✔
146
147	# prevent double/triple reporting
148	self.is_crash_reported = False	1✔
149
150	@property	1✔
151	def id(self):	1✔
152	return self.name, self.conf_name	1✔
153
154	def passed(self):	1✔
155	"""Return true if this test was run successfully."""
156	return (self.is_executed and	×
157	self.is_executed_ok and
158	self.is_equal_result)
159
160	def execute(self, server):	1✔
161	# Note: don't forget to set 'server.current_test = self' in
162	# inherited classes. Crash reporting relying on that.
163	server.current_test = self	1✔
164	# All the test runs must be isolated between each other on each worker.
165	server.pretest_clean()	1✔
166
167	def run(self, server):	1✔
168	""" Execute the test assuming it's a python program. If the test
169	aborts, print its output to stdout, and raise an exception. Else,
170	comprare result and reject files. If there is a difference, print
171	it to stdout.
172
173	Returns short status of the test as a string: 'skip', 'pass',
174	'new', 'updated' or 'fail'.
175	There is also one possible value for short_status, 'disabled',
176	but it returned in the caller, TestSuite.run_test().
177	"""
178
179	# Note: test was created before certain worker become known, so we need
180	# to update temporary result directory here as it depends on 'vardir'.
181	self.tmp_result = os.path.join(self.suite_ini['vardir'],	1✔
182	os.path.basename(self.result))
183
184	diagnostics = "unknown"	1✔
185	save_stdout = sys.stdout	1✔
186	try:	1✔
187	self.skip = False	1✔
188	if os.path.exists(self.skip_cond):	1!
189	sys.stdout = FilteredStream(self.tmp_result)	×
190	stdout_fileno = sys.stdout.stream.fileno()	×
191	new_globals = dict(locals(), **server.__dict__)	×
192	with open(self.skip_cond, 'r') as f:	×
193	code = compile(f.read(), self.skip_cond, 'exec')	×
194	exec(code, new_globals)	×
195	sys.stdout.close()	×
196	sys.stdout = save_stdout	×
197	if not self.skip:	1!
198	sys.stdout = FilteredStream(self.tmp_result)	1✔
199	stdout_fileno = sys.stdout.stream.fileno()	1✔
200	self.execute(server)	1✔
201	sys.stdout.flush()	1✔
202	self.is_executed_ok = True	1✔
203	except TestExecutionError:	×
204	self.is_executed_ok = False	×
205	except Exception as e:	×
206	if e.__class__.__name__ == 'TarantoolStartError':	×
207	# worker should stop
208	raise	×
209	color_stdout('\nTest.run() received the following error:\n'	×
210	'{0}\n'.format(traceback.format_exc()),
211	schema='error')
212	diagnostics = str(e)	×
213	finally:
214	if sys.stdout and sys.stdout != save_stdout:	1!
215	sys.stdout.close()	1✔
216	sys.stdout = save_stdout	1!
217	self.is_executed = True	1✔
218	sys.stdout.flush()	1✔
219
220	is_tap = False	1✔
221	if not self.skip:	1!
222	if not os.path.exists(self.tmp_result):	1!
223	self.is_executed_ok = False	×
224	self.is_equal_result = False	×
225	elif self.is_executed_ok and os.path.isfile(self.result):	1✔
226	self.is_equal_result = filecmp.cmp(self.result,	1✔
227	self.tmp_result)
228	elif self.is_executed_ok:	1!
229	if Options().args.is_verbose:	1!
230	color_stdout('\n')	×
NEW 231	with open(self.tmp_result, 'r', encoding='utf-8',	×
232	errors='replace') as f:
UNCOV 233	color_stdout(f.read(), schema='log')	×
234	is_tap, is_ok, is_skip = self.check_tap_output()	1✔
235	self.is_equal_result = is_ok	1✔
236	self.skip = is_skip	1✔
237	else:
238	self.is_equal_result = 1	×
239
240	if self.args.valgrind:	1!
241	non_empty_logs = non_empty_valgrind_logs(	×
242	server.current_valgrind_logs(for_test=True))
243	self.is_valgrind_clean = not bool(non_empty_logs)	×
244
245	short_status = None	1✔
246
247	if self.skip:	1!
248	short_status = 'skip'	×
249	color_stdout("[ skip ]\n", schema='test_skip')	×
250	if os.path.exists(self.tmp_result):	×
251	os.remove(self.tmp_result)	×
252	elif (self.is_executed_ok and	1!
253	self.is_equal_result and
254	self.is_valgrind_clean):
255	short_status = 'pass'	1✔
256	color_stdout("[ pass ]\n", schema='test_pass')	1✔
257	if os.path.exists(self.tmp_result):	1!
258	os.remove(self.tmp_result)	1✔
259	elif (self.is_executed_ok and	×
260	not self.is_equal_result and
261	not os.path.isfile(self.result) and
262	not is_tap and
263	Options().args.update_result):
264	shutil.copy(self.tmp_result, self.result)	×
265	short_status = 'new'	×
266	color_stdout("[ new ]\n", schema='test_new')	×
267	elif (self.is_executed_ok and	×
268	not self.is_equal_result and
269	os.path.isfile(self.result) and
270	not is_tap and
271	Options().args.update_result):
272	shutil.copy(self.tmp_result, self.result)	×
273	short_status = 'updated'	×
274	color_stdout("[ updated ]\n", schema='test_new')	×
275	else:
276	has_result = os.path.exists(self.tmp_result)	×
277	if has_result:	×
278	safe_makedirs(self.var_suite_path)	×
279	shutil.copy(self.tmp_result, self.reject)	×
280	short_status = 'fail'	×
281	color_stdout("[ fail ]\n", schema='test_fail')	×
282
283	where = ""	×
284	if not self.is_crash_reported and not has_result:	×
285	color_stdout('\nCannot open %s\n' % self.tmp_result,	×
286	schema='error')
287	elif not self.is_crash_reported and not self.is_executed_ok:	×
288	self.print_diagnostics(self.reject,	×
289	"Test failed! Output from reject file "
290	"{0}:\n".format(self.reject))
291	server.print_log(15)	×
292	where = ": test execution aborted, reason " \	×
293	"'{0}'".format(diagnostics)
294	elif not self.is_crash_reported and not self.is_equal_result:	×
295	self.print_unidiff()	×
296	server.print_log(15)	×
297	where = ": wrong test output"	×
298	elif not self.is_crash_reported and not self.is_valgrind_clean:	×
299	os.remove(self.reject)	×
300	for log_file in non_empty_logs:	×
301	self.print_diagnostics(log_file,	×
302	"Test failed! Output from log file "
303	"{0}:\n".format(log_file))
304	where = ": there were warnings in the valgrind log file(s)"	×
305	return short_status	1✔
306
307	def print_diagnostics(self, log_file, message):	1✔
308	"""Print whole lines of client program output leading to test
309	failure. Used to diagnose a failure of the client program"""
310
311	color_stdout(message, schema='error')	×
312	print_tail_n(log_file)	×
313
314	def print_unidiff(self):	1✔
315	"""Print a unified diff between .test and .result files. Used
316	to establish the cause of a failure when .test differs
317	from .result."""
318
319	color_stdout("\nTest failed! Result content mismatch:\n",	×
320	schema='error')
321	utils_print_unidiff(self.result, self.reject)	×
322
323	def tap_parse_print_yaml(self, yml):	1✔
324	if 'expected' in yml and 'got' in yml:	×
325	color_stdout('Expected: %s\n' % yml['expected'], schema='error')	×
326	color_stdout('Got: %s\n' % yml['got'], schema='error')	×
327	del yml['expected']	×
328	del yml['got']	×
329	if 'trace' in yml:	×
330	color_stdout('Traceback:\n', schema='error')	×
331	for fr in yml['trace']:	×
332	fname = fr.get('name', '')	×
333	if fname:	×
334	fname = " function '%s'" % fname	×
335	line = '[%-4s]%s at <%s:%d>\n' % (	×
336	fr['what'], fname, fr['filename'], fr['line']
337	)
338	color_stdout(line, schema='error')	×
339	del yml['trace']	×
340	if 'filename' in yml:	×
341	del yml['filename']	×
342	if 'line' in yml:	×
343	del yml['line']	×
344	yaml_str = pprint.pformat(yml)	×
345	color_stdout('\n', schema='error')	×
346	if len(yml):	×
347	for line in yaml_str.splitlines():	×
348	color_stdout(line + '\n', schema='error')	×
349	color_stdout('\n', schema='error')	×
350
351	def check_tap_output(self):	1✔
352	""" Returns is_tap, is_ok, is_skip """
353	try:	1✔
354	with open(self.tmp_result, 'r', encoding='utf-8', errors='replace') as f:	1✔
355	content = f.read()	1✔
356	tap = pytap13.TAP13()	1✔
357	tap.parse(content)	1✔
358	except (ValueError, UnicodeDecodeError) as e:	×
359	color_stdout('\nTAP13 parse failed (%s).\n' % str(e),	×
360	schema='error')
361	color_stdout('\nNo result file (%s) found.\n' % self.result,	×
362	schema='error')
363	if not Options().args.update_result:	×
364	msg = 'Run the test with --update-result option to write the new result file.\n'	×
365	color_stdout(msg, schema='error')	×
366	self.is_crash_reported = True	×
367	return False, False, False	×
368
369	is_ok = True	1✔
370	is_skip = False	1✔
371	num_skipped_tests = 0	1✔
372	for test_case in tap.tests:	1✔
373	if test_case.directive == "SKIP":	1!
374	num_skipped_tests += 1	×
375	if test_case.result == 'ok':	1!
376	continue	1✔
377	if is_ok:	×
378	color_stdout('\n')	×
379	color_stdout('%s %s %s # %s %s\n' % (	×
380	test_case.result,
381	test_case.id or '',
382	test_case.description or '-',
383	test_case.directive or '',
384	test_case.comment or ''), schema='error')
385	if test_case.yaml:	×
386	self.tap_parse_print_yaml(test_case.yaml)	×
387	is_ok = False	×
388	if not is_ok:	1!
389	color_stdout('Rejected result file: %s\n' % self.reject,	×
390	schema='test_var')
391	self.is_crash_reported = True	×
392	if num_skipped_tests == len(tap.tests):	1!
393	is_skip = True	×
394
395	return True, is_ok, is_skip	1✔

tarantool / test-run / 11499903942

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