1#!/usr/bin/env python3 2# 3# Copyright 2019, The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16 17"""Unittest for atest_execution_info.""" 18 19 20import os 21import pathlib 22import time 23import unittest 24from unittest.mock import patch 25from atest import arg_parser 26from atest import atest_execution_info as aei 27from atest import constants 28from atest import result_reporter 29from atest.metrics import metrics 30from atest.test_runners import test_runner_base 31from pyfakefs import fake_filesystem_unittest 32 33RESULT_TEST_TEMPLATE = test_runner_base.TestResult( 34 runner_name='someRunner', 35 group_name='someModule', 36 test_name='someClassName#sostName', 37 status=test_runner_base.PASSED_STATUS, 38 details=None, 39 test_count=1, 40 test_time='(10ms)', 41 runner_total=None, 42 group_total=2, 43 additional_info={}, 44 test_run_name='com.android.UnitTests', 45) 46 47 48class CopyBuildTraceToLogsTests(fake_filesystem_unittest.TestCase): 49 50 def setUp(self): 51 super().setUp() 52 self.setUpPyfakefs() 53 self.fs.create_dir(constants.ATEST_RESULT_ROOT) 54 55 def test_copy_build_artifacts_to_log_dir_new_trace_copy(self): 56 start_time = 10 57 log_path = pathlib.Path('/logs') 58 self.fs.create_dir(log_path) 59 out_path = pathlib.Path('/out') 60 build_trace_path = out_path / 'build.trace' 61 self.fs.create_file(build_trace_path) 62 # Set the trace file's mtime greater than start time 63 os.utime(build_trace_path, (20, 20)) 64 end_time = 30 65 66 aei.AtestExecutionInfo._copy_build_artifacts_to_log_dir( 67 start_time, end_time, out_path, log_path, 'build.trace' 68 ) 69 70 self.assertTrue( 71 self._is_dir_contains_files_with_prefix(log_path, 'build.trace') 72 ) 73 74 def test_copy_build_artifacts_to_log_dir_old_trace_does_not_copy(self): 75 start_time = 10 76 log_path = pathlib.Path('/logs') 77 self.fs.create_dir(log_path) 78 out_path = pathlib.Path('/out') 79 build_trace_path = out_path / 'build.trace' 80 self.fs.create_file(build_trace_path) 81 # Set the trace file's mtime smaller than start time 82 os.utime(build_trace_path, (5, 5)) 83 end_time = 30 84 85 aei.AtestExecutionInfo._copy_build_artifacts_to_log_dir( 86 start_time, end_time, out_path, log_path, 'build.trace' 87 ) 88 89 self.assertFalse( 90 self._is_dir_contains_files_with_prefix(log_path, 'build.trace') 91 ) 92 93 def test_copy_multiple_build_trace_to_log_dir(self): 94 start_time = 10 95 log_path = pathlib.Path('/logs') 96 self.fs.create_dir(log_path) 97 out_path = pathlib.Path('/out') 98 build_trace_path1 = out_path / 'build.trace.1.gz' 99 build_trace_path2 = out_path / 'build.trace.2.gz' 100 self.fs.create_file(build_trace_path1) 101 self.fs.create_file(build_trace_path2) 102 # Set the trace file's mtime greater than start time 103 os.utime(build_trace_path1, (20, 20)) 104 os.utime(build_trace_path2, (20, 20)) 105 end_time = 30 106 107 aei.AtestExecutionInfo._copy_build_artifacts_to_log_dir( 108 start_time, end_time, out_path, log_path, 'build.trace' 109 ) 110 111 self.assertTrue( 112 self._is_dir_contains_files_with_prefix(log_path, 'build.trace.1.gz') 113 ) 114 self.assertTrue( 115 self._is_dir_contains_files_with_prefix(log_path, 'build.trace.2.gz') 116 ) 117 118 def _is_dir_contains_files_with_prefix( 119 self, dir: pathlib.Path, prefix: str 120 ) -> bool: 121 for file in dir.iterdir(): 122 if file.is_file() and file.name.startswith(prefix): 123 return True 124 return False 125 126 127# pylint: disable=protected-access 128class AtestExecutionInfoUnittests(unittest.TestCase): 129 """Unit tests for atest_execution_info.py""" 130 131 @patch('atest.metrics.metrics.is_internal_user', return_value=False) 132 def test_create_bug_report_url_is_external_user_return_empty(self, _): 133 url = aei.AtestExecutionInfo._create_bug_report_url() 134 135 self.assertFalse(url) 136 137 @patch('atest.metrics.metrics.is_internal_user', return_value=True) 138 def test_create_bug_report_url_is_internal_user_return_url(self, _): 139 url = aei.AtestExecutionInfo._create_bug_report_url() 140 141 self.assertTrue(url) 142 143 @patch('atest.metrics.metrics.is_internal_user', return_value=True) 144 @patch('atest.logstorage.log_uploader.is_uploading_logs', return_value=True) 145 def test_create_bug_report_url_is_uploading_logs_use_contains_run_id( 146 self, _, __ 147 ): 148 url = aei.AtestExecutionInfo._create_bug_report_url() 149 150 self.assertIn(metrics.get_run_id(), url) 151 152 @patch('atest.metrics.metrics.is_internal_user', return_value=True) 153 @patch('atest.logstorage.log_uploader.is_uploading_logs', return_value=False) 154 def test_create_bug_report_url_is_not_uploading_logs_use_contains_run_id( 155 self, _, __ 156 ): 157 url = aei.AtestExecutionInfo._create_bug_report_url() 158 159 self.assertNotIn(metrics.get_run_id(), url) 160 161 def test_arrange_test_result_one_module(self): 162 """Test _arrange_test_result method with only one module.""" 163 pass_1 = self._create_test_result(status=test_runner_base.PASSED_STATUS) 164 pass_2 = self._create_test_result(status=test_runner_base.PASSED_STATUS) 165 pass_3 = self._create_test_result(status=test_runner_base.PASSED_STATUS) 166 fail_1 = self._create_test_result(status=test_runner_base.FAILED_STATUS) 167 fail_2 = self._create_test_result(status=test_runner_base.FAILED_STATUS) 168 ignore_1 = self._create_test_result(status=test_runner_base.IGNORED_STATUS) 169 reporter_1 = result_reporter.ResultReporter() 170 reporter_1.all_test_results.extend([pass_1, pass_2, pass_3]) 171 reporter_2 = result_reporter.ResultReporter() 172 reporter_2.all_test_results.extend([fail_1, fail_2, ignore_1]) 173 info_dict = {} 174 aei.AtestExecutionInfo._arrange_test_result( 175 info_dict, [reporter_1, reporter_2] 176 ) 177 expect_summary = { 178 aei._STATUS_IGNORED_KEY: 1, 179 aei._STATUS_FAILED_KEY: 2, 180 aei._STATUS_PASSED_KEY: 3, 181 } 182 self.assertEqual(expect_summary, info_dict[aei._TOTAL_SUMMARY_KEY]) 183 184 def test_arrange_test_result_multi_module(self): 185 """Test _arrange_test_result method with multi module.""" 186 group_a_pass_1 = self._create_test_result( 187 group_name='grpup_a', status=test_runner_base.PASSED_STATUS 188 ) 189 group_b_pass_1 = self._create_test_result( 190 group_name='grpup_b', status=test_runner_base.PASSED_STATUS 191 ) 192 group_c_pass_1 = self._create_test_result( 193 group_name='grpup_c', status=test_runner_base.PASSED_STATUS 194 ) 195 group_b_fail_1 = self._create_test_result( 196 group_name='grpup_b', status=test_runner_base.FAILED_STATUS 197 ) 198 group_c_fail_1 = self._create_test_result( 199 group_name='grpup_c', status=test_runner_base.FAILED_STATUS 200 ) 201 group_c_ignore_1 = self._create_test_result( 202 group_name='grpup_c', status=test_runner_base.IGNORED_STATUS 203 ) 204 reporter_1 = result_reporter.ResultReporter() 205 reporter_1.all_test_results.extend( 206 [group_a_pass_1, group_b_pass_1, group_c_pass_1] 207 ) 208 reporter_2 = result_reporter.ResultReporter() 209 reporter_2.all_test_results.extend( 210 [group_b_fail_1, group_c_fail_1, group_c_ignore_1] 211 ) 212 213 info_dict = {} 214 aei.AtestExecutionInfo._arrange_test_result( 215 info_dict, [reporter_1, reporter_2] 216 ) 217 expect_group_a_summary = { 218 aei._STATUS_IGNORED_KEY: 0, 219 aei._STATUS_FAILED_KEY: 0, 220 aei._STATUS_PASSED_KEY: 1, 221 } 222 self.assertEqual( 223 expect_group_a_summary, 224 info_dict[aei._TEST_RUNNER_KEY]['someRunner']['grpup_a'][ 225 aei._SUMMARY_KEY 226 ], 227 ) 228 229 expect_group_b_summary = { 230 aei._STATUS_IGNORED_KEY: 0, 231 aei._STATUS_FAILED_KEY: 1, 232 aei._STATUS_PASSED_KEY: 1, 233 } 234 self.assertEqual( 235 expect_group_b_summary, 236 info_dict[aei._TEST_RUNNER_KEY]['someRunner']['grpup_b'][ 237 aei._SUMMARY_KEY 238 ], 239 ) 240 241 expect_group_c_summary = { 242 aei._STATUS_IGNORED_KEY: 1, 243 aei._STATUS_FAILED_KEY: 1, 244 aei._STATUS_PASSED_KEY: 1, 245 } 246 self.assertEqual( 247 expect_group_c_summary, 248 info_dict[aei._TEST_RUNNER_KEY]['someRunner']['grpup_c'][ 249 aei._SUMMARY_KEY 250 ], 251 ) 252 253 expect_total_summary = { 254 aei._STATUS_IGNORED_KEY: 1, 255 aei._STATUS_FAILED_KEY: 2, 256 aei._STATUS_PASSED_KEY: 3, 257 } 258 self.assertEqual(expect_total_summary, info_dict[aei._TOTAL_SUMMARY_KEY]) 259 260 def test_preparation_time(self): 261 """Test preparation_time method.""" 262 start_time = time.time() 263 aei.PREPARE_END_TIME = None 264 self.assertTrue(aei.preparation_time(start_time) is None) 265 aei.PREPARE_END_TIME = time.time() 266 self.assertFalse(aei.preparation_time(start_time) is None) 267 268 def test_arrange_test_result_multi_runner(self): 269 """Test _arrange_test_result method with multi runner.""" 270 runner_a_pass_1 = self._create_test_result( 271 runner_name='runner_a', status=test_runner_base.PASSED_STATUS 272 ) 273 runner_a_pass_2 = self._create_test_result( 274 runner_name='runner_a', status=test_runner_base.PASSED_STATUS 275 ) 276 runner_a_pass_3 = self._create_test_result( 277 runner_name='runner_a', status=test_runner_base.PASSED_STATUS 278 ) 279 runner_b_fail_1 = self._create_test_result( 280 runner_name='runner_b', status=test_runner_base.FAILED_STATUS 281 ) 282 runner_b_fail_2 = self._create_test_result( 283 runner_name='runner_b', status=test_runner_base.FAILED_STATUS 284 ) 285 runner_b_ignore_1 = self._create_test_result( 286 runner_name='runner_b', status=test_runner_base.IGNORED_STATUS 287 ) 288 289 reporter_1 = result_reporter.ResultReporter() 290 reporter_1.all_test_results.extend( 291 [runner_a_pass_1, runner_a_pass_2, runner_a_pass_3] 292 ) 293 reporter_2 = result_reporter.ResultReporter() 294 reporter_2.all_test_results.extend( 295 [runner_b_fail_1, runner_b_fail_2, runner_b_ignore_1] 296 ) 297 info_dict = {} 298 aei.AtestExecutionInfo._arrange_test_result( 299 info_dict, [reporter_1, reporter_2] 300 ) 301 expect_group_a_summary = { 302 aei._STATUS_IGNORED_KEY: 0, 303 aei._STATUS_FAILED_KEY: 0, 304 aei._STATUS_PASSED_KEY: 3, 305 } 306 self.assertEqual( 307 expect_group_a_summary, 308 info_dict[aei._TEST_RUNNER_KEY]['runner_a']['someModule'][ 309 aei._SUMMARY_KEY 310 ], 311 ) 312 313 expect_group_b_summary = { 314 aei._STATUS_IGNORED_KEY: 1, 315 aei._STATUS_FAILED_KEY: 2, 316 aei._STATUS_PASSED_KEY: 0, 317 } 318 self.assertEqual( 319 expect_group_b_summary, 320 info_dict[aei._TEST_RUNNER_KEY]['runner_b']['someModule'][ 321 aei._SUMMARY_KEY 322 ], 323 ) 324 325 expect_total_summary = { 326 aei._STATUS_IGNORED_KEY: 1, 327 aei._STATUS_FAILED_KEY: 2, 328 aei._STATUS_PASSED_KEY: 3, 329 } 330 self.assertEqual(expect_total_summary, info_dict[aei._TOTAL_SUMMARY_KEY]) 331 332 def _create_test_result(self, **kwargs): 333 """A Helper to create TestResult""" 334 test_info = test_runner_base.TestResult(**RESULT_TEST_TEMPLATE._asdict()) 335 return test_info._replace(**kwargs) 336 337 338if __name__ == '__main__': 339 unittest.main() 340