# Copyright 2017, The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """ Utility functions for atest. """ # pylint: disable=import-outside-toplevel # pylint: disable=too-many-lines from __future__ import print_function import enum import datetime import fnmatch import hashlib import html import importlib import itertools import json import logging import os import pickle import platform import re import shutil import subprocess import sys import time import urllib import zipfile from dataclasses import dataclass from multiprocessing import Process from pathlib import Path from typing import Any, Dict, List, Set import xml.etree.ElementTree as ET from atest.atest_enum import DetectType, ExitCode, FilterType #pylint: disable=wrong-import-position from atest import atest_decorator from atest import atest_error from atest import constants from atest.metrics import metrics from atest.metrics import metrics_utils from atest.tf_proto import test_record_pb2 _BASH_RESET_CODE = '\033[0m\n' DIST_OUT_DIR = Path(os.environ.get(constants.ANDROID_BUILD_TOP, os.getcwd()) + '/out/dist/') MAINLINE_MODULES_EXT_RE = re.compile(r'(.apex|.apks|.apk)$') # Arbitrary number to limit stdout for failed runs in _run_limited_output. # Reason for its use is that the make command itself has its own carriage # return output mechanism that when collected line by line causes the streaming # full_output list to be extremely large. _FAILED_OUTPUT_LINE_LIMIT = 100 # Regular expression to match the start of a ninja compile: # ex: [ 99% 39710/39711] _BUILD_COMPILE_STATUS = re.compile(r'\[\s*(\d{1,3}%\s+)?\d+/\d+\]') _BUILD_FAILURE = 'FAILED: ' CMD_RESULT_PATH = os.path.join(os.environ.get(constants.ANDROID_BUILD_TOP, os.getcwd()), 'tools/asuite/atest/test_data', 'test_commands.json') BUILD_TOP_HASH = hashlib.md5(os.environ.get(constants.ANDROID_BUILD_TOP, ''). encode()).hexdigest() _DEFAULT_TERMINAL_WIDTH = 80 _DEFAULT_TERMINAL_HEIGHT = 25 _BUILD_CMD = 'build/soong/soong_ui.bash' _FIND_MODIFIED_FILES_CMDS = ( "cd {};" "local_branch=$(git rev-parse --abbrev-ref HEAD);" "remote_branch=$(git branch -r | grep '\\->' | awk '{{print $1}}');" # Get the number of commits from local branch to remote branch. "ahead=$(git rev-list --left-right --count $local_branch...$remote_branch " "| awk '{{print $1}}');" # Get the list of modified files from HEAD to previous $ahead generation. "git diff HEAD~$ahead --name-only") _ANDROID_BUILD_EXT = ('.bp', '.mk') # Set of special chars for various purposes. _REGEX_CHARS = {'[', '(', '{', '|', '\\', '*', '?', '+', '^'} _WILDCARD_CHARS = {'?', '*'} _ROOT_PREPARER = "com.android.tradefed.targetprep.RootTargetPreparer" _WILDCARD_FILTER_RE = re.compile(r'.*[?|*]$') _REGULAR_FILTER_RE = re.compile(r'.*\w$') SUGGESTIONS = { # (b/198581508) Do not run "adb sync" for the users. 'CANNOT LINK EXECUTABLE': 'Please run "adb sync" or reflash the device(s).', # (b/177626045) If Atest does not install target application properly. 'Runner reported an invalid method': 'Please reflash the device(s).' } _BUILD_ENV = {} @dataclass class BuildEnvProfiler: """Represents the condition before and after trigging build.""" ninja_file: Path ninja_file_mtime: float variable_file: Path variable_file_md5: str clean_out: bool build_files_integrity: bool @enum.unique class BuildOutputMode(enum.Enum): "Represents the different ways to display build output." STREAMED = 'streamed' LOGGED = 'logged' def __init__(self, arg_name: str): self._description = arg_name # pylint: disable=missing-function-docstring def description(self): return self._description def get_build_cmd(dump=False): """Compose build command with no-absolute path and flag "--make-mode". Args: dump: boolean that determines the option of build/soong/soong_iu.bash. True: used to dump build variables, equivalent to printconfig. e.g. build/soong/soong_iu.bash --dumpvar-mode False: (default) used to build targets in make mode. e.g. build/soong/soong_iu.bash --make-mode Returns: A list of soong build command. """ make_cmd = ('%s/%s' % (os.path.relpath(os.environ.get( constants.ANDROID_BUILD_TOP, os.getcwd()), os.getcwd()), _BUILD_CMD)) if dump: return [make_cmd, '--dumpvar-mode', 'report_config'] return [make_cmd, '--make-mode'] def _capture_fail_section(full_log): """Return the error message from the build output. Args: full_log: List of strings representing full output of build. Returns: capture_output: List of strings that are build errors. """ am_capturing = False capture_output = [] for line in full_log: if am_capturing and _BUILD_COMPILE_STATUS.match(line): break if am_capturing or line.startswith(_BUILD_FAILURE): capture_output.append(line) am_capturing = True continue return capture_output def _capture_limited_output(full_log): """Return the limited error message from capture_failed_section. Args: full_log: List of strings representing full output of build. Returns: output: List of strings that are build errors. """ # Parse out the build error to output. output = _capture_fail_section(full_log) if not output: output = full_log if len(output) >= _FAILED_OUTPUT_LINE_LIMIT: output = output[-_FAILED_OUTPUT_LINE_LIMIT:] output = 'Output (may be trimmed):\n%s' % ''.join(output) return output # TODO: b/187122993 refine subprocess with 'with-statement' in fixit week. def _run_limited_output(cmd, env_vars=None): """Runs a given command and streams the output on a single line in stdout. Args: cmd: A list of strings representing the command to run. env_vars: Optional arg. Dict of env vars to set during build. Raises: subprocess.CalledProcessError: When the command exits with a non-0 exitcode. """ # Send stderr to stdout so we only have to deal with a single pipe. proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=env_vars) sys.stdout.write('\n') term_width, _ = get_terminal_size() white_space = " " * int(term_width) full_output = [] while proc.poll() is None: line = proc.stdout.readline().decode('utf-8') # Readline will often return empty strings. if not line: continue full_output.append(line) # Trim the line to the width of the terminal. # Note: Does not handle terminal resizing, which is probably not worth # checking the width every loop. if len(line) >= term_width: line = line[:term_width - 1] # Clear the last line we outputted. sys.stdout.write('\r%s\r' % white_space) sys.stdout.write('%s' % line.strip()) sys.stdout.flush() # Reset stdout (on bash) to remove any custom formatting and newline. sys.stdout.write(_BASH_RESET_CODE) sys.stdout.flush() # Wait for the Popen to finish completely before checking the returncode. proc.wait() if proc.returncode != 0: # get error log from "OUT_DIR/error.log" error_log_file = os.path.join(get_build_out_dir(), "error.log") output = [] if os.path.isfile(error_log_file): if os.stat(error_log_file).st_size > 0: with open(error_log_file) as f: output = f.read() if not output: output = _capture_limited_output(full_output) raise subprocess.CalledProcessError(proc.returncode, cmd, output) def get_build_out_dir() -> str: """Get android build out directory. The order of the rules are: 1. OUT_DIR 2. OUT_DIR_COMMON_BASE 3. ANDROID_BUILD_TOP/out Returns: String of the out directory. """ build_top = os.environ.get(constants.ANDROID_BUILD_TOP, '/') # Get the out folder if user specified $OUT_DIR custom_out_dir = os.environ.get(constants.ANDROID_OUT_DIR) custom_out_dir_common_base = os.environ.get( constants.ANDROID_OUT_DIR_COMMON_BASE) user_out_dir = None # If OUT_DIR == /output, the output dir will always be /outdir # regardless of branch names. (Not recommended.) if custom_out_dir: if os.path.isabs(custom_out_dir): user_out_dir = custom_out_dir else: user_out_dir = os.path.join(build_top, custom_out_dir) # https://source.android.com/setup/build/initializing#using-a-separate-output-directory # If OUT_DIR_COMMON_BASE is /output and the source tree is /src/master1, # the output dir will be /output/master1. elif custom_out_dir_common_base: build_top_basename = os.path.basename(build_top) if os.path.isabs(custom_out_dir_common_base): user_out_dir = os.path.join(custom_out_dir_common_base, build_top_basename) else: user_out_dir = os.path.join(build_top, custom_out_dir_common_base, build_top_basename) if user_out_dir: return user_out_dir return os.path.join(build_top, "out") def update_build_env(env: Dict[str, str]): """Method that updates build environment variables.""" # pylint: disable=global-statement global _BUILD_ENV _BUILD_ENV.update(env) def build(build_targets: Set[str]): """Shell out and invoke run_build_cmd to make build_targets. Args: build_targets: A set of strings of build targets to make. Returns: Boolean of whether build command was successful, True if nothing to build. """ if not build_targets: logging.debug('No build targets, skipping build.') return True # pylint: disable=global-statement global _BUILD_ENV full_env_vars = os.environ.copy() update_build_env(full_env_vars) print('\n%s\n%s' % ( colorize("Building Dependencies...", constants.CYAN), ', '.join(build_targets))) logging.debug('Building Dependencies: %s', ' '.join(build_targets)) cmd = get_build_cmd() + list(build_targets) return _run_build_cmd(cmd, _BUILD_ENV) def _run_build_cmd(cmd: List[str], env_vars: Dict[str, str]): """The main process of building targets. Args: cmd: A list of soong command. env_vars: Dict of environment variables used for build. Returns: Boolean of whether build command was successful, True if nothing to build. """ logging.debug('Executing command: %s', cmd) build_profiler = _build_env_profiling() try: if env_vars.get('BUILD_OUTPUT_MODE') == BuildOutputMode.STREAMED.value: print() subprocess.check_call(cmd, stderr=subprocess.STDOUT, env=env_vars) else: # Note that piping stdout forces Soong to switch to 'dumb terminal # mode' which only prints completed actions. This gives users the # impression that actions are taking longer than they really are. # See b/233044822 for more details. log_path = Path(get_build_out_dir()).joinpath('verbose.log.gz') print('\n(Build log may not reflect actual status in simple output' 'mode; check {} for detail after build finishes.)'.format( colorize(f'{log_path}', constants.CYAN) ), end='') _run_limited_output(cmd, env_vars=env_vars) _send_build_condition_metrics(build_profiler, cmd) logging.info('Build successful') return True except subprocess.CalledProcessError as err: logging.error('Build failure when running: %s', ' '.join(cmd)) if err.output: logging.error(err.output) return False # pylint: disable=unused-argument def get_result_server_args(for_test_mapping=False): """Return list of args for communication with result server. Args: for_test_mapping: True if the test run is for Test Mapping to include additional reporting args. Default is False. """ # Customize test mapping argument here if needed. return constants.RESULT_SERVER_ARGS def sort_and_group(iterable, key): """Sort and group helper function.""" return itertools.groupby(sorted(iterable, key=key), key=key) def is_test_mapping(args): """Check if the atest command intends to run tests in test mapping. When atest runs tests in test mapping, it must have at most one test specified. If a test is specified, it must be started with `:`, which means the test value is a test group name in TEST_MAPPING file, e.g., `:postsubmit`. If --host-unit-test-only or --smart-testing-local was applied, it doesn't intend to be a test_mapping test. If any test mapping options is specified, the atest command must also be set to run tests in test mapping files. Args: args: arg parsed object. Returns: True if the args indicates atest shall run tests in test mapping. False otherwise. """ if any((args.host_unit_test_only, args.smart_testing_local)): return False if any((args.test_mapping, args.include_subdirs, not args.tests)): return True # ':postsubmit' implicitly indicates running in test-mapping mode. return all((len(args.tests) == 1, args.tests[0][0] == ':')) @atest_decorator.static_var("cached_has_colors", {}) def _has_colors(stream): """Check the output stream is colorful. Args: stream: The standard file stream. Returns: True if the file stream can interpreter the ANSI color code. """ cached_has_colors = _has_colors.cached_has_colors if stream in cached_has_colors: return cached_has_colors[stream] cached_has_colors[stream] = True # Following from Python cookbook, #475186 if not hasattr(stream, "isatty"): cached_has_colors[stream] = False return False if not stream.isatty(): # Auto color only on TTYs cached_has_colors[stream] = False return False # curses.tigetnum() cannot be used for telling supported color numbers # because it does not come with the prebuilt py3-cmd. return cached_has_colors[stream] def colorize(text, color, bp_color=None): """ Convert to colorful string with ANSI escape code. Args: text: A string to print. color: Forground(Text) color which is an ANSI code shift for colorful print. They are defined in constants_default.py. bp_color: Backgroud color which is an ANSI code shift for colorful print. Returns: Colorful string with ANSI escape code. """ clr_pref = '\033[1;' clr_suff = '\033[0m' has_colors = _has_colors(sys.stdout) if has_colors: background_color = '' if bp_color: # Foreground(Text) ranges from 30-37 text_color = 30 + color # Background ranges from 40-47 background_color = ';%d' % (40 + bp_color) else: text_color = 30 + color clr_str = "%s%d%sm%s%s" % (clr_pref, text_color, background_color, text, clr_suff) else: clr_str = text return clr_str def colorful_print(text, color, bp_color=None, auto_wrap=True): """Print out the text with color. Args: text: A string to print. color: Forground(Text) color which is an ANSI code shift for colorful print. They are defined in constants_default.py. bp_color: Backgroud color which is an ANSI code shift for colorful print. auto_wrap: If True, Text wraps while print. """ output = colorize(text, color, bp_color) if auto_wrap: print(output) else: print(output, end="") def get_terminal_size(): """Get terminal size and return a tuple. Returns: 2 integers: the size of X(columns) and Y(lines/rows). """ # Determine the width of the terminal. We'll need to clear this many # characters when carriage returning. Set default value as 80. columns, rows = shutil.get_terminal_size( fallback=(_DEFAULT_TERMINAL_WIDTH, _DEFAULT_TERMINAL_HEIGHT)) return columns, rows def handle_test_runner_cmd(input_test, test_cmds, do_verification=False, result_path=constants.VERIFY_DATA_PATH): """Handle the runner command of input tests. Args: input_test: A string of input tests pass to atest. test_cmds: A list of strings for running input tests. do_verification: A boolean to indicate the action of this method. True: Do verification without updating result map and raise DryRunVerificationError if verifying fails. False: Update result map, if the former command is different with current command, it will confirm with user if they want to update or not. result_path: The file path for saving result. """ full_result_content = {} if os.path.isfile(result_path): with open(result_path) as json_file: full_result_content = json.load(json_file) former_test_cmds = full_result_content.get(input_test, []) test_cmds = _normalize(test_cmds) former_test_cmds = _normalize(former_test_cmds) if not _are_identical_cmds(test_cmds, former_test_cmds): if do_verification: raise atest_error.DryRunVerificationError( 'Dry run verification failed, former commands: {}'.format( former_test_cmds)) if former_test_cmds: # If former_test_cmds is different from test_cmds, ask users if they # are willing to update the result. print('Former cmds = %s' % former_test_cmds) print('Current cmds = %s' % test_cmds) if not prompt_with_yn_result('Do you want to update former result ' 'to the latest one?', True): print('SKIP updating result!!!') return else: # If current commands are the same as the formers, no need to update # result. return full_result_content[input_test] = test_cmds with open(result_path, 'w') as outfile: json.dump(full_result_content, outfile, indent=0) print('Save result mapping to %s' % result_path) def _normalize(cmd_list): """Method that normalize commands. Note that '--atest-log-file-path' is not considered a critical argument, therefore, it will be removed during the comparison. Also, atest can be ran in any place, so verifying relative path, LD_LIBRARY_PATH, and --proto-output-file is regardless as well. Args: cmd_list: A list with one element. E.g. ['cmd arg1 arg2 True'] Returns: A list with elements. E.g. ['cmd', 'arg1', 'arg2', 'True'] """ _cmd = ' '.join(cmd_list).split() for cmd in _cmd: if cmd.startswith('--skip-all-system-status-check'): _cmd.remove(cmd) continue if cmd.startswith('--atest-log-file-path'): _cmd.remove(cmd) continue if cmd.startswith('LD_LIBRARY_PATH='): _cmd.remove(cmd) continue if cmd.startswith('--proto-output-file='): _cmd.remove(cmd) continue if cmd.startswith('--log-root-path'): _cmd.remove(cmd) continue if _BUILD_CMD in cmd: _cmd.remove(cmd) _cmd.append(os.path.join('./', _BUILD_CMD)) continue return _cmd def _are_identical_cmds(current_cmds, former_cmds): """Tell two commands are identical. Args: current_cmds: A list of strings for running input tests. former_cmds: A list of strings recorded from the previous run. Returns: True if both commands are identical, False otherwise. """ # Always sort cmd list to make it comparable. current_cmds.sort() former_cmds.sort() return current_cmds == former_cmds def _get_hashed_file_name(main_file_name): """Convert the input string to a md5-hashed string. If file_extension is given, returns $(hashed_string).$(file_extension), otherwise $(hashed_string).cache. Args: main_file_name: The input string need to be hashed. Returns: A string as hashed file name with .cache file extension. """ hashed_fn = hashlib.md5(str(main_file_name).encode()) hashed_name = hashed_fn.hexdigest() return hashed_name + '.cache' def md5sum(filename): """Generate MD5 checksum of a file. Args: name: A string of a filename. Returns: A string of hashed MD5 checksum. """ filename = Path(filename) if not filename.is_file(): return "" with open(filename, 'rb') as target: content = target.read() if not isinstance(content, bytes): content = content.encode('utf-8') return hashlib.md5(content).hexdigest() def check_md5(check_file, missing_ok=False): """Method equivalent to 'md5sum --check /file/to/check'. Args: check_file: A string of filename that stores filename and its md5 checksum. missing_ok: A boolean that considers OK even when the check_file does not exist. Using missing_ok=True allows ignoring md5 check especially for initial run that the check_file has not yet generated. Using missing_ok=False ensures the consistency of files, and guarantees the process is successfully completed. Returns: When missing_ok is True (soft check): - True if the checksum is consistent with the actual MD5, even the check_file is missing or not a valid JSON. - False when the checksum is inconsistent with the actual MD5. When missing_ok is False (ensure the process completed properly): - True if the checksum is consistent with the actual MD5. - False otherwise. """ if not Path(check_file).is_file(): if not missing_ok: logging.debug( 'Unable to verify: %s not found.', check_file) return missing_ok content = load_json_safely(check_file) if content: for filename, md5 in content.items(): if md5sum(filename) != md5: logging.debug('%s has altered.', filename) return False return True return False def save_md5(filenames, save_file): """Method equivalent to 'md5sum file1 file2 > /file/to/check' Args: filenames: A list of filenames. save_file: Filename for storing files and their md5 checksums. """ data = {} for f in filenames: name = Path(f) if not name.is_file(): logging.warning(' ignore %s: not a file.', name) data.update({str(name): md5sum(name)}) with open(save_file, 'w+') as _file: json.dump(data, _file) def get_cache_root(): """Get the root path dir for cache. Use branch and target information as cache_root. The path will look like ~/.atest/info_cache/$hash(branch+target) Returns: A string of the path of the root dir of cache. """ manifest_branch = get_manifest_branch() if not manifest_branch: manifest_branch = os.environ.get( constants.ANDROID_BUILD_TOP, constants.ANDROID_BUILD_TOP) # target build_target = os.path.basename( os.environ.get(constants.ANDROID_PRODUCT_OUT, constants.ANDROID_PRODUCT_OUT)) branch_target_hash = hashlib.md5( (constants.MODE + manifest_branch + build_target).encode()).hexdigest() return os.path.join(get_misc_dir(), '.atest', 'info_cache', branch_target_hash[:8]) def get_test_info_cache_path(test_reference, cache_root=None): """Get the cache path of the desired test_infos. Args: test_reference: A string of the test. cache_root: Folder path where stores caches. Returns: A string of the path of test_info cache. """ if not cache_root: cache_root = get_cache_root() return os.path.join(cache_root, _get_hashed_file_name(test_reference)) def update_test_info_cache(test_reference, test_infos, cache_root=None): """Update cache content which stores a set of test_info objects through pickle module, each test_reference will be saved as a cache file. Args: test_reference: A string referencing a test. test_infos: A set of TestInfos. cache_root: Folder path for saving caches. """ if not cache_root: cache_root = get_cache_root() if not os.path.isdir(cache_root): os.makedirs(cache_root) cache_path = get_test_info_cache_path(test_reference, cache_root) # Save test_info to files. try: with open(cache_path, 'wb') as test_info_cache_file: logging.debug('Saving cache %s.', cache_path) pickle.dump(test_infos, test_info_cache_file, protocol=2) except (pickle.PicklingError, TypeError, IOError) as err: # Won't break anything, just log this error, and collect the exception # by metrics. logging.debug('Exception raised: %s', err) metrics_utils.handle_exc_and_send_exit_event( constants.ACCESS_CACHE_FAILURE) def load_test_info_cache(test_reference, cache_root=None): """Load cache by test_reference to a set of test_infos object. Args: test_reference: A string referencing a test. cache_root: Folder path for finding caches. Returns: A list of TestInfo namedtuple if cache found, else None. """ if not cache_root: cache_root = get_cache_root() cache_file = get_test_info_cache_path(test_reference, cache_root) if os.path.isfile(cache_file): logging.debug('Loading cache %s.', cache_file) try: with open(cache_file, 'rb') as config_dictionary_file: return pickle.load(config_dictionary_file, encoding='utf-8') except (pickle.UnpicklingError, ValueError, TypeError, EOFError, IOError, ImportError) as err: # Won't break anything, just remove the old cache, log this error, # and collect the exception by metrics. logging.debug('Exception raised: %s', err) os.remove(cache_file) metrics_utils.handle_exc_and_send_exit_event( constants.ACCESS_CACHE_FAILURE) return None def clean_test_info_caches(tests, cache_root=None): """Clean caches of input tests. Args: tests: A list of test references. cache_root: Folder path for finding caches. """ if not cache_root: cache_root = get_cache_root() for test in tests: cache_file = get_test_info_cache_path(test, cache_root) if os.path.isfile(cache_file): logging.debug('Removing cache: %s', cache_file) try: os.remove(cache_file) except IOError as err: logging.debug('Exception raised: %s', err) metrics_utils.handle_exc_and_send_exit_event( constants.ACCESS_CACHE_FAILURE) def get_modified_files(root_dir): """Get the git modified files. The git path here is git top level of the root_dir. It's inevitable to utilise different commands to fulfill 2 scenario: 1. locate unstaged/staged files 2. locate committed files but not yet merged. the 'git_status_cmd' fulfils the former while the 'find_modified_files' fulfils the latter. Args: root_dir: the root where it starts finding. Returns: A set of modified files altered since last commit. """ modified_files = set() try: find_git_cmd = 'cd {}; git rev-parse --show-toplevel'.format(root_dir) git_paths = subprocess.check_output( find_git_cmd, shell=True).decode().splitlines() for git_path in git_paths: # Find modified files from git working tree status. git_status_cmd = ("repo forall {} -c git status --short | " "awk '{{print $NF}}'").format(git_path) modified_wo_commit = subprocess.check_output( git_status_cmd, shell=True).decode().rstrip().splitlines() for change in modified_wo_commit: modified_files.add( os.path.normpath('{}/{}'.format(git_path, change))) # Find modified files that are committed but not yet merged. find_modified_files = _FIND_MODIFIED_FILES_CMDS.format(git_path) commit_modified_files = subprocess.check_output( find_modified_files, shell=True).decode().splitlines() for line in commit_modified_files: modified_files.add(os.path.normpath('{}/{}'.format( git_path, line))) except (OSError, subprocess.CalledProcessError) as err: logging.debug('Exception raised: %s', err) return modified_files def delimiter(char, length=_DEFAULT_TERMINAL_WIDTH, prenl=0, postnl=0): """A handy delimiter printer. Args: char: A string used for delimiter. length: An integer for the replication. prenl: An integer that insert '\n' before delimiter. postnl: An integer that insert '\n' after delimiter. Returns: A string of delimiter. """ return prenl * '\n' + char * length + postnl * '\n' def find_files(path, file_name=constants.TEST_MAPPING): """Find all files with given name under the given path. Args: path: A string of path in source. file_name: The file name pattern for finding matched files. Returns: A list of paths of the files with the matching name under the given path. """ match_files = [] for root, _, filenames in os.walk(path): try: for filename in fnmatch.filter(filenames, file_name): match_files.append(os.path.join(root, filename)) except re.error as e: msg = "Unable to locate %s among %s" % (file_name, filenames) logging.debug(msg) logging.debug("Exception: %s", e) metrics.AtestExitEvent( duration=metrics_utils.convert_duration(0), exit_code=ExitCode.COLLECT_ONLY_FILE_NOT_FOUND, stacktrace=msg, logs=str(e)) return match_files def extract_zip_text(zip_path): """Extract the text files content for input zip file. Args: zip_path: The file path of zip. Returns: The string in input zip file. """ content = '' try: with zipfile.ZipFile(zip_path) as zip_file: for filename in zip_file.namelist(): if os.path.isdir(filename): continue # Force change line if multiple text files in zip content = content + '\n' # read the file with zip_file.open(filename) as extract_file: for line in extract_file: if matched_tf_error_log(line.decode()): content = content + line.decode() except zipfile.BadZipfile as err: logging.debug('Exception raised: %s', err) return content def matched_tf_error_log(content): """Check if the input content matched tradefed log pattern. The format will look like this. 05-25 17:37:04 W/XXXXXX 05-25 17:37:04 E/XXXXXX Args: content: Log string. Returns: True if the content matches the regular expression for tradefed error or warning log. """ reg = ('^((0[1-9])|(1[0-2]))-((0[1-9])|([12][0-9])|(3[0-1])) ' '(([0-1][0-9])|([2][0-3])):([0-5][0-9]):([0-5][0-9]) (E|W/)') if re.search(reg, content): return True return False def has_valid_cert(): """Check whether the certificate is valid. Returns: True if the cert is valid. """ if not constants.CERT_STATUS_CMD: return False try: return (not subprocess.check_call(constants.CERT_STATUS_CMD, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)) except subprocess.CalledProcessError: return False # pylint: disable=too-many-locals def get_flakes(branch='', target='', test_name='', test_module='', test_method=''): """Get flake information. Args: branch: A string of branch name. target: A string of target. test_name: A string of test suite name. test_module: A string of test module. test_method: A string of test method. Returns: A dictionary of flake info. None if no flakes service exists. """ if not branch: branch = constants.FLAKE_BRANCH if not target: target = constants.FLAKE_TARGET if not test_name: test_name = constants.FLAKE_TEST_NAME # Currently lock the flake information from test-mapping test # which only runs on cuttlefish(x86) devices. # TODO: extend supporting other devices if test_module: test_module = 'x86 {}'.format(test_module) flake_service = os.path.join(constants.FLAKE_SERVICE_PATH, constants.FLAKE_FILE) if not os.path.exists(flake_service): logging.debug('Get flakes: Flake service path not exist.') # Send (3, 0) to present no flakes info because service does not exist. metrics.LocalDetectEvent( detect_type=DetectType.NO_FLAKE, result=0) return None if not has_valid_cert(): logging.debug('Get flakes: No valid cert.') # Send (3, 1) to present no flakes info because no valid cert. metrics.LocalDetectEvent( detect_type=DetectType.NO_FLAKE, result=1) return None flake_info = {} start = time.time() try: shutil.copy2(flake_service, constants.FLAKE_TMP_PATH) tmp_service = os.path.join(constants.FLAKE_TMP_PATH, constants.FLAKE_FILE) os.chmod(tmp_service, 0o0755) cmd = [tmp_service, branch, target, test_name, test_module, test_method] logging.debug('Executing: %s', ' '.join(cmd)) output = subprocess.check_output(cmd).decode() percent_template = "{}:".format(constants.FLAKE_PERCENT) postsubmit_template = "{}:".format(constants.FLAKE_POSTSUBMIT) for line in output.splitlines(): if line.startswith(percent_template): flake_info[constants.FLAKE_PERCENT] = line.replace( percent_template, '') if line.startswith(postsubmit_template): flake_info[constants.FLAKE_POSTSUBMIT] = line.replace( postsubmit_template, '') # pylint: disable=broad-except except Exception as e: logging.debug('Exception:%s', e) return None # Send (4, time) to present having flakes info and it spent time. duration = round(time.time()-start) logging.debug('Took %ss to get flakes info', duration) metrics.LocalDetectEvent( detect_type=DetectType.HAS_FLAKE, result=duration) return flake_info def read_test_record(path): """A Helper to read test record proto. Args: path: The proto file path. Returns: The test_record proto instance. """ with open(path, 'rb') as proto_file: msg = test_record_pb2.TestRecord() msg.ParseFromString(proto_file.read()) return msg def has_python_module(module_name): """Detect if the module can be loaded without importing it in real. Args: cmd: A string of the tested module name. Returns: True if found, False otherwise. """ return bool(importlib.util.find_spec(module_name)) def load_json_safely(jsonfile): """Load the given json file as an object. Args: jsonfile: The json file path. Returns: The content of the give json file. Null dict when: 1. the given path doesn't exist. 2. the given path is not a json or invalid format. """ if isinstance(jsonfile, bytes): jsonfile = jsonfile.decode('utf-8') if Path(jsonfile).is_file(): try: with open(jsonfile, 'r') as cache: return json.load(cache) except json.JSONDecodeError: logging.debug('Exception happened while loading %s.', jsonfile) else: logging.debug('%s: File not found.', jsonfile) return {} def get_atest_version(): """Get atest version. Returns: Version string from the VERSION file, e.g. prebuilt 2022-11-24_9314547 (_) If VERSION does not exist (src or local built): 2022-11-24_5d448c50 (_) If the git command fails for unexpected reason: 2022-11-24_unknown (_unknown) """ atest_dir = Path(__file__).resolve().parent version_file = atest_dir.joinpath('VERSION') if Path(version_file).is_file(): return open(version_file).read() # Try fetching commit date (%ci) and commit hash (%h). git_cmd = 'git log -1 --pretty=format:"%ci;%h"' try: # commit date/hash are only available when running from the source # and the local built. result = subprocess.run( git_cmd, shell=True, check=False, capture_output=True, cwd=Path( os.getenv(constants.ANDROID_BUILD_TOP), '').joinpath( 'tools/asuite/atest')) if result.stderr: raise subprocess.CalledProcessError( returncode=0, cmd=git_cmd) raw_date, commit = result.stdout.decode().split(';') date = datetime.datetime.strptime(raw_date, '%Y-%m-%d %H:%M:%S %z').date() # atest_dir doesn't exist will throw FileNotFoundError. except (subprocess.CalledProcessError, FileNotFoundError): # Use today as the commit date for unexpected conditions. date = datetime.datetime.today().date() commit = 'unknown' return f'{date}_{commit}' def get_manifest_branch(show_aosp=False): """Get the manifest branch. (portal xml) (default xml) +--------------------+ _get_include() +-----------------------------+ | .repo/manifest.xml |--------------->| .repo/manifests/default.xml | +--------------------+ +---------------+-------------+ V +--------+ | master | +--------+ Args: show_aosp: A boolean that shows 'aosp' prefix by checking the 'remote' attribute. Returns: The value of 'revision' of the included xml or default.xml. None when no ANDROID_BUILD_TOP or unable to access default.xml. """ build_top = os.getenv(constants.ANDROID_BUILD_TOP) if not build_top: return None portal_xml = Path(build_top).joinpath('.repo', 'manifest.xml') default_xml = Path(build_top).joinpath('.repo/manifests', 'default.xml') def _get_revision(xml): try: xml_root = ET.parse(xml).getroot() except (IOError, OSError, ET.ParseError): # TODO(b/274989179) Change back to warning once warning if not going # to be treat as test failure. Or test_get_manifest_branch unit test # could be fix if return None if portal_xml or default_xml not # exist. logging.info('%s could not be read.', xml) return '' default_tags = xml_root.findall('./default') if default_tags: prefix = '' for tag in default_tags: branch = tag.attrib.get('revision') if show_aosp and tag.attrib.get('remote') == 'aosp': prefix = 'aosp-' return f'{prefix}{branch}' return '' def _get_include(xml): try: xml_root = ET.parse(xml).getroot() except (IOError, OSError, ET.ParseError): # TODO(b/274989179) Change back to warning once warning if not going # to be treat as test failure. Or test_get_manifest_branch unit test # could be fix if return None if portal_xml or default_xml not # exist. logging.info('%s could not be read.', xml) return Path() include_tags = xml_root.findall('./include') if include_tags: for tag in include_tags: name = tag.attrib.get('name') if name: return Path(build_top).joinpath('.repo/manifests', name) return default_xml # 1. Try getting revision from .repo/manifests/default.xml if default_xml.is_file(): return _get_revision(default_xml) # 2. Try getting revision from the included xml of .repo/manifest.xml include_xml = _get_include(portal_xml) if include_xml.is_file(): return _get_revision(include_xml) # 3. Try getting revision directly from manifest.xml (unlikely to happen) return _get_revision(portal_xml) def get_build_target(): """Get the build target form system environment TARGET_PRODUCT.""" build_target = '%s-%s' % ( os.getenv(constants.ANDROID_TARGET_PRODUCT, None), os.getenv(constants.TARGET_BUILD_VARIANT, None)) return build_target def build_module_info_target(module_info_target): """Build module-info.json after deleting the original one. Args: module_info_target: the target name that soong is going to build. """ module_file = 'module-info.json' logging.debug('Generating %s - this is required for ' 'initial runs or forced rebuilds.', module_file) build_start = time.time() product_out = os.getenv(constants.ANDROID_PRODUCT_OUT, None) module_info_path = Path(product_out).joinpath('module-info.json') if module_info_path.is_file(): os.remove(module_info_path) if not build([module_info_target]): sys.exit(ExitCode.BUILD_FAILURE) build_duration = time.time() - build_start metrics.LocalDetectEvent( detect_type=DetectType.ONLY_BUILD_MODULE_INFO, result=int(build_duration)) def has_wildcard(test_name): """ Tell whether the test_name(either a list or string) contains wildcard symbols. Args: test_name: A list or a str. Return: True if test_name contains wildcard, False otherwise. """ if isinstance(test_name, str): return any(char in test_name for char in _WILDCARD_CHARS) if isinstance(test_name, list): for name in test_name: if has_wildcard(name): return True return False def is_build_file(path): """ If input file is one of an android build file. Args: path: A string of file path. Return: True if path is android build file, False otherwise. """ return bool(os.path.splitext(path)[-1] in _ANDROID_BUILD_EXT) def quote(input_str): """ If the input string -- especially in custom args -- contains shell-aware characters, insert a pair of "\" to the input string. e.g. unit(test|testing|testing) -> 'unit(test|testing|testing)' Args: input_str: A string from user input. Returns: A string with single quotes if regex chars were detected. """ if has_chars(input_str, _REGEX_CHARS): return "\'" + input_str + "\'" return input_str def has_chars(input_str, chars): """ Check if the input string contains one of the designated characters. Args: input_str: A string from user input. chars: An iterable object. Returns: True if the input string contains one of the special chars. """ for char in chars: if char in input_str: return True return False def prompt_with_yn_result(msg, default=True): """Prompt message and get yes or no result. Args: msg: The question you want asking. default: boolean to True/Yes or False/No Returns: default value if get KeyboardInterrupt or ValueError exception. """ suffix = '[Y/n]: ' if default else '[y/N]: ' try: return strtobool(input(msg+suffix)) except (ValueError, KeyboardInterrupt): return default def strtobool(val): """Convert a string representation of truth to True or False. Args: val: a string of input value. Returns: True when values are 'y', 'yes', 't', 'true', 'on', and '1'; False when 'n', 'no', 'f', 'false', 'off', and '0'. Raises ValueError if 'val' is anything else. """ if val.lower() in ('y', 'yes', 't', 'true', 'on', '1'): return True if val.lower() in ('n', 'no', 'f', 'false', 'off', '0'): return False raise ValueError("invalid truth value %r" % (val,)) def get_android_junit_config_filters(test_config): """Get the dictionary of a input config for junit config's filters Args: test_config: The path of the test config. Returns: A dictionary include all the filters in the input config. """ filter_dict = {} xml_root = ET.parse(test_config).getroot() option_tags = xml_root.findall('.//option') for tag in option_tags: name = tag.attrib['name'].strip() if name in constants.SUPPORTED_FILTERS: filter_values = filter_dict.get(name, []) value = tag.attrib['value'].strip() filter_values.append(value) filter_dict.update({name: filter_values}) return filter_dict def get_config_parameter(test_config): """Get all the parameter values for the input config Args: test_config: The path of the test config. Returns: A set include all the parameters of the input config. """ parameters = set() xml_root = ET.parse(test_config).getroot() option_tags = xml_root.findall('.//option') for tag in option_tags: name = tag.attrib['name'].strip() if name == constants.CONFIG_DESCRIPTOR: key = tag.attrib['key'].strip() if key == constants.PARAMETER_KEY: value = tag.attrib['value'].strip() parameters.add(value) return parameters def get_config_device(test_config): """Get all the device names from the input config Args: test_config: The path of the test config. Returns: A set include all the device name of the input config. """ devices = set() try: xml_root = ET.parse(test_config).getroot() device_tags = xml_root.findall('.//device') for tag in device_tags: name = tag.attrib['name'].strip() devices.add(name) except ET.ParseError as e: colorful_print('Config has invalid format.', constants.RED) colorful_print('File %s : %s' % (test_config, str(e)), constants.YELLOW) sys.exit(ExitCode.CONFIG_INVALID_FORMAT) return devices def get_mainline_param(test_config): """Get all the mainline-param values for the input config Args: test_config: The path of the test config. Returns: A set include all the parameters of the input config. """ mainline_param = set() xml_root = ET.parse(test_config).getroot() option_tags = xml_root.findall('.//option') for tag in option_tags: name = tag.attrib['name'].strip() if name == constants.CONFIG_DESCRIPTOR: key = tag.attrib['key'].strip() if key == constants.MAINLINE_PARAM_KEY: value = tag.attrib['value'].strip() mainline_param.add(value) return mainline_param def get_adb_devices(): """Run `adb devices` and return a list of devices. Returns: A list of devices. e.g. ['127.0.0.1:40623', '127.0.0.1:40625'] """ probe_cmd = "adb devices | egrep -v \"^List|^$\"||true" suts = subprocess.check_output(probe_cmd, shell=True).decode().splitlines() return [sut.split('\t')[0] for sut in suts] def get_android_config(): """Get Android config as "printconfig" shows. Returns: A dict of Android configurations. """ dump_cmd = get_build_cmd(dump=True) raw_config = subprocess.check_output(dump_cmd).decode('utf-8') android_config = {} for element in raw_config.splitlines(): if not element.startswith('='): key, value = tuple(element.split('=', 1)) android_config.setdefault(key, value) return android_config def get_config_gtest_args(test_config): """Get gtest's module-name and device-path option from the input config Args: test_config: The path of the test config. Returns: A string of gtest's module name. A string of gtest's device path. """ module_name = '' device_path = '' xml_root = ET.parse(test_config).getroot() option_tags = xml_root.findall('.//option') for tag in option_tags: name = tag.attrib['name'].strip() value = tag.attrib['value'].strip() if name == 'native-test-device-path': device_path = value elif name == 'module-name': module_name = value return module_name, device_path def get_arch_name(module_name, is_64=False): """Get the arch folder name for the input module. Scan the test case folders to get the matched arch folder name. Args: module_name: The module_name of test is_64: If need 64 bit arch name, False otherwise. Returns: A string of the arch name. """ arch_32 = ['arm', 'x86'] arch_64 = ['arm64', 'x86_64'] arch_list = arch_32 if is_64: arch_list = arch_64 test_case_root = os.path.join( os.environ.get(constants.ANDROID_TARGET_OUT_TESTCASES, ''), module_name ) if not os.path.isdir(test_case_root): logging.debug('%s does not exist.', test_case_root) return '' for f in os.listdir(test_case_root): if f in arch_list: return f return '' def copy_single_arch_native_symbols( symbol_root, module_name, device_path, is_64=False): """Copy symbol files for native tests which belong to input arch. Args: module_name: The module_name of test device_path: The device path define in test config. is_64: True if need to copy 64bit symbols, False otherwise. """ src_symbol = os.path.join(symbol_root, 'data', 'nativetest', module_name) if is_64: src_symbol = os.path.join( symbol_root, 'data', 'nativetest64', module_name) dst_symbol = os.path.join( symbol_root, device_path[1:], module_name, get_arch_name(module_name, is_64)) if os.path.isdir(src_symbol): # TODO: Use shutil.copytree(src, dst, dirs_exist_ok=True) after # python3.8 if os.path.isdir(dst_symbol): shutil.rmtree(dst_symbol) shutil.copytree(src_symbol, dst_symbol) def copy_native_symbols(module_name, device_path): """Copy symbol files for native tests to match with tradefed file structure. The original symbols will locate at $(PRODUCT_OUT)/symbols/data/nativetest(64)/$(module)/$(stem). From TF, the test binary will locate at /data/local/tmp/$(module)/$(arch)/$(stem). In order to make trace work need to copy the original symbol to $(PRODUCT_OUT)/symbols/data/local/tmp/$(module)/$(arch)/$(stem) Args: module_name: The module_name of test device_path: The device path define in test config. """ symbol_root = os.path.join( os.environ.get(constants.ANDROID_PRODUCT_OUT, ''), 'symbols') if not os.path.isdir(symbol_root): logging.debug('Symbol dir:%s not exist, skip copy symbols.', symbol_root) return # Copy 32 bit symbols if get_arch_name(module_name, is_64=False): copy_single_arch_native_symbols( symbol_root, module_name, device_path, is_64=False) # Copy 64 bit symbols if get_arch_name(module_name, is_64=True): copy_single_arch_native_symbols( symbol_root, module_name, device_path, is_64=True) def get_config_preparer_options(test_config, class_name): """Get all the parameter values for the input config Args: test_config: The path of the test config. class_name: A string of target_preparer Returns: A set include all the parameters of the input config. """ options = {} xml_root = ET.parse(test_config).getroot() option_tags = xml_root.findall( './/target_preparer[@class="%s"]/option' % class_name) for tag in option_tags: name = tag.attrib['name'].strip() value = tag.attrib['value'].strip() options[name] = value return options def is_adb_root(args): """Check whether device has root permission. Args: args: An argspace.Namespace class instance holding parsed args. Returns: True if adb has root permission. """ try: serial = os.environ.get(constants.ANDROID_SERIAL, '') if not serial: serial = args.serial serial_options = ('-s ' + serial) if serial else '' output = subprocess.check_output("adb %s shell id" % serial_options, shell=True, stderr=subprocess.STDOUT).decode() return "uid=0(root)" in output except subprocess.CalledProcessError as err: logging.debug('Exception raised(): %s, Output: %s', err, err.output) raise err def perm_metrics(config_path, adb_root): """Compare adb root permission with RootTargetPreparer in config. Args: config_path: A string of AndroidTest.xml file path. adb_root: A boolean of whether device is root or not. """ # RootTargetPreparer's force-root set in config options = get_config_preparer_options(config_path, _ROOT_PREPARER) if not options: return logging.debug('preparer_options: %s', options) preparer_force_root = True if options.get('force-root', '').upper() == "FALSE": preparer_force_root = False logging.debug(' preparer_force_root: %s', preparer_force_root) if preparer_force_root and not adb_root: logging.debug('DETECT_TYPE_PERMISSION_INCONSISTENT:0') metrics.LocalDetectEvent( detect_type=DetectType.PERMISSION_INCONSISTENT, result=0) elif not preparer_force_root and adb_root: logging.debug('DETECT_TYPE_PERMISSION_INCONSISTENT:1') metrics.LocalDetectEvent( detect_type=DetectType.PERMISSION_INCONSISTENT, result=1) def get_verify_key(tests, extra_args): """Compose test command key. Args: test_name: A list of input tests. extra_args: Dict of extra args to add to test run. Returns: A composed test commands. """ # test_commands is a concatenated string of sorted test_ref+extra_args. # For example, "ITERATIONS=5 hello_world_test" test_commands = tests for key, value in extra_args.items(): if key not in constants.SKIP_VARS: test_commands.append('%s=%s' % (key, str(value))) test_commands.sort() return ' '.join(test_commands) def gen_runner_cmd_to_file(tests, dry_run_cmd, result_path=constants.RUNNER_COMMAND_PATH): """Generate test command and save to file. Args: tests: A String of input tests. dry_run_cmd: A String of dry run command. result_path: A file path for saving result. Returns: A composed run commands. """ normalized_cmd = dry_run_cmd root_path = os.environ.get(constants.ANDROID_BUILD_TOP) if root_path in dry_run_cmd: normalized_cmd = dry_run_cmd.replace(root_path, f"${constants.ANDROID_BUILD_TOP}") results = {} if not os.path.isfile(result_path): results[tests] = normalized_cmd else: with open(result_path) as json_file: results = json.load(json_file) if results.get(tests) != normalized_cmd: results[tests] = normalized_cmd with open(result_path, 'w+') as _file: json.dump(results, _file, indent=0) return results.get(tests, '') def handle_test_env_var(input_test, result_path=constants.VERIFY_ENV_PATH, pre_verify=False): """Handle the environment variable of input tests. Args: input_test: A string of input tests pass to atest. result_path: The file path for saving result. pre_verify: A booloan to separate into pre-verify and actually verify. Returns: 0 is no variable needs to verify, 1 has some variables to next verify. """ full_result_content = {} if os.path.isfile(result_path): with open(result_path) as json_file: full_result_content = json.load(json_file) demand_env_vars = [] demand_env_vars = full_result_content.get(input_test) if demand_env_vars is None: raise atest_error.DryRunVerificationError( '{}: No verify key.'.format(input_test)) # No mapping variables. if demand_env_vars == []: return 0 if pre_verify: return 1 verify_error = [] for env in demand_env_vars: if '=' in env: key, value = env.split('=', 1) env_value = os.environ.get(key, None) if env_value is None or env_value != value: verify_error.append('Environ verification failed, ({0},{1})!=' '({0},{2})'.format(key, value, env_value)) else: if not os.environ.get(env, None): verify_error.append('Missing environ:{}'.format(env)) if verify_error: raise atest_error.DryRunVerificationError('\n'.join(verify_error)) return 1 def generate_buildfiles_checksum(target_dir: Path): """ Method that generate md5 checksum of Android.{bp,mk} files. The checksum of build files are stores in $ANDROID_HOST_OUT/indexes/buildfiles.md5 """ plocate_db = Path(target_dir).joinpath(constants.LOCATE_CACHE) checksum_file = Path(target_dir).joinpath(constants.BUILDFILES_MD5) if plocate_db.is_file(): cmd = (f'locate -d{plocate_db} --existing ' r'--regex "/Android\.(bp|mk)$"') try: result = subprocess.check_output(cmd, shell=True).decode('utf-8') save_md5(result.split(), checksum_file) except subprocess.CalledProcessError: logging.error('Failed to generate %s', checksum_file) def run_multi_proc(func, *args, **kwargs): """Start a process with multiprocessing and return Process object. Args: func: A string of function name which will be the target name. args/kwargs: check doc page: https://docs.python.org/3.8/library/multiprocessing.html#process-and-exceptions Returns: multiprocessing.Process object. """ proc = Process(target=func, *args, **kwargs) proc.start() return proc def get_prebuilt_sdk_tools_dir(): """Get the path for the prebuilt sdk tools root dir. Returns: The absolute path of prebuilt sdk tools directory. """ build_top = Path(os.environ.get(constants.ANDROID_BUILD_TOP, '')) return build_top.joinpath( 'prebuilts/sdk/tools/', str(platform.system()).lower(), 'bin') def is_writable(path): """Check if the given path is writable. Returns: True if input path is writable, False otherwise. """ if not os.path.exists(path): return is_writable(os.path.dirname(path)) return os.access(path, os.W_OK) def get_misc_dir(): """Get the path for the ATest data root dir. Returns: The absolute path of the ATest data root dir. """ home_dir = os.path.expanduser('~') if is_writable(home_dir): return home_dir return get_build_out_dir() def get_full_annotation_class_name(module_info, class_name): """ Get fully qualified class name from a class name. If the given keyword(class_name) is "smalltest", this method can search among source codes and grep the accurate annotation class name: android.test.suitebuilder.annotation.SmallTest Args: module_info: A dict of module_info. class_name: A string of class name. Returns: A string of fully qualified class name, empty string otherwise. """ fullname_re = re.compile( r'import\s+(?P{})(|;)$'.format(class_name), re.I) keyword_re = re.compile( r'import\s+(?P.*\.{})(|;)$'.format(class_name), re.I) build_top = Path(os.environ.get(constants.ANDROID_BUILD_TOP, '')) for f in module_info.get(constants.MODULE_SRCS, []): full_path = build_top.joinpath(f) with open(full_path, 'r') as cache: for line in cache.readlines(): # Accept full class name. match = fullname_re.match(line) if match: return match.group('fqcn') # Search annotation class from keyword. match = keyword_re.match(line) if match: return match.group('fqcn') return "" def has_mixed_type_filters(test_infos): """ There are different types in a test module. Dict test_to_types is mapping module name and the set of types. For example, { 'module_1': {'wildcard class_method'}, 'module_2': {'wildcard class_method', 'regular class_method'}, 'module_3': set() } Args: test_infos: A set of TestInfos. Returns: True if more than one filter type in a test module, False otherwise. """ test_to_types = dict() for test_info in test_infos: filters = test_info.data.get(constants.TI_FILTER, []) filter_types = set() for flt in filters: filter_types |= get_filter_types(flt.to_set_of_tf_strings()) filter_types |= test_to_types.get(test_info.test_name, set()) test_to_types[test_info.test_name] = filter_types for _, types in test_to_types.items(): if len(types) > 1: return True return False def get_filter_types(tf_filter_set): """ Get filter types. Args: tf_filter_set: A set of tf filter strings. Returns: A set of FilterType. """ type_set = set() for tf_filter in tf_filter_set: if _WILDCARD_FILTER_RE.match(tf_filter): logging.debug('Filter and type: (%s, %s)', tf_filter, FilterType.WILDCARD_FILTER.value) type_set.add(FilterType.WILDCARD_FILTER.value) if _REGULAR_FILTER_RE.match(tf_filter): logging.debug('Filter and type: (%s, %s)', tf_filter, FilterType.REGULAR_FILTER.value) type_set.add(FilterType.REGULAR_FILTER.value) return type_set def has_index_files(): """Determine whether the essential index files are done. (b/206886222) checksum may be different even the src is not changed; so the main process needs to wait when the essential index files do not exist. Returns: False if one of the index file does not exist; True otherwise. """ return all(Path(f).is_file() for f in [ constants.CLASS_INDEX, constants.CC_CLASS_INDEX, constants.QCLASS_INDEX, constants.PACKAGE_INDEX]) # pylint: disable=anomalous-backslash-in-string,too-many-branches def get_bp_content(filename: Path, module_type: str) -> Dict: """Get essential content info from an Android.bp. By specifying module_type (e.g. 'android_test', 'android_app'), this method can parse the given starting point and grab 'name', 'instrumentation_for' and 'manifest'. Returns: A dict of mapping test module and target module; e.g. { 'FooUnitTests': {'manifest': 'AndroidManifest.xml', 'target_module': 'Foo'}, 'Foo': {'manifest': 'AndroidManifest-common.xml', 'target_module': ''} } Null dict if there is no content of the given module_type. """ build_file = Path(filename) if not any((build_file.suffix == '.bp', build_file.is_file())): return {} start_from = re.compile(f'^{module_type}\s*\{{') end_with = re.compile(r'^\}$') context_re = re.compile( r'\s*(?P(name|manifest|instrumentation_for))\s*:' r'\s*\"(?P.*)\"\s*,', re.M) with open(build_file, 'r') as cache: data = cache.readlines() content_dict = {} start_recording = False for _line in data: line = _line.strip() if re.match(start_from, line): start_recording = True _dict = {} continue if start_recording: if not re.match(end_with, line): match = re.match(context_re, line) if match: _dict.update( {match.group('key'): match.group('value')}) else: start_recording = False module_name = _dict.get('name') if module_name: content_dict.update( {module_name: { 'manifest': _dict.get( 'manifest', 'AndroidManifest.xml'), 'target_module': _dict.get( 'instrumentation_for', '')} }) return content_dict def get_manifest_info(manifest: Path) -> Dict[str, Any]: """Get the essential info from the given manifest file. This method cares only three attributes: * package * targetPackage * persistent For an instrumentation test, the result will be like: { 'package': 'com.android.foo.tests.unit', 'targetPackage': 'com.android.foo', 'persistent': False } For a target module of the instrumentation test: { 'package': 'com.android.foo', 'targetPackage': '', 'persistent': True } """ mdict = {'package': '', 'target_package': '', 'persistent': False} try: xml_root = ET.parse(manifest).getroot() except (ET.ParseError, FileNotFoundError): return mdict manifest_package_re = re.compile(r'[a-z][\w]+(\.[\w]+)*') # 1. Must probe 'package' name from the top. for item in xml_root.findall('.'): if 'package' in item.attrib.keys(): pkg = item.attrib.get('package') match = manifest_package_re.match(pkg) if match: mdict['package'] = pkg break for item in xml_root.findall('*'): # 2. Probe 'targetPackage' in 'instrumentation' tag. if item.tag == 'instrumentation': for key, value in item.attrib.items(): if 'targetPackage' in key: mdict['target_package'] = value break # 3. Probe 'persistent' in any tags. for key, value in item.attrib.items(): if 'persistent' in key: mdict['persistent'] = value.lower() == 'true' break return mdict # pylint: disable=broad-except def generate_print_result_html(result_file: Path): """Generate a html that collects all log files.""" result_file = Path(result_file) search_dir = Path(result_file).parent.joinpath('log') result_html = Path(search_dir, 'test_logs.html') try: logs = sorted(find_files(str(search_dir), file_name='*')) with open(result_html, 'w') as cache: cache.write('') result = load_json_safely(result_file) if result: cache.write(f'

{"atest " + result.get("args")}

') timestamp = datetime.datetime.fromtimestamp( result_file.stat().st_ctime) cache.write(f'

{timestamp}

') for log in logs: cache.write(f'

' f'{html.escape(Path(log).name)}

') cache.write('') print(f'\nTo access logs, press "ctrl" and click on\n' f'file://{result_html}\n') except Exception as e: logging.debug('Did not generate log html for reason: %s', e) # pylint: disable=broad-except def prompt_suggestions(result_file: Path): """Generate suggestions when detecting keywords in logs.""" result_file = Path(result_file) search_dir = Path(result_file).parent.joinpath('log') logs = sorted(find_files(str(search_dir), file_name='*')) for log in logs: for keyword, suggestion in SUGGESTIONS.items(): try: with open(log, 'r') as cache: content = cache.read() if keyword in content: colorful_print( '[Suggestion] ' + suggestion, color=constants.RED) break # If the given is not a plain text, just ignore it. except Exception: pass def build_files_integrity_is_ok() -> bool: """Return Whether the integrity of build files is OK.""" # 0. Inexistence of the checksum file means a fresh repo sync. if not Path(constants.BUILDFILES_MD5).is_file(): return False # 1. Ensure no build files were added/deleted. with open(constants.BUILDFILES_MD5, 'r') as cache: recorded_amount = len(json.load(cache).keys()) cmd = (f'locate -d{constants.LOCATE_CACHE} --regex ' r'"/Android\.(bp|mk)$" | wc -l') if int(subprocess.getoutput(cmd)) != recorded_amount: return False # 2. Ensure the consistency of all build files. return check_md5(constants.BUILDFILES_MD5, missing_ok=False) def _build_env_profiling() -> BuildEnvProfiler: """Determine the status profile before build. The BuildEnvProfiler object can help use determine whether a build is: 1. clean build. (empty out/ dir) 2. Build files Integrity (Android.bp/Android.mk changes). 3. Environment variables consistency. 4. New Ninja file generated. (mtime of soong/build.ninja) Returns: the BuildProfile object. """ out_dir = Path(get_build_out_dir()) ninja_file = out_dir.joinpath('soong/build.ninja') mtime = ninja_file.stat().st_mtime if ninja_file.is_file() else 0 variables_file = out_dir.joinpath('soong/soong.environment.used.build') return BuildEnvProfiler( ninja_file=ninja_file, ninja_file_mtime=mtime, variable_file=variables_file, variable_file_md5=md5sum(variables_file), clean_out=not ninja_file.exists(), build_files_integrity=build_files_integrity_is_ok() ) def _send_build_condition_metrics( build_profile: BuildEnvProfiler, cmd: List[str]): """Send build conditions by comparing build env profilers.""" # when build module-info.json only, 'module-info.json' will be # the last element. m_mod_info_only = 'module-info.json' in cmd.pop() def ninja_file_is_changed(env_profiler: BuildEnvProfiler) -> bool: """Determine whether the ninja file had been renewal.""" if not env_profiler.ninja_file.is_file(): return True return (env_profiler.ninja_file.stat().st_mtime != env_profiler.ninja_file_mtime) def env_var_is_changed(env_profiler: BuildEnvProfiler) -> bool: """Determine whether soong-related variables had changed.""" return (md5sum(env_profiler.variable_file) != env_profiler.variable_file_md5) def send_data(detect_type): """A simple wrapper of metrics.LocalDetectEvent.""" metrics.LocalDetectEvent(detect_type=detect_type, result=1) # Determine the correct detect type before profiling. # (build module-info.json or build dependencies.) clean_out = (DetectType.MODULE_INFO_CLEAN_OUT if m_mod_info_only else DetectType.BUILD_CLEAN_OUT) ninja_generation = (DetectType.MODULE_INFO_GEN_NINJA if m_mod_info_only else DetectType.BUILD_GEN_NINJA) bpmk_change = (DetectType.MODULE_INFO_BPMK_CHANGE if m_mod_info_only else DetectType.BUILD_BPMK_CHANGE) env_change = (DetectType.MODULE_INFO_ENV_CHANGE if m_mod_info_only else DetectType.BUILD_ENV_CHANGE) src_change = (DetectType.MODULE_INFO_SRC_CHANGE if m_mod_info_only else DetectType.BUILD_SRC_CHANGE) other = (DetectType.MODULE_INFO_OTHER if m_mod_info_only else DetectType.BUILD_OTHER) incremental =(DetectType.MODULE_INFO_INCREMENTAL if m_mod_info_only else DetectType.BUILD_INCREMENTAL) if build_profile.clean_out: send_data(clean_out) else: send_data(incremental) if ninja_file_is_changed(build_profile): send_data(ninja_generation) other_condition = True if not build_profile.build_files_integrity: send_data(bpmk_change) other_condition = False if env_var_is_changed(build_profile): send_data(env_change) other_condition = False if bool(get_modified_files(os.getcwd())): send_data(src_change) other_condition = False if other_condition: send_data(other) def get_local_auto_shardable_tests(): """Get the auto shardable test names in shardable file. The path will be ~/.atest/auto_shard/local_auto_shardable_tests Returns: A list of auto shardable test names. """ shardable_tests_file = Path(get_misc_dir()).joinpath( '.atest/auto_shard/local_auto_shardable_tests') if not shardable_tests_file.exists(): return [] return open(shardable_tests_file, 'r').read().split() def update_shardable_tests(test_name: str, run_time_in_sec: int): """Update local_auto_shardable_test file. Strategy: - Determine to add the module by the run time > 10 mins. - local_auto_shardable_test file path : ~/.atest/auto_shard/local_auto_shardable_tests - The file content template is module name per line: ... """ if run_time_in_sec < 600: return shardable_tests = get_local_auto_shardable_tests() if test_name not in shardable_tests: shardable_tests.append(test_name) logging.info('%s takes %ss (> 600s) to finish. Adding to shardable ' 'test list.', test_name, run_time_in_sec) if not shardable_tests: logging.info('No shardable tests to run.') return shardable_dir = Path(get_misc_dir()).joinpath('.atest/auto_shard') shardable_dir.mkdir(parents=True, exist_ok=True) shardable_tests_file = shardable_dir.joinpath('local_auto_shardable_tests') with open(shardable_tests_file, 'w') as file: file.write('\n'.join(shardable_tests))