Roll typ to the latest version

third_party/typ/typ/runner.py

 # Copyright 2014 Google Inc. All rights reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
 #    http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
@@ skipping 93 matching lines @@
         self.cov = None
         self.context = None
         self.coverage_source = None
         self.host = host or Host()
         self.loader = unittest.loader.TestLoader()
         self.printer = None
         self.setup_fn = None
         self.stats = None
         self.teardown_fn = None
         self.top_level_dir = None
+        self.top_level_dirs = []
         self.win_multiprocessing = WinMultiprocessing.spawn
         self.final_responses = []
 
         # initialize self.args to the defaults.
         parser = ArgumentParser(self.host)
         self.parse_args(parser, [])
 
     def main(self, argv=None, **defaults):
         parser = ArgumentParser(self.host)
         self.parse_args(parser, argv, **defaults)
@@ skipping 28 matching lines @@
 
         if self.args.version:
             self.print_(VERSION)
             return ret, None, None
 
         should_spawn = self._check_win_multiprocessing()
         if should_spawn:
             return self._spawn(test_set)
 
         ret = self._set_up_runner()
-        if ret:  # pragma: no cover
+        if ret:
             return ret, None, None
 
         find_start = h.time()
         if self.cov:  # pragma: no cover
             self.cov.erase()
             self.cov.start()
 
         full_results = None
         result_set = ResultSet()
 
@@ skipping 109 matching lines @@
         return ret, full_results, trace
 
     def _set_up_runner(self):
         h = self.host
         args = self.args
 
         self.stats = Stats(args.status_format, h.time, args.jobs)
         self.printer = Printer(
             self.print_, args.overwrite, args.terminal_width)
 
-        self.top_level_dir = args.top_level_dir
-        if not self.top_level_dir:
-            if args.tests and h.isdir(args.tests[0]):
-                # TODO: figure out what to do if multiple files are
-                # specified and they don't all have the same correct
-                # top level dir.
-                d = h.realpath(h.dirname(args.tests[0]))
-                if h.exists(d, '__init__.py'):
-                    top_dir = d
+        if self.args.top_level_dirs and self.args.top_level_dir:
+            self.print_(
+                'Cannot specify both --top-level-dir and --top-level-dirs',
+                stream=h.stderr)
+            return 1
+
+        self.top_level_dirs = args.top_level_dirs
+        if not self.top_level_dirs and args.top_level_dir:
+            self.top_level_dirs = [args.top_level_dir]
+
+        if not self.top_level_dirs:
+            for test in [t for t in args.tests if h.exists(t)]:
+                if h.isdir(test):
+                    top_dir = test
                 else:
-                    top_dir = args.tests[0]
-            else:
-                top_dir = h.getcwd()
+                    top_dir = h.dirname(test)
+                while h.exists(top_dir, '__init__.py'):
+                    top_dir = h.dirname(top_dir)
+                top_dir = h.realpath(top_dir)
+                if not top_dir in self.top_level_dirs:
+                    self.top_level_dirs.append(top_dir)
+        if not self.top_level_dirs:
+            top_dir = h.getcwd()
             while h.exists(top_dir, '__init__.py'):
                 top_dir = h.dirname(top_dir)
-            self.top_level_dir = h.realpath(top_dir)
+            top_dir = h.realpath(top_dir)
+            self.top_level_dirs.append(top_dir)
 
-        h.add_to_path(self.top_level_dir)
+        if not self.top_level_dir and self.top_level_dirs:
+            self.top_level_dir = self.top_level_dirs[0]
+
+        for path in self.top_level_dirs:
+            h.add_to_path(path)
 
         for path in args.path:
             h.add_to_path(path)
 
         if args.coverage:  # pragma: no cover
             try:
                 import coverage
             except ImportError:
-                h.print_("Error: coverage is not installed")
                 return 1
+
             source = self.args.coverage_source
             if not source:
-                source = [self.top_level_dir] + self.args.path
+                source = self.top_level_dirs + self.args.path
             self.coverage_source = source
             self.cov = coverage.coverage(source=self.coverage_source,
                                          data_suffix=True)
             self.cov.erase()
         return 0
 
     def find_tests(self, args):
         test_set = TestSet()
 
         orig_skip = unittest.skip
         orig_skip_if = unittest.skipIf
         if args.all:
             unittest.skip = lambda reason: lambda x: x
             unittest.skipIf = lambda condition, reason: lambda x: x
 
         try:
             names = self._name_list_from_args(args)
             classifier = self.classifier or _default_classifier(args)
 
             for name in names:
                 try:
                     self._add_tests_to_set(test_set, args.suffixes,
-                                           self.top_level_dir, classifier,
+                                           self.top_level_dirs, classifier,
                                            name)
                 except (AttributeError, ImportError, SyntaxError) as e:
                     ex_str = traceback.format_exc()
                     self.print_('Failed to load "%s" in find_tests: %s' %
                                 (name, e))
                     self.print_('  %s' %
                                 '\n  '.join(ex_str.splitlines()))
                     self.print_(ex_str)
                     return 1, None
                 except _AddTestsError as e:
@@ skipping 22 matching lines @@
     def _name_list_from_args(self, args):
         if args.tests:
             names = args.tests
         elif args.file_list:
             if args.file_list == '-':
                 s = self.host.stdin.read()
             else:
                 s = self.host.read_text_file(args.file_list)
             names = [line.strip() for line in s.splitlines()]
         else:
-            names = [self.top_level_dir]
+            names = self.top_level_dirs
         return names
 
-    def _add_tests_to_set(self, test_set, suffixes, top_level_dir, classifier,
+    def _add_tests_to_set(self, test_set, suffixes, top_level_dirs, classifier,
                           name):
         h = self.host
         loader = self.loader
         add_tests = _test_adder(test_set, classifier)
 
-        if h.isfile(name):
-            rpath = h.relpath(name, top_level_dir)
-            if rpath.endswith('.py'):
-                rpath = rpath[:-3]
-            module = rpath.replace(h.sep, '.')
-            add_tests(loader.loadTestsFromName(module))
-        elif h.isdir(name):
-            for suffix in suffixes:
-                add_tests(loader.discover(name, suffix, top_level_dir))
-        else:
-            possible_dir = name.replace('.', h.sep)
-            if h.isdir(top_level_dir, possible_dir):
+        found = set()
+        for d in top_level_dirs:
+            if h.isfile(name):
+                rpath = h.relpath(name, d)
+                if rpath.startswith('..'):
+                    continue
+                if rpath.endswith('.py'):
+                    rpath = rpath[:-3]
+                module = rpath.replace(h.sep, '.')
+                if module not in found:
+                    found.add(module)
+                    add_tests(loader.loadTestsFromName(module))
+            elif h.isdir(name):
+                rpath = h.relpath(name, d)
+                if rpath.startswith('..'):
+                    continue
                 for suffix in suffixes:
-                    path = h.join(top_level_dir, possible_dir)
-                    suite = loader.discover(path, suffix, top_level_dir)
-                    add_tests(suite)
+                    if not name in found:
+                        found.add(name + '/' + suffix)
+                        add_tests(loader.discover(name, suffix, d))
             else:
-                add_tests(loader.loadTestsFromName(name))
+                possible_dir = name.replace('.', h.sep)
+                if h.isdir(d, possible_dir):
+                    for suffix in suffixes:
+                        path = h.join(d, possible_dir)
+                        if not path in found:
+                            found.add(path + '/' + suffix)
+                            suite = loader.discover(path, suffix, d)
+                            add_tests(suite)
+                elif not name in found:
+                    found.add(name)
+                    add_tests(loader.loadTestsFromName(name))
 
         # pylint: disable=no-member
         if hasattr(loader, 'errors') and loader.errors:  # pragma: python3
             # In Python3's version of unittest, loader failures get converted
             # into failed test cases, rather than raising exceptions. However,
             # the errors also get recorded so you can err out immediately.
             raise ImportError(loader.errors)
 
     def _run_tests(self, result_set, test_set):
         h = self.host
@@ skipping 323 matching lines @@
         self.coverage = parent.args.coverage and parent.args.jobs > 1
         self.coverage_source = parent.coverage_source
         self.dry_run = parent.args.dry_run
         self.loader = parent.loader
         self.passthrough = parent.args.passthrough
         self.context = parent.context
         self.setup_fn = parent.setup_fn
         self.teardown_fn = parent.teardown_fn
         self.context_after_setup = None
         self.top_level_dir = parent.top_level_dir
+        self.top_level_dirs = parent.top_level_dirs
         self.loaded_suites = {}
         self.cov = None
 
 
 def _setup_process(host, worker_num, child):
     child.host = host
     child.worker_num = worker_num
     # pylint: disable=protected-access
 
     if child.coverage:  # pragma: no cover
@@ skipping 185 matching lines @@
     return new_suite
 
 
 def _sort_inputs(inps):
     return sorted(inps, key=lambda inp: inp.name)
 
 
 if __name__ == '__main__':  # pragma: no cover
     sys.modules['__main__'].__file__ = path_to_file
     sys.exit(main(win_multiprocessing=WinMultiprocessing.importable))