#!/usr/bin/env vpython3 # Copyright 2021 The Chromium Authors # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. """Unittests for test_result_util.py.""" import collections import copy import mock import unittest import test_result_util from test_result_util import TestResult, TestStatus, ResultCollection import test_runner_test FAKE_TEST_LOC = {'repo': 'https://test', 'fileName': '//test.cc'} PASSED_RESULT = TestResult( 'passed/test', TestStatus.PASS, duration=1233, test_log='Logs' ) PASSED_RESULT_WITH_LOC = TestResult( 'passed/test', TestStatus.PASS, duration=1233, test_log='Logs', test_loc=FAKE_TEST_LOC, ) FAILED_RESULT = TestResult( 'failed/test', TestStatus.FAIL, duration=1233, test_log='line1\nline2' ) FAILED_RESULT_DUPLICATE = TestResult( 'failed/test', TestStatus.FAIL, test_log='line3\nline4' ) DISABLED_RESULT = TestResult( 'disabled/test', TestStatus.SKIP, expected_status=TestStatus.SKIP, attachments={'name': '/path/to/name'}, ) UNEXPECTED_SKIPPED_RESULT = TestResult( 'unexpected/skipped_test', TestStatus.SKIP ) CRASHED_RESULT = TestResult('crashed/test', TestStatus.CRASH) FLAKY_PASS_RESULT = TestResult('flaky/test', TestStatus.PASS) FLAKY_FAIL_RESULT = TestResult( 'flaky/test', TestStatus.FAIL, test_log='line1\nline2' ) ABORTED_RESULT = TestResult('aborted/test', TestStatus.ABORT) class UtilTest(test_runner_test.TestCase): """Tests util methods in test_result_util module.""" def test_validate_kwargs(self): """Tests _validate_kwargs.""" with self.assertRaises(AssertionError) as context: TestResult('name', TestStatus.PASS, unknown='foo') expected_message = "Invalid keyword argument(s) in" self.assertTrue(expected_message in str(context.exception)) with self.assertRaises(AssertionError) as context: ResultCollection(test_log='foo') expected_message = "Invalid keyword argument(s) in" self.assertTrue(expected_message in str(context.exception)) def test_validate_test_status(self): """Tests exception raised from validation.""" with self.assertRaises(TypeError) as context: test_result_util._validate_test_status('TIMEOUT') expected_message = 'Invalid test status: TIMEOUT. Should be one of' self.assertTrue(expected_message in str(context.exception)) def test_to_standard_json_literal(self): """Tests _to_standard_json_literal.""" status = test_result_util._to_standard_json_literal(TestStatus.FAIL) self.assertEqual(status, 'FAIL') status = test_result_util._to_standard_json_literal(TestStatus.ABORT) self.assertEqual(status, 'TIMEOUT') class TestResultTest(test_runner_test.TestCase): """Tests TestResult class APIs.""" def test_init(self): """Tests class initialization.""" test_result = PASSED_RESULT self.assertEqual(test_result.name, 'passed/test') self.assertEqual(test_result.status, TestStatus.PASS) self.assertEqual(test_result.expected_status, TestStatus.PASS) self.assertEqual(test_result.test_log, 'Logs') def test_compose_result_sink_tags(self): """Tests _compose_result_sink_tags.""" disabled_test_tags = [ ('test_name', 'disabled/test'), ('disabled_test', 'true'), ] unexpected_skip_test_tags = [ ('test_name', 'unexpected/skipped_test'), ('disabled_test', 'false'), ] not_skip_test_tags = [('test_name', 'passed/test')] not_skip_test_result = PASSED_RESULT self.assertEqual( not_skip_test_tags, not_skip_test_result._compose_result_sink_tags() ) disabled_test_result = DISABLED_RESULT self.assertEqual( disabled_test_tags, disabled_test_result._compose_result_sink_tags() ) unexpected_skip_test_result = UNEXPECTED_SKIPPED_RESULT self.assertEqual( unexpected_skip_test_tags, unexpected_skip_test_result._compose_result_sink_tags(), ) @mock.patch('result_sink_util.ResultSinkClient.post') def test_report_to_result_sink(self, mock_post): disabled_test_result = DISABLED_RESULT client = mock.MagicMock() disabled_test_result.report_to_result_sink(client) client.post.assert_called_with( 'disabled/test', 'SKIP', True, duration=None, test_log='', tags=[('test_name', 'disabled/test'), ('disabled_test', 'true')], test_loc=None, file_artifacts={'name': '/path/to/name'}, ) # Duplicate calls will only report once. disabled_test_result.report_to_result_sink(client) self.assertEqual(client.post.call_count, 1) disabled_test_result.report_to_result_sink(client) self.assertEqual(client.post.call_count, 1) faileded_result = FAILED_RESULT client = mock.MagicMock() faileded_result.report_to_result_sink(client) client.post.assert_called_with( 'failed/test', 'FAIL', False, duration=1233, file_artifacts={}, tags=[('test_name', 'failed/test')], test_loc=None, test_log='line1\nline2', ) passed_result = PASSED_RESULT_WITH_LOC client = mock.MagicMock() passed_result.report_to_result_sink(client) client.post.assert_called_with( 'passed/test', 'PASS', True, duration=1233, file_artifacts={}, tags=[('test_name', 'passed/test')], test_loc=FAKE_TEST_LOC, test_log='Logs', ) class ResultCollectionTest(test_runner_test.TestCase): """Tests ResultCollection class APIs.""" def setUp(self): super(ResultCollectionTest, self).setUp() self.full_collection = ResultCollection( test_results=[ PASSED_RESULT, FAILED_RESULT, FAILED_RESULT_DUPLICATE, DISABLED_RESULT, UNEXPECTED_SKIPPED_RESULT, CRASHED_RESULT, FLAKY_PASS_RESULT, FLAKY_FAIL_RESULT, ABORTED_RESULT, ] ) def test_init(self): """Tests class initialization.""" collection = ResultCollection( test_results=[PASSED_RESULT, DISABLED_RESULT, UNEXPECTED_SKIPPED_RESULT], crashed=True, ) self.assertTrue(collection.crashed) self.assertEqual(collection.crash_message, '') self.assertEqual( collection.test_results, [PASSED_RESULT, DISABLED_RESULT, UNEXPECTED_SKIPPED_RESULT], ) def test_add_result(self): """Tests add_test_result.""" collection = ResultCollection(test_results=[FAILED_RESULT]) collection.add_test_result(DISABLED_RESULT) self.assertEqual(collection.test_results, [FAILED_RESULT, DISABLED_RESULT]) def test_add_result_collection_default(self): """Tests add_result_collection default (merge crash info).""" collection = ResultCollection(test_results=[FAILED_RESULT]) self.assertFalse(collection.crashed) collection.append_crash_message('Crash1') crashed_collection = ResultCollection( test_results=[PASSED_RESULT], crashed=True ) crashed_collection.append_crash_message('Crash2') collection.add_result_collection(crashed_collection) self.assertTrue(collection.crashed) self.assertEqual(collection.crash_message, 'Crash1\nCrash2') self.assertEqual(collection.test_results, [FAILED_RESULT, PASSED_RESULT]) def test_add_result_collection_overwrite(self): """Tests add_result_collection overwrite.""" collection = ResultCollection(test_results=[FAILED_RESULT], crashed=True) self.assertTrue(collection.crashed) collection.append_crash_message('Crash1') crashed_collection = ResultCollection(test_results=[PASSED_RESULT]) collection.add_result_collection(crashed_collection, overwrite_crash=True) self.assertFalse(collection.crashed) self.assertEqual(collection.crash_message, '') self.assertEqual(collection.test_results, [FAILED_RESULT, PASSED_RESULT]) def test_add_result_collection_ignore(self): """Tests add_result_collection overwrite.""" collection = ResultCollection(test_results=[FAILED_RESULT]) self.assertFalse(collection.crashed) crashed_collection = ResultCollection( test_results=[PASSED_RESULT], crashed=True ) crashed_collection.append_crash_message('Crash2') collection.add_result_collection(crashed_collection, ignore_crash=True) self.assertFalse(collection.crashed) self.assertEqual(collection.crash_message, '') self.assertEqual(collection.test_results, [FAILED_RESULT, PASSED_RESULT]) def test_add_results(self): """Tests add_results.""" collection = ResultCollection(test_results=[PASSED_RESULT]) collection.add_results([FAILED_RESULT, DISABLED_RESULT]) self.assertEqual( collection.test_results, [PASSED_RESULT, FAILED_RESULT, DISABLED_RESULT] ) def test_add_name_prefix_to_tests(self): """Tests add_name_prefix_to_tests.""" passed = copy.copy(PASSED_RESULT) disabeld = copy.copy(DISABLED_RESULT) collection = ResultCollection(test_results=[passed, disabeld]) some_prefix = 'Some/prefix' collection.add_name_prefix_to_tests(some_prefix) for test_result in collection.test_results: self.assertTrue(test_result.name.startswith(some_prefix)) def test_add_test_names_status(self): """Tests add_test_names_status.""" test_names = ['test1', 'test2', 'test3'] collection = ResultCollection(test_results=[PASSED_RESULT]) collection.add_test_names_status(test_names, TestStatus.SKIP) disabled_test_names = ['test4', 'test5', 'test6'] collection.add_test_names_status( disabled_test_names, TestStatus.SKIP, expected_status=TestStatus.SKIP ) self.assertEqual(collection.test_results[0], PASSED_RESULT) unexpected_skipped = collection.tests_by_expression( lambda t: not t.expected() and t.status == TestStatus.SKIP ) self.assertEqual(unexpected_skipped, set(['test1', 'test2', 'test3'])) self.assertEqual( collection.disabled_tests(), set(['test4', 'test5', 'test6']) ) @mock.patch('test_result_util.TestResult.report_to_result_sink') @mock.patch('result_sink_util.ResultSinkClient.close') @mock.patch('result_sink_util.ResultSinkClient.__init__', return_value=None) def test_add_and_report_test_names_status( self, mock_sink_init, mock_sink_close, mock_report ): """Tests add_test_names_status.""" test_names = ['test1', 'test2', 'test3'] collection = ResultCollection(test_results=[PASSED_RESULT]) collection.add_and_report_test_names_status(test_names, TestStatus.SKIP) self.assertEqual(collection.test_results[0], PASSED_RESULT) unexpected_skipped = collection.tests_by_expression( lambda t: not t.expected() and t.status == TestStatus.SKIP ) self.assertEqual(unexpected_skipped, set(['test1', 'test2', 'test3'])) self.assertEqual(1, len(mock_sink_init.mock_calls)) self.assertEqual(3, len(mock_report.mock_calls)) self.assertEqual(1, len(mock_sink_close.mock_calls)) def testappend_crash_message(self): """Tests append_crash_message.""" collection = ResultCollection(test_results=[PASSED_RESULT]) collection.append_crash_message('Crash message 1.') self.assertEqual(collection.crash_message, 'Crash message 1.') collection.append_crash_message('Crash message 2.') self.assertEqual( collection.crash_message, 'Crash message 1.\nCrash message 2.' ) def test_tests_by_expression(self): """Tests tests_by_expression.""" collection = self.full_collection exp = lambda result: result.status == TestStatus.SKIP skipped_tests = collection.tests_by_expression(exp) self.assertEqual( skipped_tests, set(['unexpected/skipped_test', 'disabled/test']) ) def test_get_spcific_tests(self): """Tests getting sets of tests of specific status.""" collection = self.full_collection self.assertEqual( collection.all_test_names(), set( [ 'passed/test', 'disabled/test', 'failed/test', 'unexpected/skipped_test', 'crashed/test', 'flaky/test', 'aborted/test', ] ), ) self.assertEqual(collection.crashed_tests(), set(['crashed/test'])) self.assertEqual(collection.disabled_tests(), set(['disabled/test'])) self.assertEqual( collection.expected_tests(), set(['passed/test', 'disabled/test', 'flaky/test']), ) self.assertEqual( collection.unexpected_tests(), set( [ 'failed/test', 'unexpected/skipped_test', 'crashed/test', 'flaky/test', 'aborted/test', ] ), ) self.assertEqual( collection.passed_tests(), set(['passed/test', 'flaky/test']) ) self.assertEqual( collection.failed_tests(), set(['failed/test', 'flaky/test']) ) self.assertEqual(collection.flaky_tests(), set(['flaky/test'])) self.assertEqual( collection.never_expected_tests(), set( [ 'failed/test', 'unexpected/skipped_test', 'crashed/test', 'aborted/test', ] ), ) self.assertEqual( collection.pure_expected_tests(), set(['passed/test', 'disabled/test']) ) def test_add_and_report_crash(self): """Tests add_and_report_crash.""" collection = copy.copy(self.full_collection) collection.set_crashed_with_prefix('Prefix Line') self.assertEqual(collection.crash_message, 'Prefix Line\n') self.assertTrue(collection.crashed) @mock.patch('test_result_util.TestResult.report_to_result_sink') @mock.patch('result_sink_util.ResultSinkClient.close') @mock.patch('result_sink_util.ResultSinkClient.__init__', return_value=None) def test_report_to_result_sink( self, mock_sink_init, mock_sink_close, mock_report ): """Tests report_to_result_sink.""" collection = copy.copy(self.full_collection) collection.report_to_result_sink() mock_sink_init.assert_called_once() self.assertEqual(len(collection.test_results), len(mock_report.mock_calls)) mock_sink_close.assert_called() @mock.patch('shard_util.gtest_shard_index', return_value=0) @mock.patch('time.time', return_value=10000) def test_standard_json_output(self, *args): """Tests standard_json_output.""" passed_test_value = { 'expected': 'PASS', 'actual': 'PASS', 'shard': 0, 'is_unexpected': False, } failed_test_value = { 'expected': 'PASS', 'actual': 'FAIL FAIL', 'shard': 0, 'is_unexpected': True, } disabled_test_value = { 'expected': 'SKIP', 'actual': 'SKIP', 'shard': 0, 'is_unexpected': False, } unexpected_skip_test_value = { 'expected': 'PASS', 'actual': 'SKIP', 'shard': 0, 'is_unexpected': True, } crashed_test_value = { 'expected': 'PASS', 'actual': 'CRASH', 'shard': 0, 'is_unexpected': True, } flaky_test_value = { 'expected': 'PASS', 'actual': 'PASS FAIL', 'shard': 0, 'is_unexpected': False, 'is_flaky': True, } aborted_test_value = { 'expected': 'PASS', 'actual': 'TIMEOUT', 'shard': 0, 'is_unexpected': True, } expected_tests = collections.OrderedDict() expected_tests['passed/test'] = passed_test_value expected_tests['failed/test'] = failed_test_value expected_tests['disabled/test'] = disabled_test_value expected_tests['unexpected/skipped_test'] = unexpected_skip_test_value expected_tests['crashed/test'] = crashed_test_value expected_tests['flaky/test'] = flaky_test_value expected_tests['aborted/test'] = aborted_test_value expected_num_failures_by_type = { 'PASS': 2, 'FAIL': 1, 'CRASH': 1, 'SKIP': 2, 'TIMEOUT': 1, } expected_json = { 'version': 3, 'path_delimiter': '/', 'seconds_since_epoch': 10000, 'interrupted': False, 'num_failures_by_type': expected_num_failures_by_type, 'tests': expected_tests, } self.assertEqual( self.full_collection.standard_json_output(path_delimiter='/'), expected_json, ) def test_test_runner_logs(self): """Test test_runner_logs.""" expected_logs = collections.OrderedDict() expected_logs['passed tests'] = ['passed/test'] expected_logs['disabled tests'] = ['disabled/test'] flaky_logs = ['Failure log of attempt 1:', 'line1', 'line2'] failed_logs = [ 'Failure log of attempt 1:', 'line1', 'line2', 'Failure log of attempt 2:', 'line3', 'line4', ] no_logs = ['Failure log of attempt 1:', ''] expected_logs['flaked tests'] = {'flaky/test': flaky_logs} expected_logs['failed tests'] = { 'failed/test': failed_logs, 'crashed/test': no_logs, 'unexpected/skipped_test': no_logs, 'aborted/test': no_logs, } expected_logs['failed/test'] = failed_logs expected_logs['unexpected/skipped_test'] = no_logs expected_logs['flaky/test'] = flaky_logs expected_logs['crashed/test'] = no_logs expected_logs['aborted/test'] = no_logs generated_logs = self.full_collection.test_runner_logs() keys = [ 'passed tests', 'disabled tests', 'flaked tests', 'failed tests', 'failed/test', 'unexpected/skipped_test', 'flaky/test', 'crashed/test', 'aborted/test', ] for key in keys: self.assertEqual(generated_logs[key], expected_logs[key]) if __name__ == '__main__': unittest.main()