xref: /aosp_15_r20/external/toolchain-utils/bestflags/pipeline_worker_test.py (revision 760c253c1ed00ce9abd48f8546f08516e57485fe)

# Copyright 2013 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Unittest for the pipeline_worker functions in the build/test stage.

Part of the Chrome build flags optimization.

This module tests the helper method and the worker method.
"""

__author__ = "[email protected] (Yuheng Long)"

import multiprocessing
import random
import sys
import unittest

from mock_task import MockTask
import pipeline_process
import pipeline_worker


# Pick an integer at random.
TEST_STAGE = -3


def MockTaskCostGenerator():
    """Calls a random number generator and returns a negative number."""
    return random.randint(-sys.maxint - 1, -1)


class PipelineWorkerTest(unittest.TestCase):
    """This class tests the pipeline_worker functions.

    Given the same identifier, the cost should result the same from the
    pipeline_worker functions.
    """

    def testHelper(self):
        """ "Test the helper.

        Call the helper method twice, and test the results. The results should be
        the same, i.e., the cost should be the same.
        """

        # Set up the input, helper and output queue for the helper method.
        manager = multiprocessing.Manager()
        helper_queue = manager.Queue()
        result_queue = manager.Queue()
        completed_queue = manager.Queue()

        # Set up the helper process that holds the helper method.
        helper_process = multiprocessing.Process(
            target=pipeline_worker.Helper,
            args=(TEST_STAGE, {}, helper_queue, completed_queue, result_queue),
        )
        helper_process.start()

        # A dictionary defines the mock result to the helper.
        mock_result = {1: 1995, 2: 59, 9: 1027}

        # Test if there is a task that is done before, whether the duplicate task
        # will have the same result. Here, two different scenarios are tested. That
        # is the mock results are added to the completed_queue before and after the
        # corresponding mock tasks being added to the input queue.
        completed_queue.put((9, mock_result[9]))

        # The output of the helper should contain all the following tasks.
        results = [1, 1, 2, 9]

        # Testing the correctness of having tasks having the same identifier, here
        # 1.
        for result in results:
            helper_queue.put(
                MockTask(TEST_STAGE, result, MockTaskCostGenerator())
            )

        completed_queue.put((2, mock_result[2]))
        completed_queue.put((1, mock_result[1]))

        # Signal there is no more duplicate task.
        helper_queue.put(pipeline_process.POISONPILL)
        helper_process.join()

        while results:
            task = result_queue.get()
            identifier = task.GetIdentifier(TEST_STAGE)
            self.assertTrue(identifier in results)
            if identifier in mock_result:
                self.assertTrue(
                    task.GetResult(TEST_STAGE), mock_result[identifier]
                )
            results.remove(identifier)

    def testWorker(self):
        """ "Test the worker method.

        The worker should process all the input tasks and output the tasks to the
        helper and result queue.
        """

        manager = multiprocessing.Manager()
        result_queue = manager.Queue()
        completed_queue = manager.Queue()

        # A dictionary defines the mock tasks and their corresponding results.
        mock_work_tasks = {1: 86, 2: 788}

        mock_tasks = []

        for flag, cost in mock_work_tasks.iteritems():
            mock_tasks.append(MockTask(TEST_STAGE, flag, cost))

        # Submit the mock tasks to the worker.
        for mock_task in mock_tasks:
            pipeline_worker.Worker(
                TEST_STAGE, mock_task, completed_queue, result_queue
            )

        # The tasks, from the output queue, should be the same as the input and
        # should be performed.
        for task in mock_tasks:
            output = result_queue.get()
            self.assertEqual(output, task)
            self.assertTrue(output.Done(TEST_STAGE))

        # The tasks, from the completed_queue, should be defined in the
        # mock_work_tasks dictionary.
        for flag, cost in mock_work_tasks.iteritems():
            helper_input = completed_queue.get()
            self.assertEqual(helper_input, (flag, cost))


if __name__ == "__main__":
    unittest.main()