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_enum 27from atest import atest_execution_info as aei 28from atest import constants 29from atest import result_reporter 30from atest.metrics import metrics 31from atest.test_runners import test_runner_base 32from pyfakefs import fake_filesystem_unittest 33 34RESULT_TEST_TEMPLATE = test_runner_base.TestResult( 35 runner_name='someRunner', 36 group_name='someModule', 37 test_name='someClassName#sostName', 38 status=test_runner_base.PASSED_STATUS, 39 details=None, 40 test_count=1, 41 test_time='(10ms)', 42 runner_total=None, 43 group_total=2, 44 additional_info={}, 45 test_run_name='com.android.UnitTests', 46) 47 48 49class CopyBuildTraceToLogsTests(fake_filesystem_unittest.TestCase): 50 51 def setUp(self): 52 super().setUp() 53 self.setUpPyfakefs() 54 self.fs.create_dir(constants.ATEST_RESULT_ROOT) 55 56 def test_copy_build_artifacts_to_log_dir_new_trace_copy(self): 57 start_time = 10 58 log_path = pathlib.Path('/logs') 59 self.fs.create_dir(log_path) 60 out_path = pathlib.Path('/out') 61 build_trace_path = out_path / 'build.trace' 62 self.fs.create_file(build_trace_path) 63 # Set the trace file's mtime greater than start time 64 os.utime(build_trace_path, (20, 20)) 65 end_time = 30 66 67 aei.AtestExecutionInfo._copy_build_artifacts_to_log_dir( 68 start_time, end_time, out_path, log_path, 'build.trace' 69 ) 70 71 self.assertTrue( 72 self._is_dir_contains_files_with_prefix(log_path, 'build.trace') 73 ) 74 75 def test_copy_build_artifacts_to_log_dir_old_trace_does_not_copy(self): 76 start_time = 10 77 log_path = pathlib.Path('/logs') 78 self.fs.create_dir(log_path) 79 out_path = pathlib.Path('/out') 80 build_trace_path = out_path / 'build.trace' 81 self.fs.create_file(build_trace_path) 82 # Set the trace file's mtime smaller than start time 83 os.utime(build_trace_path, (5, 5)) 84 end_time = 30 85 86 aei.AtestExecutionInfo._copy_build_artifacts_to_log_dir( 87 start_time, end_time, out_path, log_path, 'build.trace' 88 ) 89 90 self.assertFalse( 91 self._is_dir_contains_files_with_prefix(log_path, 'build.trace') 92 ) 93 94 def test_copy_multiple_build_trace_to_log_dir(self): 95 start_time = 10 96 log_path = pathlib.Path('/logs') 97 self.fs.create_dir(log_path) 98 out_path = pathlib.Path('/out') 99 build_trace_path1 = out_path / 'build.trace.1.gz' 100 build_trace_path2 = out_path / 'build.trace.2.gz' 101 self.fs.create_file(build_trace_path1) 102 self.fs.create_file(build_trace_path2) 103 # Set the trace file's mtime greater than start time 104 os.utime(build_trace_path1, (20, 20)) 105 os.utime(build_trace_path2, (20, 20)) 106 end_time = 30 107 108 aei.AtestExecutionInfo._copy_build_artifacts_to_log_dir( 109 start_time, end_time, out_path, log_path, 'build.trace' 110 ) 111 112 self.assertTrue( 113 self._is_dir_contains_files_with_prefix(log_path, 'build.trace.1.gz') 114 ) 115 self.assertTrue( 116 self._is_dir_contains_files_with_prefix(log_path, 'build.trace.2.gz') 117 ) 118 119 def _is_dir_contains_files_with_prefix( 120 self, dir: pathlib.Path, prefix: str 121 ) -> bool: 122 for file in dir.iterdir(): 123 if file.is_file() and file.name.startswith(prefix): 124 return True 125 return False 126 127 128class SendIncrementalSetupStatTests(fake_filesystem_unittest.TestCase): 129 130 _HOST_LOG_1_CONTENT = ( 131 '[ApkChangeDetector] Skipping the installation of SystemUIApp\nInstalling' 132 ' apk android.CtsApp\n[ApkChangeDetector] Skipping the uninstallation of' 133 ' SystemUIApp' 134 ) 135 136 _HOST_LOG_2_CONTENT = ( 137 '[ApkChangeDetector] Skipping the installation of SysUIRobolectricApp\n' 138 ' Installing apk a.b.c.d \n [UnrelatedClass] Skipping the installation' 139 ' of SomeClass' 140 ) 141 142 def setUp(self): 143 super().setUp() 144 self.setUpPyfakefs() 145 self.fs.create_dir(constants.ATEST_RESULT_ROOT) 146 self._log_path = pathlib.Path('/tmp') 147 148 def tearDown(self): 149 if os.path.exists(str(self._log_path)): 150 self.fs.remove_object(str(self._log_path)) 151 super().tearDown() 152 153 @patch('atest.metrics.metrics.LocalDetectEvent') 154 def test_parse_test_log_and_send_app_installation_stats_metrics_get_stats_successful( 155 self, 156 mock_detect_event, 157 ): 158 host_log_path1 = self._log_path / 'host_log_1.txt' 159 host_log_path2 = self._log_path / 'invocation' / 'host_log_2.txt' 160 self.fs.create_file( 161 host_log_path1, 162 contents=self.__class__._HOST_LOG_1_CONTENT, 163 create_missing_dirs=True, 164 ) 165 self.fs.create_file( 166 host_log_path2, 167 contents=self.__class__._HOST_LOG_2_CONTENT, 168 create_missing_dirs=True, 169 ) 170 expected_calls = [ 171 unittest.mock.call( 172 detect_type=atest_enum.DetectType.APP_INSTALLATION_SKIPPED_COUNT, 173 result=2, 174 ), 175 unittest.mock.call( 176 detect_type=atest_enum.DetectType.APP_INSTALLATION_NOT_SKIPPED_COUNT, 177 result=2, 178 ), 179 ] 180 181 aei.parse_test_log_and_send_app_installation_stats_metrics(self._log_path) 182 183 mock_detect_event.assert_has_calls(expected_calls, any_order=True) 184 185 @patch('atest.metrics.metrics.LocalDetectEvent') 186 def test_parse_test_log_and_send_app_installation_stats_metrics_no_host_log( 187 self, 188 mock_detect_event, 189 ): 190 aei.parse_test_log_and_send_app_installation_stats_metrics(self._log_path) 191 192 mock_detect_event.assert_not_called() 193 194 @patch('atest.metrics.metrics.LocalDetectEvent') 195 def test_parse_test_log_and_send_app_installation_stats_metrics_no_info_in_host_log( 196 self, 197 mock_detect_event, 198 ): 199 host_log_path1 = self._log_path / 'host_log_1.txt' 200 host_log_path2 = self._log_path / 'invocation' / 'host_log_2.txt' 201 self.fs.create_file(host_log_path1, contents='', create_missing_dirs=True) 202 self.fs.create_file(host_log_path2, contents='', create_missing_dirs=True) 203 expected_calls = [ 204 unittest.mock.call( 205 detect_type=atest_enum.DetectType.APP_INSTALLATION_SKIPPED_COUNT, 206 result=0, 207 ), 208 unittest.mock.call( 209 detect_type=atest_enum.DetectType.APP_INSTALLATION_NOT_SKIPPED_COUNT, 210 result=0, 211 ), 212 ] 213 214 aei.parse_test_log_and_send_app_installation_stats_metrics(self._log_path) 215 216 mock_detect_event.assert_has_calls(expected_calls, any_order=True) 217 218 219# pylint: disable=protected-access 220class AtestExecutionInfoUnittests(unittest.TestCase): 221 """Unit tests for atest_execution_info.py""" 222 223 @patch('atest.metrics.metrics.is_internal_user', return_value=False) 224 def test_create_bug_report_url_is_external_user_return_empty(self, _): 225 url = aei.AtestExecutionInfo._create_bug_report_url() 226 227 self.assertFalse(url) 228 229 @patch('atest.metrics.metrics.is_internal_user', return_value=True) 230 def test_create_bug_report_url_is_internal_user_return_url(self, _): 231 url = aei.AtestExecutionInfo._create_bug_report_url() 232 233 self.assertTrue(url) 234 235 @patch('atest.metrics.metrics.is_internal_user', return_value=True) 236 @patch('atest.logstorage.log_uploader.is_uploading_logs', return_value=True) 237 def test_create_bug_report_url_is_uploading_logs_use_contains_run_id( 238 self, _, __ 239 ): 240 url = aei.AtestExecutionInfo._create_bug_report_url() 241 242 self.assertIn(metrics.get_run_id(), url) 243 244 @patch('atest.metrics.metrics.is_internal_user', return_value=True) 245 @patch('atest.logstorage.log_uploader.is_uploading_logs', return_value=False) 246 def test_create_bug_report_url_is_not_uploading_logs_use_contains_run_id( 247 self, _, __ 248 ): 249 url = aei.AtestExecutionInfo._create_bug_report_url() 250 251 self.assertNotIn(metrics.get_run_id(), url) 252 253 def test_arrange_test_result_one_module(self): 254 """Test _arrange_test_result method with only one module.""" 255 pass_1 = self._create_test_result(status=test_runner_base.PASSED_STATUS) 256 pass_2 = self._create_test_result(status=test_runner_base.PASSED_STATUS) 257 pass_3 = self._create_test_result(status=test_runner_base.PASSED_STATUS) 258 fail_1 = self._create_test_result(status=test_runner_base.FAILED_STATUS) 259 fail_2 = self._create_test_result(status=test_runner_base.FAILED_STATUS) 260 ignore_1 = self._create_test_result(status=test_runner_base.IGNORED_STATUS) 261 reporter_1 = result_reporter.ResultReporter() 262 reporter_1.all_test_results.extend([pass_1, pass_2, pass_3]) 263 reporter_2 = result_reporter.ResultReporter() 264 reporter_2.all_test_results.extend([fail_1, fail_2, ignore_1]) 265 info_dict = {} 266 aei.AtestExecutionInfo._arrange_test_result( 267 info_dict, [reporter_1, reporter_2] 268 ) 269 expect_summary = { 270 aei._STATUS_IGNORED_KEY: 1, 271 aei._STATUS_FAILED_KEY: 2, 272 aei._STATUS_PASSED_KEY: 3, 273 } 274 self.assertEqual(expect_summary, info_dict[aei._TOTAL_SUMMARY_KEY]) 275 276 def test_arrange_test_result_multi_module(self): 277 """Test _arrange_test_result method with multi module.""" 278 group_a_pass_1 = self._create_test_result( 279 group_name='grpup_a', status=test_runner_base.PASSED_STATUS 280 ) 281 group_b_pass_1 = self._create_test_result( 282 group_name='grpup_b', status=test_runner_base.PASSED_STATUS 283 ) 284 group_c_pass_1 = self._create_test_result( 285 group_name='grpup_c', status=test_runner_base.PASSED_STATUS 286 ) 287 group_b_fail_1 = self._create_test_result( 288 group_name='grpup_b', status=test_runner_base.FAILED_STATUS 289 ) 290 group_c_fail_1 = self._create_test_result( 291 group_name='grpup_c', status=test_runner_base.FAILED_STATUS 292 ) 293 group_c_ignore_1 = self._create_test_result( 294 group_name='grpup_c', status=test_runner_base.IGNORED_STATUS 295 ) 296 reporter_1 = result_reporter.ResultReporter() 297 reporter_1.all_test_results.extend( 298 [group_a_pass_1, group_b_pass_1, group_c_pass_1] 299 ) 300 reporter_2 = result_reporter.ResultReporter() 301 reporter_2.all_test_results.extend( 302 [group_b_fail_1, group_c_fail_1, group_c_ignore_1] 303 ) 304 305 info_dict = {} 306 aei.AtestExecutionInfo._arrange_test_result( 307 info_dict, [reporter_1, reporter_2] 308 ) 309 expect_group_a_summary = { 310 aei._STATUS_IGNORED_KEY: 0, 311 aei._STATUS_FAILED_KEY: 0, 312 aei._STATUS_PASSED_KEY: 1, 313 } 314 self.assertEqual( 315 expect_group_a_summary, 316 info_dict[aei._TEST_RUNNER_KEY]['someRunner']['grpup_a'][ 317 aei._SUMMARY_KEY 318 ], 319 ) 320 321 expect_group_b_summary = { 322 aei._STATUS_IGNORED_KEY: 0, 323 aei._STATUS_FAILED_KEY: 1, 324 aei._STATUS_PASSED_KEY: 1, 325 } 326 self.assertEqual( 327 expect_group_b_summary, 328 info_dict[aei._TEST_RUNNER_KEY]['someRunner']['grpup_b'][ 329 aei._SUMMARY_KEY 330 ], 331 ) 332 333 expect_group_c_summary = { 334 aei._STATUS_IGNORED_KEY: 1, 335 aei._STATUS_FAILED_KEY: 1, 336 aei._STATUS_PASSED_KEY: 1, 337 } 338 self.assertEqual( 339 expect_group_c_summary, 340 info_dict[aei._TEST_RUNNER_KEY]['someRunner']['grpup_c'][ 341 aei._SUMMARY_KEY 342 ], 343 ) 344 345 expect_total_summary = { 346 aei._STATUS_IGNORED_KEY: 1, 347 aei._STATUS_FAILED_KEY: 2, 348 aei._STATUS_PASSED_KEY: 3, 349 } 350 self.assertEqual(expect_total_summary, info_dict[aei._TOTAL_SUMMARY_KEY]) 351 352 def test_preparation_time(self): 353 """Test preparation_time method.""" 354 start_time = time.time() 355 aei.PREPARE_END_TIME = None 356 self.assertTrue(aei.preparation_time(start_time) is None) 357 aei.PREPARE_END_TIME = time.time() 358 self.assertFalse(aei.preparation_time(start_time) is None) 359 360 def test_arrange_test_result_multi_runner(self): 361 """Test _arrange_test_result method with multi runner.""" 362 runner_a_pass_1 = self._create_test_result( 363 runner_name='runner_a', status=test_runner_base.PASSED_STATUS 364 ) 365 runner_a_pass_2 = self._create_test_result( 366 runner_name='runner_a', status=test_runner_base.PASSED_STATUS 367 ) 368 runner_a_pass_3 = self._create_test_result( 369 runner_name='runner_a', status=test_runner_base.PASSED_STATUS 370 ) 371 runner_b_fail_1 = self._create_test_result( 372 runner_name='runner_b', status=test_runner_base.FAILED_STATUS 373 ) 374 runner_b_fail_2 = self._create_test_result( 375 runner_name='runner_b', status=test_runner_base.FAILED_STATUS 376 ) 377 runner_b_ignore_1 = self._create_test_result( 378 runner_name='runner_b', status=test_runner_base.IGNORED_STATUS 379 ) 380 381 reporter_1 = result_reporter.ResultReporter() 382 reporter_1.all_test_results.extend( 383 [runner_a_pass_1, runner_a_pass_2, runner_a_pass_3] 384 ) 385 reporter_2 = result_reporter.ResultReporter() 386 reporter_2.all_test_results.extend( 387 [runner_b_fail_1, runner_b_fail_2, runner_b_ignore_1] 388 ) 389 info_dict = {} 390 aei.AtestExecutionInfo._arrange_test_result( 391 info_dict, [reporter_1, reporter_2] 392 ) 393 expect_group_a_summary = { 394 aei._STATUS_IGNORED_KEY: 0, 395 aei._STATUS_FAILED_KEY: 0, 396 aei._STATUS_PASSED_KEY: 3, 397 } 398 self.assertEqual( 399 expect_group_a_summary, 400 info_dict[aei._TEST_RUNNER_KEY]['runner_a']['someModule'][ 401 aei._SUMMARY_KEY 402 ], 403 ) 404 405 expect_group_b_summary = { 406 aei._STATUS_IGNORED_KEY: 1, 407 aei._STATUS_FAILED_KEY: 2, 408 aei._STATUS_PASSED_KEY: 0, 409 } 410 self.assertEqual( 411 expect_group_b_summary, 412 info_dict[aei._TEST_RUNNER_KEY]['runner_b']['someModule'][ 413 aei._SUMMARY_KEY 414 ], 415 ) 416 417 expect_total_summary = { 418 aei._STATUS_IGNORED_KEY: 1, 419 aei._STATUS_FAILED_KEY: 2, 420 aei._STATUS_PASSED_KEY: 3, 421 } 422 self.assertEqual(expect_total_summary, info_dict[aei._TOTAL_SUMMARY_KEY]) 423 424 def _create_test_result(self, **kwargs): 425 """A Helper to create TestResult""" 426 test_info = test_runner_base.TestResult(**RESULT_TEST_TEMPLATE._asdict()) 427 return test_info._replace(**kwargs) 428 429 430if __name__ == '__main__': 431 unittest.main() 432