1#!/usr/bin/env python3
2# Copyright 2021 the V8 project authors. All rights reserved.
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5"""
6Test integrating the sequence processor into a simple test pipeline.
7"""
8
9import os
10import sys
11import unittest
12
13# Needed because the test runner contains relative imports.
14TOOLS_PATH = os.path.dirname(
15    os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
16sys.path.append(TOOLS_PATH)
17
18from testrunner.testproc import base
19from testrunner.testproc.loader import LoadProc
20from testrunner.testproc.sequence import SequenceProc
21
22
23class FakeExecutionProc(base.TestProc):
24  """Simulates the pipeline sink consuming and running the tests.
25
26  Test execution is simulated for each test by calling run().
27  """
28
29  def __init__(self):
30    super(FakeExecutionProc, self).__init__()
31    self.tests = []
32
33  def next_test(self, test):
34    self.tests.append(test)
35    return True
36
37  def run(self):
38    test = self.tests.pop()
39    self._send_result(test, test.n)
40
41
42class FakeResultObserver(base.TestProcObserver):
43  """Observer to track all results sent back through the pipeline."""
44
45  def __init__(self):
46    super(FakeResultObserver, self).__init__()
47    self.tests = set([])
48
49  def _on_result_for(self, test, result):
50    self.tests.add(test.n)
51
52
53class FakeTest(object):
54  """Simple test representation to differentiate light/heavy tests."""
55
56  def __init__(self, n, is_heavy):
57    self.n = n
58    self.is_heavy = is_heavy
59    self.keep_output = False
60
61
62class TestSequenceProc(unittest.TestCase):
63
64  def _test(self, tests, batch_size, max_heavy):
65    # Set up a simple processing pipeline:
66    # Loader -> observe results -> sequencer -> execution.
67    loader = LoadProc(iter(tests))
68    results = FakeResultObserver()
69    sequence_proc = SequenceProc(max_heavy)
70    execution = FakeExecutionProc()
71    loader.connect_to(results)
72    results.connect_to(sequence_proc)
73    sequence_proc.connect_to(execution)
74
75    # Fill the execution queue (with the number of tests potentially
76    # executed in parallel).
77    loader.load_initial_tests(batch_size)
78
79    # Simulate the execution test by test.
80    while execution.tests:
81      # Assert the invariant of maximum heavy tests executed simultaneously.
82      self.assertLessEqual(
83          sum(int(test.is_heavy) for test in execution.tests), max_heavy)
84
85      # As in the real pipeline, running a test and returning its result
86      # will add another test into the pipeline.
87      execution.run()
88
89    # Ensure that all tests are processed and deliver results.
90    self.assertEqual(set(test.n for test in tests), results.tests)
91
92  def test_wrong_usage(self):
93    with self.assertRaises(Exception):
94      SequenceProc(0)
95
96  def test_no_tests(self):
97    self._test([], 1, 1)
98
99  def test_large_batch_light(self):
100    self._test([
101        FakeTest(0, False),
102        FakeTest(1, False),
103        FakeTest(2, False),
104    ], 4, 1)
105
106  def test_small_batch_light(self):
107    self._test([
108        FakeTest(0, False),
109        FakeTest(1, False),
110        FakeTest(2, False),
111    ], 2, 1)
112
113  def test_large_batch_heavy(self):
114    self._test([
115        FakeTest(0, True),
116        FakeTest(1, True),
117        FakeTest(2, True),
118    ], 4, 1)
119
120  def test_small_batch_heavy(self):
121    self._test([
122        FakeTest(0, True),
123        FakeTest(1, True),
124        FakeTest(2, True),
125    ], 2, 1)
126
127  def test_large_batch_mixed(self):
128    self._test([
129        FakeTest(0, True),
130        FakeTest(1, False),
131        FakeTest(2, True),
132        FakeTest(3, False),
133    ], 4, 1)
134
135  def test_small_batch_mixed(self):
136    self._test([
137        FakeTest(0, True),
138        FakeTest(1, False),
139        FakeTest(2, True),
140        FakeTest(3, False),
141    ], 2, 1)
142
143  def test_large_batch_more_heavy(self):
144    self._test([
145        FakeTest(0, True),
146        FakeTest(1, True),
147        FakeTest(2, True),
148        FakeTest(3, False),
149        FakeTest(4, True),
150        FakeTest(5, True),
151        FakeTest(6, False),
152    ], 4, 2)
153
154  def test_small_batch_more_heavy(self):
155    self._test([
156        FakeTest(0, True),
157        FakeTest(1, True),
158        FakeTest(2, True),
159        FakeTest(3, False),
160        FakeTest(4, True),
161        FakeTest(5, True),
162        FakeTest(6, False),
163    ], 2, 2)
164
165
166if __name__ == '__main__':
167  unittest.main()
168