1#
2# Licensed to the Apache Software Foundation (ASF) under one
3# or more contributor license agreements. See the NOTICE file
4# distributed with this work for additional information
5# regarding copyright ownership. The ASF licenses this file
6# to you under the Apache License, Version 2.0 (the
7# "License"); you may not use this file except in compliance
8# with the License. You may obtain a copy of the License at
9#
10#   http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing,
13# software distributed under the License is distributed on an
14# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15# KIND, either express or implied. See the License for the
16# specific language governing permissions and limitations
17# under the License.
18#
19
20from __future__ import print_function
21import datetime
22import json
23import multiprocessing
24import os
25import platform
26import re
27import subprocess
28import sys
29import time
30import traceback
31
32from .compat import logfile_open, path_join, str_join
33from .test import TestEntry
34
35LOG_DIR = 'log'
36RESULT_HTML = 'index.html'
37RESULT_JSON = 'results.json'
38FAIL_JSON = 'known_failures_%s.json'
39
40
41def generate_known_failures(testdir, overwrite, save, out):
42    def collect_failures(results):
43        success_index = 5
44        for r in results:
45            if not r[success_index]:
46                yield TestEntry.get_name(*r)
47    try:
48        with logfile_open(path_join(testdir, RESULT_JSON), 'r') as fp:
49            results = json.load(fp)
50    except IOError:
51        sys.stderr.write('Unable to load last result. Did you run tests ?\n')
52        return False
53    fails = collect_failures(results['results'])
54    if not overwrite:
55        known = load_known_failures(testdir)
56        known.extend(fails)
57        fails = known
58    fails_json = json.dumps(sorted(set(fails)), indent=2, separators=(',', ': '))
59    if save:
60        with logfile_open(os.path.join(testdir, FAIL_JSON % platform.system()), 'w+') as fp:
61            fp.write(fails_json)
62        sys.stdout.write('Successfully updated known failures.\n')
63    if out:
64        sys.stdout.write(fails_json)
65        sys.stdout.write('\n')
66    return True
67
68
69def load_known_failures(testdir):
70    try:
71        with logfile_open(path_join(testdir, FAIL_JSON % platform.system()), 'r') as fp:
72            return json.load(fp)
73    except IOError:
74        return []
75
76
77class TestReporter(object):
78    # Unfortunately, standard library doesn't handle timezone well
79    # DATETIME_FORMAT = '%a %b %d %H:%M:%S %Z %Y'
80    DATETIME_FORMAT = '%a %b %d %H:%M:%S %Y'
81
82    def __init__(self):
83        self._log = multiprocessing.get_logger()
84        self._lock = multiprocessing.Lock()
85
86    @classmethod
87    def test_logfile(cls, test_name, prog_kind, dir=None):
88        relpath = path_join('log', '%s_%s.log' % (test_name, prog_kind))
89        return relpath if not dir else os.path.realpath(path_join(dir, relpath))
90
91    def _start(self):
92        self._start_time = time.time()
93
94    @property
95    def _elapsed(self):
96        return time.time() - self._start_time
97
98    @classmethod
99    def _format_date(cls):
100        return '%s' % datetime.datetime.now().strftime(cls.DATETIME_FORMAT)
101
102    def _print_date(self):
103        print(self._format_date(), file=self.out)
104
105    def _print_bar(self, out=None):
106        print(
107            '===============================================================================',
108            file=(out or self.out))
109
110    def _print_exec_time(self):
111        print('Test execution took {:.1f} seconds.'.format(self._elapsed), file=self.out)
112
113
114class ExecReporter(TestReporter):
115    def __init__(self, testdir, test, prog):
116        super(ExecReporter, self).__init__()
117        self._test = test
118        self._prog = prog
119        self.logpath = self.test_logfile(test.name, prog.kind, testdir)
120        self.out = None
121
122    def begin(self):
123        self._start()
124        self._open()
125        if self.out and not self.out.closed:
126            self._print_header()
127        else:
128            self._log.debug('Output stream is not available.')
129
130    def end(self, returncode):
131        self._lock.acquire()
132        try:
133            if self.out and not self.out.closed:
134                self._print_footer(returncode)
135                self._close()
136                self.out = None
137            else:
138                self._log.debug('Output stream is not available.')
139        finally:
140            self._lock.release()
141
142    def killed(self):
143        print(file=self.out)
144        print('Server process is successfully killed.', file=self.out)
145        self.end(None)
146
147    def died(self):
148        print(file=self.out)
149        print('*** Server process has died unexpectedly ***', file=self.out)
150        self.end(None)
151
152    _init_failure_exprs = {
153        'server': list(map(re.compile, [
154            '[Aa]ddress already in use',
155            'Could not bind',
156            'EADDRINUSE',
157        ])),
158        'client': list(map(re.compile, [
159            '[Cc]onnection refused',
160            'Could not connect to',
161            'Could not open UNIX ',       # domain socket (rb)
162            'ECONNREFUSED',
163            'econnrefused',               # erl
164            'CONNECTION-REFUSED-ERROR',   # cl
165            'connect ENOENT',             # nodejs domain socket
166            'No such file or directory',  # domain socket
167            'Sockets.TcpClient.Connect',  # csharp
168        ])),
169    }
170
171    def maybe_false_positive(self):
172        """Searches through log file for socket bind error.
173        Returns True if suspicious expression is found, otherwise False"""
174        try:
175            if self.out and not self.out.closed:
176                self.out.flush()
177            exprs = self._init_failure_exprs[self._prog.kind]
178
179            def match(line):
180                for expr in exprs:
181                    if expr.search(line):
182                        self._log.info("maybe false positive: %s" % line)
183                        return True
184
185            with logfile_open(self.logpath, 'r') as fp:
186                if any(map(match, fp)):
187                    return True
188        except (KeyboardInterrupt, SystemExit):
189            raise
190        except Exception as ex:
191            self._log.warn('[%s]: Error while detecting false positive: %s' % (self._test.name, str(ex)))
192            self._log.info(traceback.print_exc())
193        return False
194
195    def _open(self):
196        self.out = logfile_open(self.logpath, 'w+')
197
198    def _close(self):
199        self.out.close()
200
201    def _print_header(self):
202        self._print_date()
203        print('Executing: %s' % str_join(' ', self._prog.command), file=self.out)
204        print('Directory: %s' % self._prog.workdir, file=self.out)
205        print('config:delay: %s' % self._test.delay, file=self.out)
206        print('config:timeout: %s' % self._test.timeout, file=self.out)
207        self._print_bar()
208        self.out.flush()
209
210    def _print_footer(self, returncode=None):
211        self._print_bar()
212        if returncode is not None:
213            print('Return code: %d (negative values indicate kill by signal)' % returncode, file=self.out)
214        else:
215            print('Process is killed.', file=self.out)
216        self._print_exec_time()
217        self._print_date()
218
219
220class SummaryReporter(TestReporter):
221    def __init__(self, basedir, testdir_relative, concurrent=True):
222        super(SummaryReporter, self).__init__()
223        self._basedir = basedir
224        self._testdir_rel = testdir_relative
225        self.logdir = path_join(self.testdir, LOG_DIR)
226        self.out_path = path_join(self.testdir, RESULT_JSON)
227        self.concurrent = concurrent
228        self.out = sys.stdout
229        self._platform = platform.system()
230        self._revision = self._get_revision()
231        self._tests = []
232        if not os.path.exists(self.logdir):
233            os.mkdir(self.logdir)
234        self._known_failures = load_known_failures(self.testdir)
235        self._unexpected_success = []
236        self._flaky_success = []
237        self._unexpected_failure = []
238        self._expected_failure = []
239        self._print_header()
240
241    @property
242    def testdir(self):
243        return path_join(self._basedir, self._testdir_rel)
244
245    def _result_string(self, test):
246        if test.success:
247            if test.retry_count == 0:
248                return 'success'
249            elif test.retry_count == 1:
250                return 'flaky(1 retry)'
251            else:
252                return 'flaky(%d retries)' % test.retry_count
253        elif test.expired:
254            return 'failure(timeout)'
255        else:
256            return 'failure(%d)' % test.returncode
257
258    def _get_revision(self):
259        p = subprocess.Popen(['git', 'rev-parse', '--short', 'HEAD'],
260                             cwd=self.testdir, stdout=subprocess.PIPE)
261        out, _ = p.communicate()
262        return out.strip()
263
264    def _format_test(self, test, with_result=True):
265        name = '%s-%s' % (test.server.name, test.client.name)
266        trans = '%s-%s' % (test.transport, test.socket)
267        if not with_result:
268            return '{:24s}{:18s}{:25s}'.format(name[:23], test.protocol[:17], trans[:24])
269        else:
270            return '{:24s}{:18s}{:25s}{:s}\n'.format(name[:23], test.protocol[:17],
271                                                     trans[:24], self._result_string(test))
272
273    def _print_test_header(self):
274        self._print_bar()
275        print(
276            '{:24s}{:18s}{:25s}{:s}'.format('server-client:', 'protocol:', 'transport:', 'result:'),
277            file=self.out)
278
279    def _print_header(self):
280        self._start()
281        print('Apache Thrift - Integration Test Suite', file=self.out)
282        self._print_date()
283        self._print_test_header()
284
285    def _print_unexpected_failure(self):
286        if len(self._unexpected_failure) > 0:
287            self.out.writelines([
288                '*** Following %d failures were unexpected ***:\n' % len(self._unexpected_failure),
289                'If it is introduced by you, please fix it before submitting the code.\n',
290                # 'If not, please report at https://issues.apache.org/jira/browse/THRIFT\n',
291            ])
292            self._print_test_header()
293            for i in self._unexpected_failure:
294                self.out.write(self._format_test(self._tests[i]))
295            self._print_bar()
296        else:
297            print('No unexpected failures.', file=self.out)
298
299    def _print_flaky_success(self):
300        if len(self._flaky_success) > 0:
301            print(
302                'Following %d tests were expected to cleanly succeed but needed retry:' % len(self._flaky_success),
303                file=self.out)
304            self._print_test_header()
305            for i in self._flaky_success:
306                self.out.write(self._format_test(self._tests[i]))
307            self._print_bar()
308
309    def _print_unexpected_success(self):
310        if len(self._unexpected_success) > 0:
311            print(
312                'Following %d tests were known to fail but succeeded (maybe flaky):' % len(self._unexpected_success),
313                file=self.out)
314            self._print_test_header()
315            for i in self._unexpected_success:
316                self.out.write(self._format_test(self._tests[i]))
317            self._print_bar()
318
319    def _http_server_command(self, port):
320        if sys.version_info[0] < 3:
321            return 'python -m SimpleHTTPServer %d' % port
322        else:
323            return 'python -m http.server %d' % port
324
325    def _print_footer(self):
326        fail_count = len(self._expected_failure) + len(self._unexpected_failure)
327        self._print_bar()
328        self._print_unexpected_success()
329        self._print_flaky_success()
330        self._print_unexpected_failure()
331        self._write_html_data()
332        self._assemble_log('unexpected failures', self._unexpected_failure)
333        self._assemble_log('known failures', self._expected_failure)
334        self.out.writelines([
335            'You can browse results at:\n',
336            '\tfile://%s/%s\n' % (self.testdir, RESULT_HTML),
337            '# If you use Chrome, run:\n',
338            '# \tcd %s\n#\t%s\n' % (self._basedir, self._http_server_command(8001)),
339            '# then browse:\n',
340            '# \thttp://localhost:%d/%s/\n' % (8001, self._testdir_rel),
341            'Full log for each test is here:\n',
342            '\ttest/log/server_client_protocol_transport_client.log\n',
343            '\ttest/log/server_client_protocol_transport_server.log\n',
344            '%d failed of %d tests in total.\n' % (fail_count, len(self._tests)),
345        ])
346        self._print_exec_time()
347        self._print_date()
348
349    def _render_result(self, test):
350        return [
351            test.server.name,
352            test.client.name,
353            test.protocol,
354            test.transport,
355            test.socket,
356            test.success,
357            test.as_expected,
358            test.returncode,
359            {
360                'server': self.test_logfile(test.name, test.server.kind),
361                'client': self.test_logfile(test.name, test.client.kind),
362            },
363        ]
364
365    def _write_html_data(self):
366        """Writes JSON data to be read by result html"""
367        results = [self._render_result(r) for r in self._tests]
368        with logfile_open(self.out_path, 'w+') as fp:
369            fp.write(json.dumps({
370                'date': self._format_date(),
371                'revision': str(self._revision),
372                'platform': self._platform,
373                'duration': '{:.1f}'.format(self._elapsed),
374                'results': results,
375            }, indent=2))
376
377    def _assemble_log(self, title, indexes):
378        if len(indexes) > 0:
379            def add_prog_log(fp, test, prog_kind):
380                print('*************************** %s message ***************************' % prog_kind,
381                      file=fp)
382                path = self.test_logfile(test.name, prog_kind, self.testdir)
383                if os.path.exists(path):
384                    with logfile_open(path, 'r') as prog_fp:
385                        print(prog_fp.read(), file=fp)
386            filename = title.replace(' ', '_') + '.log'
387            with logfile_open(os.path.join(self.logdir, filename), 'w+') as fp:
388                for test in map(self._tests.__getitem__, indexes):
389                    fp.write('TEST: [%s]\n' % test.name)
390                    add_prog_log(fp, test, test.server.kind)
391                    add_prog_log(fp, test, test.client.kind)
392                    fp.write('**********************************************************************\n\n')
393            print('%s are logged to %s/%s/%s' % (title.capitalize(), self._testdir_rel, LOG_DIR, filename))
394
395    def end(self):
396        self._print_footer()
397        return len(self._unexpected_failure) == 0
398
399    def add_test(self, test_dict):
400        test = TestEntry(self.testdir, **test_dict)
401        self._lock.acquire()
402        try:
403            if not self.concurrent:
404                self.out.write(self._format_test(test, False))
405                self.out.flush()
406            self._tests.append(test)
407            return len(self._tests) - 1
408        finally:
409            self._lock.release()
410
411    def add_result(self, index, returncode, expired, retry_count):
412        self._lock.acquire()
413        try:
414            failed = returncode is None or returncode != 0
415            flaky = not failed and retry_count != 0
416            test = self._tests[index]
417            known = test.name in self._known_failures
418            if failed:
419                if known:
420                    self._log.debug('%s failed as expected' % test.name)
421                    self._expected_failure.append(index)
422                else:
423                    self._log.info('unexpected failure: %s' % test.name)
424                    self._unexpected_failure.append(index)
425            elif flaky and not known:
426                self._log.info('unexpected flaky success: %s' % test.name)
427                self._flaky_success.append(index)
428            elif not flaky and known:
429                self._log.info('unexpected success: %s' % test.name)
430                self._unexpected_success.append(index)
431            test.success = not failed
432            test.returncode = returncode
433            test.retry_count = retry_count
434            test.expired = expired
435            test.as_expected = known == failed
436            if not self.concurrent:
437                self.out.write(self._result_string(test) + '\n')
438            else:
439                self.out.write(self._format_test(test))
440        finally:
441            self._lock.release()
442