#!/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()
Codebase Browser
chromium