[test] Implemented status file filter as a processor

Bug: v8:6917
Change-Id: I4b10091a40372e1aa614ac26452e20ed481ab686
Cq-Include-Trybots: luci.v8.try:v8_linux64_fyi_rel_ng
Reviewed-on: https://chromium-review.googlesource.com/856498Reviewed-by: Michael Achenbach <machenbach@chromium.org>
Reviewed-by: Sergiy Byelozyorov <sergiyb@chromium.org>
Commit-Queue: Michał Majewski <majeski@google.com>
Cr-Commit-Position: refs/heads/master@{#50482}

tools/testrunner/local/execution.py

@@ -39,10 +39,6 @@ from . import statusfile
 from . import utils
 from . pool import Pool
 from ..objects import predictable
-from ..testproc.execution import ExecutionProc
-from ..testproc.loader import LoadProc
-from ..testproc.progress import VerboseProgressIndicator, ResultsTracker
-from ..testproc.rerun import RerunProc
 
 
 # Base dir of the v8 checkout.
@@ -215,9 +211,6 @@ class Runner(object):
     return not has_unexpected_output
 
   def Run(self, jobs):
-    if self.context.infra_staging:
-      return self._RunTestProc(jobs)
-
     self.indicator.Starting()
     self._RunInternal(jobs)
     self.indicator.Done()
@@ -227,47 +220,6 @@ class Runner(object):
       return 2
     return 0
 
-  def _RunTestProc(self, jobs):
-    print '>>> Running with test processors'
-    procs = []
-    indicators = self.indicator.ToProgressIndicatorProcs()
-
-    # TODO(majeski): Implement all indicators and remove this filter.
-    indicators = filter(None, indicators)
-
-    loader = LoadProc()
-    procs.append(loader)
-
-    results = ResultsTracker(count_subtests=False)
-    procs.append(results)
-
-    procs += indicators
-
-    if self.context.rerun_failures_count:
-      procs.append(RerunProc(
-          self.context.rerun_failures_count,
-          self.context.rerun_failures_max
-      ))
-
-    execproc = ExecutionProc(jobs, self.context)
-    procs.append(execproc)
-
-    for i in xrange(0, len(procs) - 1):
-      procs[i].connect_to(procs[i + 1])
-
-    loader.load_tests(self.tests)
-    for indicator in indicators:
-      indicator.starting()
-    execproc.start()
-    for indicator in indicators:
-      indicator.finished()
-
-    if results.failed:
-      return 1
-    if results.remaining:
-      return 2
-    return 0
-
   def _RunInternal(self, jobs):
     pool = Pool(jobs)
     test_map = {}

tools/testrunner/standard_runner.py

@@ -26,6 +26,12 @@ from testrunner.local import verbose
 from testrunner.local.variants import ALL_VARIANTS
 from testrunner.objects import context
 from testrunner.objects import predictable
+from testrunner.testproc.execution import ExecutionProc
+from testrunner.testproc.filter import StatusFileFilterProc
+from testrunner.testproc.loader import LoadProc
+from testrunner.testproc.progress import (VerboseProgressIndicator,
+                                          ResultsTracker)
+from testrunner.testproc.rerun import RerunProc
 
 
 TIMEOUT_DEFAULT = 60
@@ -466,7 +472,9 @@ class StandardTestRunner(base_runner.BaseTestRunner):
         if options.warn_unused:
           tests = [(t.name, t.variant) for t in s.tests]
           s.statusfile.warn_unused_rules(tests, check_variant_rules=True)
-        s.FilterTestCasesByStatus(options.slow_tests, options.pass_fail_tests)
+
+        if not options.infra_staging:
+          s.FilterTestCasesByStatus(options.slow_tests, options.pass_fail_tests)
         s.tests = self._shard_tests(s.tests, options)
 
         for t in s.tests:
@@ -502,9 +510,14 @@ class StandardTestRunner(base_runner.BaseTestRunner):
         outproc_factory = predictable.get_outproc
       else:
         outproc_factory = None
-      runner = execution.Runner(suites, progress_indicator, ctx,
-                                outproc_factory)
-      exit_code = runner.Run(options.j)
+
+      if options.infra_staging:
+        exit_code = self._run_test_procs(suites, options, progress_indicator,
+                                         ctx, outproc_factory)
+      else:
+        runner = execution.Runner(suites, progress_indicator, ctx,
+                                  outproc_factory)
+        exit_code = runner.Run(options.j)
       overall_duration = time.time() - start_time
 
       if options.time:
@@ -572,6 +585,57 @@ class StandardTestRunner(base_runner.BaseTestRunner):
         count += 1
       return shard
 
+    def _run_test_procs(self, suites, options, progress_indicator, context,
+                        outproc_factory):
+      jobs = options.j
+
+      print '>>> Running with test processors'
+      procs = []
+      indicators = progress_indicator.ToProgressIndicatorProcs()
+
+      # TODO(majeski): Implement all indicators and remove this filter.
+      indicators = filter(None, indicators)
+
+      loader = LoadProc()
+      procs.append(loader)
+
+      results = ResultsTracker(count_subtests=False)
+
+      procs.append(StatusFileFilterProc(options.slow_tests,
+                                        options.pass_fail_tests))
+
+      procs.append(results)
+
+      procs += indicators
+
+      if context.rerun_failures_count:
+        procs.append(RerunProc(
+            context.rerun_failures_count,
+            context.rerun_failures_max
+        ))
+
+      execproc = ExecutionProc(jobs, context)
+      procs.append(execproc)
+
+      for i in xrange(0, len(procs) - 1):
+        procs[i].connect_to(procs[i + 1])
+
+      tests = [t for s in suites for t in s.tests]
+      tests.sort(key=lambda t: t.is_slow, reverse=True)
+      loader.load_tests(tests)
+      for indicator in indicators:
+        indicator.starting()
+      execproc.start()
+      for indicator in indicators:
+        indicator.finished()
+
+      if results.failed:
+        return 1
+      if results.remaining:
+        return 2
+      return 0
+
+
 
 if __name__ == '__main__':
   sys.exit(StandardTestRunner().execute())

tools/testrunner/testproc/base.py

@@ -38,6 +38,8 @@ class TestProc(object):
     """
     Method called by previous processor whenever it produces new test.
     This method shouldn't be called by anyone except previous processor.
+
+    Returns: bool whether test will be processed.
     """
     raise NotImplementedError()
 
@@ -65,7 +67,7 @@ class TestProc(object):
 
   def _send_result(self, test, result, is_last=True):
     """Helper method for sending result to the previous processor."""
-    return self._prev_proc.result_for(test, result, is_last=is_last)
+    self._prev_proc.result_for(test, result, is_last=is_last)
 
 
 
@@ -74,7 +76,7 @@ class TestProcObserver(TestProc):
 
   def next_test(self, test):
     self._on_next_test(test)
-    self._send_test(test)
+    return self._send_test(test)
 
   def result_for(self, test, result, is_last):
     self._on_result_for(test, result, is_last)
@@ -128,9 +130,10 @@ class TestProcProducer(TestProc):
     raise NotImplementedError()
 
   ### Managing subtests
-  def _create_subtest(self, test, subtest_id):
+  def _create_subtest(self, test, subtest_id, **kwargs):
     """Creates subtest with subtest id <processor name>-`subtest_id`."""
-    return test.create_subtest(self, '%s-%s' % (self._name, subtest_id))
+    return test.create_subtest(self, '%s-%s' % (self._name, subtest_id),
+                               **kwargs)
 
   def _get_subtest_origin(self, subtest):
     """Returns parent test that current processor used to create the subtest.
@@ -139,3 +142,17 @@ class TestProcProducer(TestProc):
     while subtest.processor and subtest.processor is not self:
       subtest = subtest.origin
     return subtest.origin
+
+
+class TestProcFilter(TestProc):
+  """Processor for filtering tests."""
+
+  def next_test(self, test):
+    return not self._filter(test) and self._send_test(test)
+
+  def result_for(self, test, result, is_last):
+    self._send_result(test, result, is_last)
+
+  def _filter(self, test):
+    """Returns whether test should be filtered out."""
+    raise NotImplementedError()

tools/testrunner/testproc/execution.py

@@ -82,5 +82,7 @@ class ExecutionProc(base.TestProc):
     outproc = test.output_proc
     self._pool.add([Job(test_id, test.cmd, outproc)])
 
+    return True
+
   def result_for(self, test, result, is_last):
     assert False, 'ExecutionProc cannot receive results'

tools/testrunner/testproc/filter.py

+# Copyright 2018 the V8 project authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+from . import base
+
+
+class StatusFileFilterProc(base.TestProcFilter):
+  """Filters tests by outcomes from status file.
+
+  Status file has to be loaded before using this function.
+
+  Args:
+    slow_tests_mode: What to do with slow tests.
+    pass_fail_tests_mode: What to do with pass or fail tests.
+
+  Mode options:
+    None (default): don't skip
+    "skip": skip if slow/pass_fail
+    "run": skip if not slow/pass_fail
+  """
+
+  def __init__(self, slow_tests_mode, pass_fail_tests_mode):
+    super(StatusFileFilterProc, self).__init__()
+    self._slow_tests_mode = slow_tests_mode
+    self._pass_fail_tests_mode = pass_fail_tests_mode
+
+  def _filter(self, test):
+    return (
+      test.do_skip or
+      self._skip_slow(test.is_slow) or
+      self._skip_pass_fail(test.is_pass_or_fail)
+    )
+
+  def _skip_slow(self, is_slow):
+    return (
+      (self._slow_tests_mode == 'run' and not is_slow) or
+      (self._slow_tests_mode == 'skip' and is_slow)
+    )
+
+  def _skip_pass_fail(self, is_pass_fail):
+    return (
+      (self._pass_fail_tests_mode == 'run' and not is_pass_fail) or
+      (self._pass_fail_tests_mode == 'skip' and is_pass_fail)
+    )

tools/testrunner/testproc/progress.py

@@ -33,8 +33,6 @@ class ResultsTracker(base.TestProcObserver):
   def _on_next_test(self, test):
     self.total += 1
     self.remaining += 1
-    # TODO(majeski): If count_subtests is set get number of subtests from the
-    # next proc.
 
   def _on_result_for(self, test, result, is_last):
     if not is_last and not self.count_subtests:
@@ -61,8 +59,6 @@ class SimpleProgressIndicator(ProgressIndicator):
     self._total = 0
 
   def _on_next_test(self, test):
-    # TODO(majeski): Collect information about subtests, e.g. for each test
-    # we create multiple variants.
     self._total += 1
 
   def _on_result_for(self, test, result, is_last):

tools/testrunner/testproc/rerun.py

@@ -15,6 +15,7 @@ class RerunProc(base.TestProcProducer):
   def _next_test(self, test):
     self._init_test(test)
     self._send_next_subtest(test)
+    return True
 
   def _result_for(self, test, subtest, result, is_last):
     # Rerun processor cannot be placed before any processor that produces more

tools/unittests/run_tests_test.py

@@ -309,7 +309,10 @@ class SystemTest(unittest.TestCase):
       # TODO(machenbach): Test some more implications of the auto-detected
       # options, e.g. that the right env variables are set.
 
-  def testSkips(self):
+  def testSkipsProc(self):
+    self.testSkips(infra_staging=True)
+
+  def testSkips(self, infra_staging=False):
     """Test skipping tests in status file for a specific variant."""
     with temp_base() as basedir:
       result = run_tests(
@@ -318,6 +321,7 @@ class SystemTest(unittest.TestCase):
           '--progress=verbose',
           '--variants=nooptimization',
           'sweet/strawberries',
+          infra_staging=infra_staging,
       )
       self.assertIn('Running 0 tests', result.stdout, result)
       self.assertEqual(0, result.returncode, result)