| #!/usr/bin/env vpython3 |
| # Copyright 2024 The Chromium Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| """Tests for output_adapter.py""" |
| |
| import io |
| import logging |
| import sys |
| import unittest |
| from unittest import mock |
| |
| import output_adapter |
| |
| |
| class PassthroughAdapterTests(unittest.TestCase): |
| def setUp(self): |
| patch_terminal_size = mock.patch('os.get_terminal_size') |
| mock_terminal_size = patch_terminal_size.start() |
| mock_terminal_size.return_value = (128, 1) |
| self.addCleanup(patch_terminal_size.stop) |
| |
| def testBasic(self): |
| adapter = output_adapter.PassthroughAdapter() |
| with self.assertLogs('basic_logger', level=logging.DEBUG) as info_log: |
| fake_output = """@@@SEED_STEP@fake_step@@@ |
| @@@STEP_CURSOR@fake_step@@@ |
| @@@STEP_STARTED@@@ |
| @@@SET_BUILD_PROPERTY@fake_property@"value"@@@ |
| @@@STEP_LOG_LINE@run_recipe@fake_step_log_line@@@ |
| @@@STEP_LOG_END@run_recipe@@@ |
| @@@STEP_LOG_END@memory_profile@@@ |
| fake std_out text""" |
| lines = fake_output.split('\n') |
| for line in lines: |
| adapter.ProcessLine(line) |
| for line in lines: |
| self.assertIn('DEBUG:basic_logger:' + line, info_log.output) |
| |
| |
| class LegacyOutputAdapterTests(unittest.TestCase): |
| def setUp(self): |
| patch_terminal_size = mock.patch('os.get_terminal_size') |
| mock_terminal_size = patch_terminal_size.start() |
| mock_terminal_size.return_value = (128, 1) |
| |
| patch_stdout_isatty = mock.patch('sys.stdout.isatty') |
| mock_stdout_isatty = patch_stdout_isatty.start() |
| mock_stdout_isatty.return_value = True |
| patch_stderr_isatty = mock.patch('sys.stderr.isatty') |
| mock_stderr_isatty = patch_stderr_isatty.start() |
| mock_stderr_isatty.return_value = True |
| |
| self.addCleanup(patch_terminal_size.stop) |
| self.addCleanup(patch_stdout_isatty.stop) |
| self.addCleanup(patch_stderr_isatty.stop) |
| |
| def testNoRecipeEngineOutput(self): |
| adapter = output_adapter.LegacyOutputAdapter() |
| with self.assertLogs('default_logger', level=logging.INFO) as root_log: |
| with self.assertLogs('basic_logger', level=logging.DEBUG) as info_log: |
| fake_output = """@@@SEED_STEP@fake_step@@@ |
| @@@STEP_CURSOR@fake_step@@@ |
| @@@STEP_STARTED@@@ |
| @@@SET_BUILD_PROPERTY@fake_property@"value"@@@ |
| @@@STEP_LOG_LINE@run_recipe@fake_step_log_line@@@ |
| @@@STEP_LOG_END@run_recipe@@@ |
| @@@STEP_LOG_END@memory_profile@@@ |
| fake std_out text |
| @@@STEP_LOG_LINE@utr_log@utr-specific log line@@@ |
| @@@STEP_LOG_END@utr_log@@@ |
| @@@STEP_CLOSED@@@""" |
| for line in fake_output.split('\n'): |
| adapter.ProcessLine(line) |
| self.assertEqual( |
| root_log.output, ['INFO:default_logger:\n[cyan]Running: fake_step[/]'] |
| ) |
| self.assertEqual( |
| info_log.output, |
| [ |
| 'INFO:basic_logger:fake std_out text', |
| 'INFO:basic_logger:utr-specific log line', |
| ], |
| ) |
| |
| def testStepNameProcessor(self): |
| adapter = output_adapter.LegacyOutputAdapter() |
| with self.assertLogs('default_logger', level=logging.DEBUG) as root_log: |
| with self.assertLogs('basic_logger', level=logging.DEBUG) as info_log: |
| fake_output = """@@@SEED_STEP@isolate tests@@@ |
| @@@STEP_CURSOR@isolate tests@@@ |
| @@@STEP_STARTED@@@ |
| @@@SET_BUILD_PROPERTY@fake_property@"value"@@@ |
| @@@STEP_LOG_LINE@run_recipe@fake_step_log_line@@@ |
| @@@STEP_LOG_END@run_recipe@@@ |
| @@@STEP_LOG_END@memory_profile@@@ |
| fake std_out text |
| @@@STEP_CLOSED@@@""" |
| for line in fake_output.split('\n'): |
| adapter.ProcessLine(line) |
| self.assertEqual( |
| root_log.output, |
| ['DEBUG:default_logger:\n[cyan]Running: isolate tests[/]'], |
| ) |
| self.assertEqual( |
| info_log.output, ['DEBUG:basic_logger:fake std_out text'] |
| ) |
| |
| def testTaskTrigger(self): |
| adapter = output_adapter.LegacyOutputAdapter() |
| with self.assertLogs('basic_logger', level=logging.INFO) as info_log: |
| fake_output = """@@@SEED_STEP@test_pre_run.[trigger] fake_test (2)@@@ |
| @@@STEP_CURSOR@test_pre_run.[trigger] fake_test (2)@@@ |
| @@@STEP_STARTED@@@ |
| @@@SET_BUILD_PROPERTY@fake_property@"value"@@@ |
| @@@STEP_LOG_LINE@run_recipe@fake_step_log_line@@@ |
| @@@STEP_LOG_END@run_recipe@@@ |
| @@@STEP_LINK@task UI: fake_suite/fakeos/1234567890123456/fake_builder/-1:0:32@https://chromium-swarm.appspot.com/task?id=1234567890123456@@@ |
| fake std_out text |
| @@@STEP_CLOSED@@@""" |
| for line in fake_output.split('\n'): |
| adapter.ProcessLine(line) |
| self.assertEqual( |
| info_log.output, |
| [ |
| ( |
| 'INFO:basic_logger:Triggered fake_test (2): ' |
| 'https://chromium-swarm.appspot.com/task?id=1234567890123456' |
| ), |
| 'INFO:basic_logger:fake std_out text', |
| ], |
| ) |
| self.assertIn('fake_test', adapter._step_to_processors) |
| self.assertIn('fake_test', adapter._step_to_log_level) |
| |
| def testCollectTasks(self): |
| # pylint: disable=line-too-long |
| adapter = output_adapter.LegacyOutputAdapter() |
| with self.assertLogs('default_logger', level=logging.INFO) as root_log: |
| with self.assertLogs('single_line_logger') as collect_steps_log: |
| fake_output = """@@@SEED_STEP@collect tasks.wait for tasks@@@ |
| @@@STEP_CURSOR@collect tasks.wait for tasks@@@ |
| @@@STEP_STARTED@@@ |
| @@@SET_BUILD_PROPERTY@fake_property@"value"@@@ |
| @@@STEP_LOG_LINE@run_recipe@fake_step_log_line@@@ |
| @@@STEP_LOG_END@run_recipe@@@ |
| 0000-00-00 00:00:00,884 - root: [INFO] prpc cmd: prpc call chromium-swarm.appspot.com swarming.v2.Tasks.ListTaskStates, stdin: {"task_id": ["1111", "22222"]} |
| 0000-00-00 00:00:00,660 - root: [INFO] sleeping for 15 seconds |
| 0000-00-00 00:00:00,884 - root: [INFO] prpc cmd: prpc call chromium-swarm.appspot.com swarming.v2.Tasks.ListTaskStates, stdin: {"task_id": ["1111", "22222"]} |
| 0000-00-00 00:00:00,660 - root: [INFO] sleeping for 15 seconds |
| 0000-00-00 00:00:00,884 - root: [INFO] prpc cmd: prpc call chromium-swarm.appspot.com swarming.v2.Tasks.ListTaskStates, stdin: {"task_id": ["1111"]} |
| 0000-00-00 00:00:00,660 - root: [INFO] sleeping for 15 seconds |
| 0000-00-00 00:00:00,884 - root: [INFO] prpc cmd: prpc call chromium-swarm.appspot.com swarming.v2.Tasks.ListTaskStates, stdin: {"task_id": ["1111"]} |
| 0000-00-00 00:00:00,660 - root: [INFO] sleeping for 15 seconds |
| 0000-00-00 00:00:00,884 - root: [INFO] prpc cmd: prpc call chromium-swarm.appspot.com swarming.v2.Tasks.ListTaskStates, stdin: {"task_id": ["1111"]} |
| 0000-00-00 00:00:00,660 - root: [INFO] sleeping for 15 seconds |
| 0000-00-00 00:00:00,884 - root: [INFO] prpc cmd: prpc call chromium-swarm.appspot.com swarming.v2.Tasks.ListTaskStates, stdin: {"task_id": ["1111"]} |
| 0000-00-00 00:00:00,660 - root: [INFO] sleeping for 15 seconds |
| 0000-00-00 00:00:00,884 - root: [INFO] prpc cmd: prpc call chromium-swarm.appspot.com swarming.v2.Tasks.ListTaskStates, stdin: {"task_id": ["1111"]} |
| 0000-00-00 00:00:00,660 - root: [INFO] sleeping for 15 seconds |
| @@@STEP_CLOSED@@@""" |
| for line in fake_output.split('\n'): |
| adapter.ProcessLine(line) |
| self.assertEqual( |
| root_log.output, |
| [ |
| 'INFO:default_logger:\n[cyan]Running: collect tasks.wait for tasks[/]' |
| ], |
| ) |
| # The ninja statuses are sent to another logger to remove new lines |
| log_prefix = 'INFO:single_line_logger:\x1b[2K\r' |
| self.assertEqual( |
| collect_steps_log.output, |
| [ |
| log_prefix + 'Still waiting on: 2 shard(s).', |
| log_prefix + 'Still waiting on: 2 shard(s)..', |
| log_prefix + 'Still waiting on: 1 shard(s)...', |
| log_prefix + 'Still waiting on: 1 shard(s)....', |
| log_prefix + 'Still waiting on: 1 shard(s).....', |
| log_prefix + 'Still waiting on: 1 shard(s).', |
| log_prefix + 'Still waiting on: 1 shard(s)..', |
| log_prefix + 'Still waiting on: 0 shard(s)...', |
| ], |
| ) |
| |
| def testResultOutput(self): |
| adapter = output_adapter.LegacyOutputAdapter() |
| with self.assertLogs('basic_logger', level=logging.DEBUG) as info_log: |
| fake_output = """@@@SEED_STEP@test_pre_run.[trigger] fake_test@@@ |
| @@@STEP_CURSOR@test_pre_run.[trigger] fake_test@@@ |
| @@@STEP_STARTED@@@ |
| @@@SET_BUILD_PROPERTY@fake_property@"value"@@@ |
| @@@STEP_LOG_LINE@run_recipe@fake_step_log_line@@@ |
| @@@STEP_LOG_END@run_recipe@@@ |
| @@@STEP_LOG_END@memory_profile@@@ |
| @@@STEP_CLOSED@@@ |
| @@@SEED_STEP@fake_test@@@ |
| @@@STEP_CURSOR@fake_test@@@ |
| @@@STEP_STARTED@@@ |
| @@@SET_BUILD_PROPERTY@fake_property@"value"@@@ |
| @@@STEP_LOG_LINE@run_recipe@fake_step_log_line@@@ |
| @@@STEP_LOG_END@run_recipe@@@ |
| @@@STEP_LOG_END@memory_profile@@@ |
| @@@STEP_LINK@shard #0 test results@https://fake-url.com@@@ |
| @@@STEP_LINK@shard #0 (runtime (13m 0s) + overhead (29s): 13m 29s)@https://faker-link.com@@@ |
| @@@STEP_CLOSED@@@""" |
| for line in fake_output.split('\n'): |
| adapter.ProcessLine(line) |
| self.assertEqual( |
| info_log.output, |
| [ |
| 'DEBUG:basic_logger:Test results for fake_test shard #0: ' |
| 'https://fake-url.com' |
| ], |
| ) |
| |
| def testInfoStepLevelOutput(self): |
| adapter = output_adapter.LegacyOutputAdapter() |
| with self.assertLogs('default_logger', level=logging.INFO) as root_log: |
| with self.assertLogs('basic_logger', level=logging.INFO) as info_log: |
| fake_output = """@@@SEED_STEP@generate_build_files@@@ |
| @@@STEP_CURSOR@generate_build_files@@@ |
| @@@STEP_STARTED@@@ |
| @@@SET_BUILD_PROPERTY@fake_property@"value"@@@ |
| @@@STEP_LOG_LINE@run_recipe@fake_step_log_line@@@ |
| @@@STEP_LOG_END@run_recipe@@@ |
| @@@STEP_LOG_END@memory_profile@@@ |
| fake info level std_out text |
| @@@STEP_CLOSED@@@""" |
| for line in fake_output.split('\n'): |
| adapter.ProcessLine(line) |
| self.assertEqual( |
| root_log.output, |
| [ |
| 'INFO:default_logger:\n[cyan]Running: generate_build_files[/]', |
| ], |
| ) |
| self.assertEqual( |
| info_log.output, ['INFO:basic_logger:fake info level std_out text'] |
| ) |
| |
| def testCompileStepOutput(self): |
| adapter = output_adapter.LegacyOutputAdapter() |
| with self.assertLogs('default_logger', level=logging.DEBUG) as root_log: |
| with self.assertLogs('basic_logger', level=logging.DEBUG) as info_log: |
| with self.assertLogs('single_line_logger') as compile_steps_log: |
| fake_output = """@@@SEED_STEP@compile@@@ |
| @@@STEP_CURSOR@compile@@@ |
| @@@STEP_STARTED@@@ |
| @@@SET_BUILD_PROPERTY@fake_property@"value"@@@ |
| @@@STEP_LOG_LINE@run_recipe@fake_step_log_line@@@ |
| @@@STEP_LOG_END@run_recipe@@@ |
| @@@STEP_LOG_END@memory_profile@@@ |
| Proxy started successfully. |
| ninja: Entering directory fake/out |
| [1/2] ACTION fake_action |
| [2/2] ACTION fake_action |
| RBE Stats: down 0 B, up 0 B, |
| @@@STEP_CLOSED@@@""" |
| for line in fake_output.split('\n'): |
| adapter.ProcessLine(line) |
| self.assertEqual( |
| root_log.output, |
| [ |
| 'INFO:default_logger:\n[cyan]Running: compile[/]', |
| ], |
| ) |
| self.assertEqual( |
| info_log.output, |
| [ |
| 'INFO:basic_logger:Proxy started successfully.', |
| 'INFO:basic_logger:ninja: Entering directory fake/out', |
| 'INFO:basic_logger:', |
| 'INFO:basic_logger:RBE Stats: down 0 B, up 0 B,', |
| ], |
| ) |
| # The ninja statuses are sent to another logger to remove new lines |
| self.assertEqual( |
| compile_steps_log.output, |
| [ |
| 'INFO:single_line_logger:\x1b[2K', |
| 'INFO:single_line_logger:\r[1/2] ACTION fake_action', |
| 'INFO:single_line_logger:\x1b[2K', |
| 'INFO:single_line_logger:\r[2/2] ACTION fake_action', |
| ], |
| ) |
| |
| def testDownloadOutputs(self): |
| if sys.version_info[:2] < (3, 10): |
| # 'assertNoLogs' was added in 3.10. Seems difficult to assert on no |
| # logging without it, so skip this test if it's not available. |
| # TODO(crbug.com/40942322): Remove the disable after Chromium's on 3.11. |
| return |
| # pylint: disable=no-member |
| adapter = output_adapter.LegacyOutputAdapter() |
| with self.assertLogs('default_logger', level=logging.DEBUG) as root_log: |
| with self.assertNoLogs('single_line_logger'): |
| fake_output = """@@@SEED_STEP@download compilation outputs@@@ |
| @@@SEED_STEP@download compilation outputs.read sql_unittests.runtime_deps@@@ |
| @@@STEP_CURSOR@download compilation outputs.read sql_unittests.runtime_deps@@@ |
| @@@STEP_STARTED@@@ |
| @@@STEP_NEST_LEVEL@1@@@ |
| @@@STEP_CLOSED@@@ |
| @@@SEED_STEP@download compilation outputs.fetch sql_unittests@@@ |
| @@@STEP_CURSOR@download compilation outputs.fetch sql_unittests@@@ |
| @@@STEP_STARTED@@@ |
| @@@STEP_NEST_LEVEL@1@@@ |
| file1 ...exists |
| file2 ...exists |
| @@@STEP_CLOSED@@@ |
| @@@STEP_CURSOR@download compilation outputs@@@ |
| @@@STEP_STARTED@@@ |
| @@@STEP_CLOSED@@@""" |
| for line in fake_output.split('\n'): |
| adapter.ProcessLine(line) |
| # Should just print the step name but none of the step's stdout. |
| nest_name = 'download compilation outputs' |
| self.assertEqual( |
| root_log.output, |
| [ |
| f'INFO:default_logger:\n[cyan]Running: {nest_name}[/]', |
| f'INFO:default_logger:\n[cyan]Running: {nest_name}.read sql_unittests.' |
| 'runtime_deps[/]', |
| f'INFO:default_logger:\n[cyan]Running: {nest_name}.fetch sql_unittests[/]', |
| ], |
| ) |
| |
| def testEnsureFailurePrinted(self): |
| adapter = output_adapter.LegacyOutputAdapter() |
| with self.assertLogs('default_logger', level=logging.DEBUG) as root_log: |
| with self.assertLogs('basic_logger', level=logging.DEBUG) as info_log: |
| fake_output = """@@@SEED_STEP@download compilation outputs@@@ |
| @@@STEP_CURSOR@download compilation outputs@@@ |
| @@@STEP_STARTED@@@ |
| @@@SET_BUILD_PROPERTY@fake_property@"value"@@@ |
| @@@STEP_LOG_LINE@run_recipe@fake_step_log_line@@@ |
| @@@STEP_LOG_END@run_recipe@@@ |
| @@@STEP_LOG_END@memory_profile@@@ |
| fake std_out text |
| @@@STEP_LOG_LINE@utr_log@utr-specific log line@@@ |
| @@@STEP_LOG_END@utr_log@@@ |
| @@@STEP_CLOSED@@@""" |
| for line in fake_output.split('\n'): |
| adapter.ProcessLine(line) |
| adapter.EnsureFailurePrinted() |
| self.assertEqual( |
| root_log.output, |
| [ |
| 'INFO:default_logger:\n[cyan]Running: download compilation outputs[/]' |
| ], |
| ) |
| self.assertEqual( |
| info_log.output, |
| ['INFO:basic_logger:fake std_out text\nutr-specific log line'], |
| ) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
