run_tests.py revision c701e571
1#!/usr/bin/env python3
2
3import sys
4import shutil
5import os
6import fnmatch
7import unittest
8import argparse
9import time
10import threading
11import signal
12import psutil
13import re
14import multiprocessing
15from multiprocessing import Process, Pipe, cpu_count
16from multiprocessing.queues import Queue
17from multiprocessing.managers import BaseManager
18import framework
19from framework import VppTestRunner, running_extended_tests, VppTestCase, \
20    get_testcase_doc_name, get_test_description, PASS, FAIL, ERROR, SKIP, \
21    TEST_RUN
22from debug import spawn_gdb
23from log import get_parallel_logger, double_line_delim, RED, YELLOW, GREEN, \
24    colorize, single_line_delim
25from discover_tests import discover_tests
26from subprocess import check_output, CalledProcessError
27from util import check_core_path, get_core_path, is_core_present
28
29# timeout which controls how long the child has to finish after seeing
30# a core dump in test temporary directory. If this is exceeded, parent assumes
31# that child process is stuck (e.g. waiting for shm mutex, which will never
32# get unlocked) and kill the child
33core_timeout = 3
34min_req_shm = 536870912  # min 512MB shm required
35# 128MB per extra process
36shm_per_process = 134217728
37
38
39class StreamQueue(Queue):
40    def write(self, msg):
41        self.put(msg)
42
43    def flush(self):
44        sys.__stdout__.flush()
45        sys.__stderr__.flush()
46
47    def fileno(self):
48        return self._writer.fileno()
49
50
51class StreamQueueManager(BaseManager):
52    pass
53
54
55StreamQueueManager.register('StreamQueue', StreamQueue)
56
57
58class TestResult(dict):
59    def __init__(self, testcase_suite, testcases_by_id=None):
60        super(TestResult, self).__init__()
61        self[PASS] = []
62        self[FAIL] = []
63        self[ERROR] = []
64        self[SKIP] = []
65        self[TEST_RUN] = []
66        self.crashed = False
67        self.testcase_suite = testcase_suite
68        self.testcases = [testcase for testcase in testcase_suite]
69        self.testcases_by_id = testcases_by_id
70
71    def was_successful(self):
72        return 0 == len(self[FAIL]) == len(self[ERROR]) \
73            and len(self[PASS] + self[SKIP]) \
74            == self.testcase_suite.countTestCases() == len(self[TEST_RUN])
75
76    def no_tests_run(self):
77        return 0 == len(self[TEST_RUN])
78
79    def process_result(self, test_id, result):
80        self[result].append(test_id)
81
82    def suite_from_failed(self):
83        rerun_ids = set([])
84        for testcase in self.testcase_suite:
85            tc_id = testcase.id()
86            if tc_id not in self[PASS] and tc_id not in self[SKIP]:
87                rerun_ids.add(tc_id)
88        if rerun_ids:
89            return suite_from_failed(self.testcase_suite, rerun_ids)
90
91    def get_testcase_names(self, test_id):
92        # could be tearDownClass (test_ipsec_esp.TestIpsecEsp1)
93        setup_teardown_match = re.match(
94            r'((tearDownClass)|(setUpClass)) \((.+\..+)\)', test_id)
95        if setup_teardown_match:
96            test_name, _, _, testcase_name = setup_teardown_match.groups()
97            if len(testcase_name.split('.')) == 2:
98                for key in self.testcases_by_id.keys():
99                    if key.startswith(testcase_name):
100                        testcase_name = key
101                        break
102            testcase_name = self._get_testcase_doc_name(testcase_name)
103        else:
104            test_name = self._get_test_description(test_id)
105            testcase_name = self._get_testcase_doc_name(test_id)
106
107        return testcase_name, test_name
108
109    def _get_test_description(self, test_id):
110        if test_id in self.testcases_by_id:
111            desc = get_test_description(descriptions,
112                                        self.testcases_by_id[test_id])
113        else:
114            desc = test_id
115        return desc
116
117    def _get_testcase_doc_name(self, test_id):
118        if test_id in self.testcases_by_id:
119            doc_name = get_testcase_doc_name(self.testcases_by_id[test_id])
120        else:
121            doc_name = test_id
122        return doc_name
123
124
125def test_runner_wrapper(suite, keep_alive_pipe, stdouterr_queue,
126                        finished_pipe, result_pipe, logger):
127    sys.stdout = stdouterr_queue
128    sys.stderr = stdouterr_queue
129    VppTestCase.parallel_handler = logger.handlers[0]
130    result = VppTestRunner(keep_alive_pipe=keep_alive_pipe,
131                           descriptions=descriptions,
132                           verbosity=verbose,
133                           result_pipe=result_pipe,
134                           failfast=failfast,
135                           print_summary=False).run(suite)
136    finished_pipe.send(result.wasSuccessful())
137    finished_pipe.close()
138    keep_alive_pipe.close()
139
140
141class TestCaseWrapper(object):
142    def __init__(self, testcase_suite, manager):
143        self.keep_alive_parent_end, self.keep_alive_child_end = Pipe(
144            duplex=False)
145        self.finished_parent_end, self.finished_child_end = Pipe(duplex=False)
146        self.result_parent_end, self.result_child_end = Pipe(duplex=False)
147        self.testcase_suite = testcase_suite
148        if sys.version[0] == '2':
149            self.stdouterr_queue = manager.StreamQueue()
150        else:
151            from multiprocessing import get_context
152            self.stdouterr_queue = manager.StreamQueue(ctx=get_context())
153        self.logger = get_parallel_logger(self.stdouterr_queue)
154        self.child = Process(target=test_runner_wrapper,
155                             args=(testcase_suite,
156                                   self.keep_alive_child_end,
157                                   self.stdouterr_queue,
158                                   self.finished_child_end,
159                                   self.result_child_end,
160                                   self.logger)
161                             )
162        self.child.start()
163        self.last_test_temp_dir = None
164        self.last_test_vpp_binary = None
165        self._last_test = None
166        self.last_test_id = None
167        self.vpp_pid = None
168        self.last_heard = time.time()
169        self.core_detected_at = None
170        self.testcases_by_id = {}
171        self.testclasess_with_core = {}
172        for testcase in self.testcase_suite:
173            self.testcases_by_id[testcase.id()] = testcase
174        self.result = TestResult(testcase_suite, self.testcases_by_id)
175
176    @property
177    def last_test(self):
178        return self._last_test
179
180    @last_test.setter
181    def last_test(self, test_id):
182        self.last_test_id = test_id
183        if test_id in self.testcases_by_id:
184            testcase = self.testcases_by_id[test_id]
185            self._last_test = testcase.shortDescription()
186            if not self._last_test:
187                self._last_test = str(testcase)
188        else:
189            self._last_test = test_id
190
191    def add_testclass_with_core(self):
192        if self.last_test_id in self.testcases_by_id:
193            test = self.testcases_by_id[self.last_test_id]
194            class_name = unittest.util.strclass(test.__class__)
195            test_name = "'{}' ({})".format(get_test_description(descriptions,
196                                                                test),
197                                           self.last_test_id)
198        else:
199            test_name = self.last_test_id
200            class_name = re.match(r'((tearDownClass)|(setUpClass)) '
201                                  r'\((.+\..+)\)', test_name).groups()[3]
202        if class_name not in self.testclasess_with_core:
203            self.testclasess_with_core[class_name] = (
204                test_name,
205                self.last_test_vpp_binary,
206                self.last_test_temp_dir)
207
208    def close_pipes(self):
209        self.keep_alive_child_end.close()
210        self.finished_child_end.close()
211        self.result_child_end.close()
212        self.keep_alive_parent_end.close()
213        self.finished_parent_end.close()
214        self.result_parent_end.close()
215
216    def was_successful(self):
217        return self.result.was_successful()
218
219
220def stdouterr_reader_wrapper(unread_testcases, finished_unread_testcases,
221                             read_testcases):
222    read_testcase = None
223    while read_testcases.is_set() or unread_testcases:
224        if finished_unread_testcases:
225            read_testcase = finished_unread_testcases.pop()
226            unread_testcases.remove(read_testcase)
227        elif unread_testcases:
228            read_testcase = unread_testcases.pop()
229        if read_testcase:
230            data = ''
231            while data is not None:
232                sys.stdout.write(data)
233                data = read_testcase.stdouterr_queue.get()
234
235            read_testcase.stdouterr_queue.close()
236            finished_unread_testcases.discard(read_testcase)
237            read_testcase = None
238
239
240def handle_failed_suite(logger, last_test_temp_dir, vpp_pid):
241    if last_test_temp_dir:
242        # Need to create link in case of a timeout or core dump without failure
243        lttd = os.path.basename(last_test_temp_dir)
244        failed_dir = os.getenv('FAILED_DIR')
245        link_path = '%s%s-FAILED' % (failed_dir, lttd)
246        if not os.path.exists(link_path):
247            os.symlink(last_test_temp_dir, link_path)
248        logger.error("Symlink to failed testcase directory: %s -> %s"
249                     % (link_path, lttd))
250
251        # Report core existence
252        core_path = get_core_path(last_test_temp_dir)
253        if os.path.exists(core_path):
254            logger.error(
255                "Core-file exists in test temporary directory: %s!" %
256                core_path)
257            check_core_path(logger, core_path)
258            logger.debug("Running 'file %s':" % core_path)
259            try:
260                info = check_output(["file", core_path])
261                logger.debug(info)
262            except CalledProcessError as e:
263                logger.error("Subprocess returned with return code "
264                             "while running `file' utility on core-file "
265                             "returned: "
266                             "rc=%s", e.returncode)
267            except OSError as e:
268                logger.error("Subprocess returned with OS error while "
269                             "running 'file' utility "
270                             "on core-file: "
271                             "(%s) %s", e.errno, e.strerror)
272            except Exception as e:
273                logger.exception("Unexpected error running `file' utility "
274                                 "on core-file")
275
276    if vpp_pid:
277        # Copy api post mortem
278        api_post_mortem_path = "/tmp/api_post_mortem.%d" % vpp_pid
279        if os.path.isfile(api_post_mortem_path):
280            logger.error("Copying api_post_mortem.%d to %s" %
281                         (vpp_pid, last_test_temp_dir))
282            shutil.copy2(api_post_mortem_path, last_test_temp_dir)
283
284
285def check_and_handle_core(vpp_binary, tempdir, core_crash_test):
286    if is_core_present(tempdir):
287        if debug_core:
288            print('VPP core detected in %s. Last test running was %s' %
289                  (tempdir, core_crash_test))
290            print(single_line_delim)
291            spawn_gdb(vpp_binary, get_core_path(tempdir))
292            print(single_line_delim)
293        elif compress_core:
294            print("Compressing core-file in test directory `%s'" % tempdir)
295            os.system("gzip %s" % get_core_path(tempdir))
296
297
298def handle_cores(failed_testcases):
299    for failed_testcase in failed_testcases:
300        tcs_with_core = failed_testcase.testclasess_with_core
301        if tcs_with_core:
302            for test, vpp_binary, tempdir in tcs_with_core.values():
303                check_and_handle_core(vpp_binary, tempdir, test)
304
305
306def process_finished_testsuite(wrapped_testcase_suite,
307                               finished_testcase_suites,
308                               failed_wrapped_testcases,
309                               results):
310    results.append(wrapped_testcase_suite.result)
311    finished_testcase_suites.add(wrapped_testcase_suite)
312    stop_run = False
313    if failfast and not wrapped_testcase_suite.was_successful():
314        stop_run = True
315
316    if not wrapped_testcase_suite.was_successful():
317        failed_wrapped_testcases.add(wrapped_testcase_suite)
318        handle_failed_suite(wrapped_testcase_suite.logger,
319                            wrapped_testcase_suite.last_test_temp_dir,
320                            wrapped_testcase_suite.vpp_pid)
321
322    return stop_run
323
324
325def run_forked(testcase_suites):
326    wrapped_testcase_suites = set()
327
328    # suites are unhashable, need to use list
329    results = []
330    unread_testcases = set()
331    finished_unread_testcases = set()
332    manager = StreamQueueManager()
333    manager.start()
334    for i in range(concurrent_tests):
335        if testcase_suites:
336            wrapped_testcase_suite = TestCaseWrapper(testcase_suites.pop(0),
337                                                     manager)
338            wrapped_testcase_suites.add(wrapped_testcase_suite)
339            unread_testcases.add(wrapped_testcase_suite)
340        else:
341            break
342
343    read_from_testcases = threading.Event()
344    read_from_testcases.set()
345    stdouterr_thread = threading.Thread(target=stdouterr_reader_wrapper,
346                                        args=(unread_testcases,
347                                              finished_unread_testcases,
348                                              read_from_testcases))
349    stdouterr_thread.start()
350
351    failed_wrapped_testcases = set()
352    stop_run = False
353
354    try:
355        while wrapped_testcase_suites:
356            finished_testcase_suites = set()
357            for wrapped_testcase_suite in wrapped_testcase_suites:
358                while wrapped_testcase_suite.result_parent_end.poll():
359                    wrapped_testcase_suite.result.process_result(
360                        *wrapped_testcase_suite.result_parent_end.recv())
361                    wrapped_testcase_suite.last_heard = time.time()
362
363                while wrapped_testcase_suite.keep_alive_parent_end.poll():
364                    wrapped_testcase_suite.last_test, \
365                        wrapped_testcase_suite.last_test_vpp_binary, \
366                        wrapped_testcase_suite.last_test_temp_dir, \
367                        wrapped_testcase_suite.vpp_pid = \
368                        wrapped_testcase_suite.keep_alive_parent_end.recv()
369                    wrapped_testcase_suite.last_heard = time.time()
370
371                if wrapped_testcase_suite.finished_parent_end.poll():
372                    wrapped_testcase_suite.finished_parent_end.recv()
373                    wrapped_testcase_suite.last_heard = time.time()
374                    stop_run = process_finished_testsuite(
375                        wrapped_testcase_suite,
376                        finished_testcase_suites,
377                        failed_wrapped_testcases,
378                        results) or stop_run
379                    continue
380
381                fail = False
382                if wrapped_testcase_suite.last_heard + test_timeout < \
383                        time.time():
384                    fail = True
385                    wrapped_testcase_suite.logger.critical(
386                        "Child test runner process timed out "
387                        "(last test running was `%s' in `%s')!" %
388                        (wrapped_testcase_suite.last_test,
389                         wrapped_testcase_suite.last_test_temp_dir))
390                elif not wrapped_testcase_suite.child.is_alive():
391                    fail = True
392                    wrapped_testcase_suite.logger.critical(
393                        "Child test runner process unexpectedly died "
394                        "(last test running was `%s' in `%s')!" %
395                        (wrapped_testcase_suite.last_test,
396                         wrapped_testcase_suite.last_test_temp_dir))
397                elif wrapped_testcase_suite.last_test_temp_dir and \
398                        wrapped_testcase_suite.last_test_vpp_binary:
399                    if is_core_present(
400                            wrapped_testcase_suite.last_test_temp_dir):
401                        wrapped_testcase_suite.add_testclass_with_core()
402                        if wrapped_testcase_suite.core_detected_at is None:
403                            wrapped_testcase_suite.core_detected_at = \
404                                time.time()
405                        elif wrapped_testcase_suite.core_detected_at + \
406                                core_timeout < time.time():
407                            wrapped_testcase_suite.logger.critical(
408                                "Child test runner process unresponsive and "
409                                "core-file exists in test temporary directory "
410                                "(last test running was `%s' in `%s')!" %
411                                (wrapped_testcase_suite.last_test,
412                                 wrapped_testcase_suite.last_test_temp_dir))
413                            fail = True
414
415                if fail:
416                    wrapped_testcase_suite.child.terminate()
417                    try:
418                        # terminating the child process tends to leave orphan
419                        # VPP process around
420                        if wrapped_testcase_suite.vpp_pid:
421                            os.kill(wrapped_testcase_suite.vpp_pid,
422                                    signal.SIGTERM)
423                    except OSError:
424                        # already dead
425                        pass
426                    wrapped_testcase_suite.result.crashed = True
427                    wrapped_testcase_suite.result.process_result(
428                        wrapped_testcase_suite.last_test_id, ERROR)
429                    stop_run = process_finished_testsuite(
430                        wrapped_testcase_suite,
431                        finished_testcase_suites,
432                        failed_wrapped_testcases,
433                        results) or stop_run
434
435            for finished_testcase in finished_testcase_suites:
436                # Somewhat surprisingly, the join below may
437                # timeout, even if client signaled that
438                # it finished - so we note it just in case.
439                join_start = time.time()
440                finished_testcase.child.join(test_finished_join_timeout)
441                join_end = time.time()
442                if join_end - join_start >= test_finished_join_timeout:
443                    finished_testcase.logger.error(
444                        "Timeout joining finished test: %s (pid %d)" %
445                        (finished_testcase.last_test,
446                         finished_testcase.child.pid))
447                finished_testcase.close_pipes()
448                wrapped_testcase_suites.remove(finished_testcase)
449                finished_unread_testcases.add(finished_testcase)
450                finished_testcase.stdouterr_queue.put(None)
451                if stop_run:
452                    while testcase_suites:
453                        results.append(TestResult(testcase_suites.pop(0)))
454                elif testcase_suites:
455                    new_testcase = TestCaseWrapper(testcase_suites.pop(0),
456                                                   manager)
457                    wrapped_testcase_suites.add(new_testcase)
458                    unread_testcases.add(new_testcase)
459            time.sleep(0.1)
460    except Exception:
461        for wrapped_testcase_suite in wrapped_testcase_suites:
462            wrapped_testcase_suite.child.terminate()
463            wrapped_testcase_suite.stdouterr_queue.put(None)
464        raise
465    finally:
466        read_from_testcases.clear()
467        stdouterr_thread.join(test_timeout)
468        manager.shutdown()
469
470    handle_cores(failed_wrapped_testcases)
471    return results
472
473
474class SplitToSuitesCallback:
475    def __init__(self, filter_callback):
476        self.suites = {}
477        self.suite_name = 'default'
478        self.filter_callback = filter_callback
479        self.filtered = unittest.TestSuite()
480
481    def __call__(self, file_name, cls, method):
482        test_method = cls(method)
483        if self.filter_callback(file_name, cls.__name__, method):
484            self.suite_name = file_name + cls.__name__
485            if self.suite_name not in self.suites:
486                self.suites[self.suite_name] = unittest.TestSuite()
487            self.suites[self.suite_name].addTest(test_method)
488
489        else:
490            self.filtered.addTest(test_method)
491
492
493test_option = "TEST"
494
495
496def parse_test_option():
497    f = os.getenv(test_option, None)
498    filter_file_name = None
499    filter_class_name = None
500    filter_func_name = None
501    if f:
502        if '.' in f:
503            parts = f.split('.')
504            if len(parts) > 3:
505                raise Exception("Unrecognized %s option: %s" %
506                                (test_option, f))
507            if len(parts) > 2:
508                if parts[2] not in ('*', ''):
509                    filter_func_name = parts[2]
510            if parts[1] not in ('*', ''):
511                filter_class_name = parts[1]
512            if parts[0] not in ('*', ''):
513                if parts[0].startswith('test_'):
514                    filter_file_name = parts[0]
515                else:
516                    filter_file_name = 'test_%s' % parts[0]
517        else:
518            if f.startswith('test_'):
519                filter_file_name = f
520            else:
521                filter_file_name = 'test_%s' % f
522    if filter_file_name:
523        filter_file_name = '%s.py' % filter_file_name
524    return filter_file_name, filter_class_name, filter_func_name
525
526
527def filter_tests(tests, filter_cb):
528    result = unittest.suite.TestSuite()
529    for t in tests:
530        if isinstance(t, unittest.suite.TestSuite):
531            # this is a bunch of tests, recursively filter...
532            x = filter_tests(t, filter_cb)
533            if x.countTestCases() > 0:
534                result.addTest(x)
535        elif isinstance(t, unittest.TestCase):
536            # this is a single test
537            parts = t.id().split('.')
538            # t.id() for common cases like this:
539            # test_classifier.TestClassifier.test_acl_ip
540            # apply filtering only if it is so
541            if len(parts) == 3:
542                if not filter_cb(parts[0], parts[1], parts[2]):
543                    continue
544            result.addTest(t)
545        else:
546            # unexpected object, don't touch it
547            result.addTest(t)
548    return result
549
550
551class FilterByTestOption:
552    def __init__(self, filter_file_name, filter_class_name, filter_func_name):
553        self.filter_file_name = filter_file_name
554        self.filter_class_name = filter_class_name
555        self.filter_func_name = filter_func_name
556
557    def __call__(self, file_name, class_name, func_name):
558        if self.filter_file_name:
559            fn_match = fnmatch.fnmatch(file_name, self.filter_file_name)
560            if not fn_match:
561                return False
562        if self.filter_class_name and class_name != self.filter_class_name:
563            return False
564        if self.filter_func_name and func_name != self.filter_func_name:
565            return False
566        return True
567
568
569class FilterByClassList:
570    def __init__(self, classes_with_filenames):
571        self.classes_with_filenames = classes_with_filenames
572
573    def __call__(self, file_name, class_name, func_name):
574        return '.'.join([file_name, class_name]) in self.classes_with_filenames
575
576
577def suite_from_failed(suite, failed):
578    failed = {x.rsplit('.', 1)[0] for x in failed}
579    filter_cb = FilterByClassList(failed)
580    suite = filter_tests(suite, filter_cb)
581    return suite
582
583
584class AllResults(dict):
585    def __init__(self):
586        super(AllResults, self).__init__()
587        self.all_testcases = 0
588        self.results_per_suite = []
589        self[PASS] = 0
590        self[FAIL] = 0
591        self[ERROR] = 0
592        self[SKIP] = 0
593        self[TEST_RUN] = 0
594        self.rerun = []
595        self.testsuites_no_tests_run = []
596
597    def add_results(self, result):
598        self.results_per_suite.append(result)
599        result_types = [PASS, FAIL, ERROR, SKIP, TEST_RUN]
600        for result_type in result_types:
601            self[result_type] += len(result[result_type])
602
603    def add_result(self, result):
604        retval = 0
605        self.all_testcases += result.testcase_suite.countTestCases()
606        self.add_results(result)
607
608        if result.no_tests_run():
609            self.testsuites_no_tests_run.append(result.testcase_suite)
610            if result.crashed:
611                retval = -1
612            else:
613                retval = 1
614        elif not result.was_successful():
615            retval = 1
616
617        if retval != 0:
618            self.rerun.append(result.testcase_suite)
619
620        return retval
621
622    def print_results(self):
623        print('')
624        print(double_line_delim)
625        print('TEST RESULTS:')
626        print('     Scheduled tests: {}'.format(self.all_testcases))
627        print('      Executed tests: {}'.format(self[TEST_RUN]))
628        print('        Passed tests: {}'.format(
629            colorize(str(self[PASS]), GREEN)))
630        if self[SKIP] > 0:
631            print('       Skipped tests: {}'.format(
632                colorize(str(self[SKIP]), YELLOW)))
633        if self.not_executed > 0:
634            print('  Not Executed tests: {}'.format(
635                colorize(str(self.not_executed), RED)))
636        if self[FAIL] > 0:
637            print('            Failures: {}'.format(
638                colorize(str(self[FAIL]), RED)))
639        if self[ERROR] > 0:
640            print('              Errors: {}'.format(
641                colorize(str(self[ERROR]), RED)))
642
643        if self.all_failed > 0:
644            print('FAILURES AND ERRORS IN TESTS:')
645            for result in self.results_per_suite:
646                failed_testcase_ids = result[FAIL]
647                errored_testcase_ids = result[ERROR]
648                old_testcase_name = None
649                if failed_testcase_ids or errored_testcase_ids:
650                    for failed_test_id in failed_testcase_ids:
651                        new_testcase_name, test_name = \
652                            result.get_testcase_names(failed_test_id)
653                        if new_testcase_name != old_testcase_name:
654                            print('  Testcase name: {}'.format(
655                                colorize(new_testcase_name, RED)))
656                            old_testcase_name = new_testcase_name
657                        print('    FAILURE: {} [{}]'.format(
658                            colorize(test_name, RED), failed_test_id))
659                    for failed_test_id in errored_testcase_ids:
660                        new_testcase_name, test_name = \
661                            result.get_testcase_names(failed_test_id)
662                        if new_testcase_name != old_testcase_name:
663                            print('  Testcase name: {}'.format(
664                                colorize(new_testcase_name, RED)))
665                            old_testcase_name = new_testcase_name
666                        print('      ERROR: {} [{}]'.format(
667                            colorize(test_name, RED), failed_test_id))
668        if self.testsuites_no_tests_run:
669            print('TESTCASES WHERE NO TESTS WERE SUCCESSFULLY EXECUTED:')
670            tc_classes = set()
671            for testsuite in self.testsuites_no_tests_run:
672                for testcase in testsuite:
673                    tc_classes.add(get_testcase_doc_name(testcase))
674            for tc_class in tc_classes:
675                print('  {}'.format(colorize(tc_class, RED)))
676
677        print(double_line_delim)
678        print('')
679
680    @property
681    def not_executed(self):
682        return self.all_testcases - self[TEST_RUN]
683
684    @property
685    def all_failed(self):
686        return self[FAIL] + self[ERROR]
687
688
689def parse_results(results):
690    """
691    Prints the number of scheduled, executed, not executed, passed, failed,
692    errored and skipped tests and details about failed and errored tests.
693
694    Also returns all suites where any test failed.
695
696    :param results:
697    :return:
698    """
699
700    results_per_suite = AllResults()
701    crashed = False
702    failed = False
703    for result in results:
704        result_code = results_per_suite.add_result(result)
705        if result_code == 1:
706            failed = True
707        elif result_code == -1:
708            crashed = True
709
710    results_per_suite.print_results()
711
712    if crashed:
713        return_code = -1
714    elif failed:
715        return_code = 1
716    else:
717        return_code = 0
718    return return_code, results_per_suite.rerun
719
720
721def parse_digit_env(env_var, default):
722    value = os.getenv(env_var, default)
723    if value != default:
724        if value.isdigit():
725            value = int(value)
726        else:
727            print('WARNING: unsupported value "%s" for env var "%s",'
728                  'defaulting to %s' % (value, env_var, default))
729            value = default
730    return value
731
732
733if __name__ == '__main__':
734
735    verbose = parse_digit_env("V", 0)
736
737    test_timeout = parse_digit_env("TIMEOUT", 600)  # default = 10 minutes
738
739    test_finished_join_timeout = 15
740
741    retries = parse_digit_env("RETRIES", 0)
742
743    debug = os.getenv("DEBUG", "n").lower() in ["gdb", "gdbserver"]
744
745    debug_core = os.getenv("DEBUG", "").lower() == "core"
746    compress_core = framework.BoolEnvironmentVariable("CORE_COMPRESS")
747
748    step = framework.BoolEnvironmentVariable("STEP")
749    force_foreground = framework.BoolEnvironmentVariable("FORCE_FOREGROUND")
750
751    run_interactive = debug or step or force_foreground
752
753    try:
754        num_cpus = len(os.sched_getaffinity(0))
755    except AttributeError:
756        num_cpus = multiprocessing.cpu_count()
757    shm_free = psutil.disk_usage('/dev/shm').free
758
759    print('OS reports %s available cpu(s). Free shm: %s' % (
760        num_cpus, "{:,}MB".format(shm_free / (1024 * 1024))))
761
762    test_jobs = os.getenv("TEST_JOBS", "1").lower()  # default = 1 process
763    if test_jobs == 'auto':
764        if run_interactive:
765            concurrent_tests = 1
766            print('Interactive mode required, running on one core')
767        else:
768            shm_max_processes = 1
769            if shm_free < min_req_shm:
770                raise Exception('Not enough free space in /dev/shm. Required '
771                                'free space is at least %sM.'
772                                % (min_req_shm >> 20))
773            else:
774                extra_shm = shm_free - min_req_shm
775                shm_max_processes += extra_shm // shm_per_process
776            concurrent_tests = min(cpu_count(), shm_max_processes)
777            print('Found enough resources to run tests with %s cores'
778                  % concurrent_tests)
779    elif test_jobs.isdigit():
780        concurrent_tests = int(test_jobs)
781        print("Running on %s core(s) as set by 'TEST_JOBS'." %
782              concurrent_tests)
783    else:
784        concurrent_tests = 1
785        print('Running on one core.')
786
787    if run_interactive and concurrent_tests > 1:
788        raise NotImplementedError(
789            'Running tests interactively (DEBUG is gdb or gdbserver or STEP '
790            'is set) in parallel (TEST_JOBS is more than 1) is not supported')
791
792    parser = argparse.ArgumentParser(description="VPP unit tests")
793    parser.add_argument("-f", "--failfast", action='store_true',
794                        help="fast failure flag")
795    parser.add_argument("-d", "--dir", action='append', type=str,
796                        help="directory containing test files "
797                             "(may be specified multiple times)")
798    args = parser.parse_args()
799    failfast = args.failfast
800    descriptions = True
801
802    print("Running tests using custom test runner")  # debug message
803    filter_file, filter_class, filter_func = parse_test_option()
804
805    print("Active filters: file=%s, class=%s, function=%s" % (
806        filter_file, filter_class, filter_func))
807
808    filter_cb = FilterByTestOption(filter_file, filter_class, filter_func)
809
810    ignore_path = os.getenv("VENV_PATH", None)
811    cb = SplitToSuitesCallback(filter_cb)
812    for d in args.dir:
813        print("Adding tests from directory tree %s" % d)
814        discover_tests(d, cb, ignore_path)
815
816    # suites are not hashable, need to use list
817    suites = []
818    tests_amount = 0
819    for testcase_suite in cb.suites.values():
820        tests_amount += testcase_suite.countTestCases()
821        suites.append(testcase_suite)
822
823    print("%s out of %s tests match specified filters" % (
824        tests_amount, tests_amount + cb.filtered.countTestCases()))
825
826    if not running_extended_tests:
827        print("Not running extended tests (some tests will be skipped)")
828
829    attempts = retries + 1
830    if attempts > 1:
831        print("Perform %s attempts to pass the suite..." % attempts)
832
833    if run_interactive and suites:
834        # don't fork if requiring interactive terminal
835        print('Running tests in foreground in the current process')
836        full_suite = unittest.TestSuite()
837        full_suite.addTests(suites)
838        result = VppTestRunner(verbosity=verbose,
839                               failfast=failfast,
840                               print_summary=True).run(full_suite)
841        was_successful = result.wasSuccessful()
842        if not was_successful:
843            for test_case_info in result.failed_test_cases_info:
844                handle_failed_suite(test_case_info.logger,
845                                    test_case_info.tempdir,
846                                    test_case_info.vpp_pid)
847                if test_case_info in result.core_crash_test_cases_info:
848                    check_and_handle_core(test_case_info.vpp_bin_path,
849                                          test_case_info.tempdir,
850                                          test_case_info.core_crash_test)
851
852        sys.exit(not was_successful)
853    else:
854        print('Running each VPPTestCase in a separate background process'
855              ' with {} parallel process(es)'.format(concurrent_tests))
856        exit_code = 0
857        while suites and attempts > 0:
858            results = run_forked(suites)
859            exit_code, suites = parse_results(results)
860            attempts -= 1
861            if exit_code == 0:
862                print('Test run was successful')
863            else:
864                print('%s attempt(s) left.' % attempts)
865        sys.exit(exit_code)
866