Roll third_party/typ to 0.9.5.

third_party/typ/typ/tests/main_test.py

 # Copyright 2014 Dirk Pranke. 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 141 matching lines @@
         pass
 
     setattr(BaseTest, "test_fail", method_fail)
     setattr(BaseTest, "test_pass", method_pass)
     suite = unittest.TestSuite()
     suite.addTest(BaseTest("test_fail"))
     suite.addTest(BaseTest("test_pass"))
     return suite
 """
 
+LOAD_TEST_FILES = {'load_test.py': LOAD_TEST_PY}
 
-LOAD_TEST_FILES = {'load_test.py': LOAD_TEST_PY}
+
+MIXED_TEST_PY = """
+import unittest
+class SampleTest(unittest.TestCase):
+
+  def test_pass_0(self):
+    self.assertEqual(1, 1)
+
+  def test_pass_1(self):
+    self.assertEqual(1, 1)
+
+  def test_fail_0(self):
+    self.assertEqual(1, 2)
+
+  def test_fail_1(self):
+    raise Exception()
+
+  @unittest.skip('Skip for no reason')
+  def test_skip_0(self):
+    pass
+"""
+
+
+LOAD_MANY_TEST_PY = """
+import unittest
+
+def generate_test_case(test_method_name, test_type):
+  class GeneratedTest(unittest.TestCase):
+    pass
+
+  if test_type == 'pass':
+    def test_method(self):
+      self.assertEqual(1, 1)
+  elif test_type == 'fail':
+    def test_method(self):
+      self.assertEqual(1, 2)
+  elif test_type == 'skip':
+    def test_method(self):
+      self.skipTest('Skipped')
+  else:
+    raise Exception
+
+  setattr(GeneratedTest, test_method_name, test_method)
+  return GeneratedTest(test_method_name)
+
+
+def load_tests(loader, standard_tests, pattern):
+  del loader, standard_tests, pattern  # unused
+
+  suite = unittest.TestSuite()
+
+  passed_test_names = [
+    str('test_pass_%s' % i) for i in range(2, 15)]
+
+  failed_test_names = [
+    str('test_fail_%s' % i) for i in range(2, 10)]
+
+  skipped_test_names = [
+    str('test_skip_%s' % i) for i in range(1, 10)]
+
+  for test_method_name in passed_test_names:
+    suite.addTest(generate_test_case(test_method_name, 'pass'))
+
+  for test_method_name in failed_test_names:
+    suite.addTest(generate_test_case(test_method_name, 'fail'))
+
+  for test_method_name in skipped_test_names:
+    suite.addTest(generate_test_case(test_method_name, 'skip'))
+
+  return suite
+"""
+
+
+MANY_TEST_FILES = {
+    'mixed_test.py': MIXED_TEST_PY,         # 2 passes, 2 fails, 1 skip
+    'load_many_test.py': LOAD_MANY_TEST_PY} # 13 passes, 13 fails, 9 skips
 
 
 path_to_main = os.path.join(
     os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
     'runner.py')
 
 
 class TestCli(test_case.MainTestCase):
     prog = [sys.executable, path_to_main]
     files_to_ignore = ['*.pyc']
 
+    def get_test_results_stat(self, test_output):
+      num_passes = test_output.count(' passed\n')
+      num_fails =  test_output.count(' failed unexpectedly:\n')
+      num_skips = test_output.count(' was skipped\n')
+      return num_passes, num_fails, num_skips
+
+    def run_and_check_test_results(self, num_shards):
+      total_passes, total_fails, total_skips = 0, 0, 0
+      min_num_tests_run = float('inf')
+      max_num_tests_run = 0
+      for shard_index in range(num_shards):
+        _, out, _, _ = self.check(
+            ['--total-shards', str(num_shards), '--shard-index',
+             str(shard_index)], files=MANY_TEST_FILES)
+        passes, fails, skips = self.get_test_results_stat(out)
+        total_passes += passes
+        total_fails += fails
+        total_skips += skips
+        num_tests_run = passes + fails
+        min_num_tests_run = min(min_num_tests_run, num_tests_run)
+        max_num_tests_run = max(max_num_tests_run, num_tests_run)
+      self.assertEqual(total_passes, 15)
+      self.assertEqual(total_fails, 10)
+      self.assertEqual(total_skips, 10)
+
+      # Make sure that we don't distribute the tests too unevenly.
+      self.assertLessEqual(max_num_tests_run - min_num_tests_run, 2)
+
     def test_bad_arg(self):
         self.check(['--bad-arg'], ret=2, out='',
                    rerr='.*: error: unrecognized arguments: --bad-arg\n')
         self.check(['-help'], ret=2, out='',
                    rerr=(".*: error: argument -h/--help: "
                          "ignored explicit argument 'elp'\n"))
 
     def test_bad_metadata(self):
         self.check(['--metadata', 'foo'], ret=2, err='',
                    out='Error: malformed --metadata "foo"\n')
@@ skipping 335 matching lines @@
                                   files=SF_TEST_FILES, ret=1, err='')
         self.assertIn('sf_test.SkipClass.test_method failed', out)
         self.assertIn('sf_test.SkipMethods.test_reason failed', out)
         self.assertIn('sf_test.SkipMethods.test_skip_if_true failed', out)
         self.assertIn('sf_test.SkipMethods.test_skip_if_false failed', out)
 
         # --all does not override explicit calls to skipTest(), only
         # the decorators.
         self.assertIn('sf_test.SkipSetup.test_notrun was skipped', out)
 
+    def test_sharding(self):
+      # Test no sharding.
+      self.run_and_check_test_results(1)
+
+      # A typical with 4 shards.
+      self.run_and_check_test_results(4)
+
+      # Case which number of shards is a prime.
+      self.run_and_check_test_results(7)
+
+      # Case which number of shards is more than number of tests.
+      self.run_and_check_test_results(50)
+
     def test_subdir(self):
         files = {
             'foo/__init__.py': '',
             'foo/bar/__init__.py': '',
             'foo/bar/pass_test.py': PASS_TEST_PY
         }
         self.check(['foo/bar'], files=files, ret=0, err='',
                    out=d("""\
                          [1/1] foo.bar.pass_test.PassingTest.test_pass passed
                          1 test run, 0 failures.
@@ skipping 145 matching lines @@
                     modules_to_unload.append(k)
             for k in modules_to_unload:
                 del sys.modules[k]
             sys.path = orig_sys_path
 
         return ret, out, err
 
     def test_debugger(self):
         # TODO: this test seems to hang under coverage.
         pass