Google Git
Sign in
android / platform / tools / asuite / 63d634a8 / . / atest / module_info.py
blob: 308fa6bd46e198dfc0ed01c432b0501f2c1b6537 [file] [log] [blame]
# Copyright 2018, 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.
"""
Module Info class used to hold cached module-info.json.
"""
# pylint: disable=line-too-long,too-many-lines
from __future__ import annotations
import json
import logging
import os
import pickle
import re
import shutil
import tempfile
import time
from pathlib import Path
from typing import Any, Callable, Dict, List, Set
from atest import atest_utils
from atest import constants
from atest.atest_enum import DetectType
from atest.metrics import metrics
# JSON file generated by build system that lists all buildable targets.
_MODULE_INFO = 'module-info.json'
# JSON file generated by build system that lists dependencies for java.
_JAVA_DEP_INFO = 'module_bp_java_deps.json'
# JSON file generated by build system that lists dependencies for cc.
_CC_DEP_INFO = 'module_bp_cc_deps.json'
# JSON file generated by atest merged the content from module-info,
# module_bp_java_deps.json, and module_bp_cc_deps.
_MERGED_INFO = 'atest_merged_dep.json'
Module = Dict[str, Any]
def load_from_file(
module_file: Path = None,
force_build: bool = False,
save_timestamps: bool = False,
) -> ModuleInfo:
"""Factory method that initializes ModuleInfo from the build-generated
JSON file"""
mod_start = time.time()
loader = Loader(
module_file=module_file, force_build=force_build)
mi = loader.load(save_timestamps=save_timestamps)
mod_stop = time.time() - mod_start
metrics.LocalDetectEvent(
detect_type=DetectType.MODULE_INFO_INIT_MS, result=int(mod_stop * 1000))
return mi
def load_from_dict(name_to_module_info: Dict[str, Any]) -> ModuleInfo:
"""Factory method that initializes ModuleInfo from a dictionary."""
path_to_module_info = get_path_to_module_info(name_to_module_info)
return ModuleInfo(
name_to_module_info=name_to_module_info,
path_to_module_info=path_to_module_info,
get_testable_modules=lambda s: _get_testable_modules(
name_to_module_info, path_to_module_info, s),
)
def create_empty() -> ModuleInfo:
"""Factory method that initializes an empty ModuleInfo."""
return ModuleInfo()
class Loader:
"""Class that handles load and merge processes."""
def __init__(self, module_file: Path=None, force_build: bool=False):
self.java_dep_path = Path(
atest_utils.get_build_out_dir()).joinpath('soong', _JAVA_DEP_INFO)
self.cc_dep_path = Path(
atest_utils.get_build_out_dir()).joinpath('soong', _CC_DEP_INFO)
self.merged_dep_path = Path(
os.getenv(constants.ANDROID_PRODUCT_OUT, '')).joinpath(_MERGED_INFO)
self.mod_info_file_path = Path(module_file) if module_file else None
# update_merge_info flag will merge dep files only when any of them have
# changed even force_build == True.
self.update_merge_info = False
# force_build could be from "-m" or smart_build(build files change).
self.force_build = force_build
# If module_file is specified, we're gonna test it so we don't care if
# module_info_target stays None.
self.module_info_target = None
self.name_to_module_info, self.path_to_module_info = self._load_module_info_file()
# Index and checksum files that will be used.
index_dir = Path(os.getenv(constants.ANDROID_HOST_OUT, '')).joinpath('indexes')
self.module_index = index_dir.joinpath(constants.MODULE_INDEX)
self.module_index_proc = None
if self.update_merge_info or not self.module_index.is_file():
# Assumably null module_file reflects a common run, and index testable
# modules only when common runs.
if not module_file:
self.module_index_proc = atest_utils.run_multi_proc(
func=self._get_testable_modules,
kwargs={'index': True})
def load(self, save_timestamps: bool=False):
if save_timestamps:
atest_utils.run_multi_proc(func=self._save_module_info_timestamp)
atest_utils.run_multi_proc(func=atest_utils.save_build_files_timestamp)
return ModuleInfo(
name_to_module_info=self.name_to_module_info,
path_to_module_info=self.path_to_module_info,
module_info_target=self.module_info_target,
mod_info_file_path=self.mod_info_file_path,
get_testable_modules=self.get_testable_modules,
)
def _load_module_info_file(self):
"""Load the module file as ModuleInfo.
+--------------+ +----------------------------------+
| ModuleInfo() | | ModuleInfo(module_file=foo.json) |
+-------+------+ +----------------+-----------------+
| _discover_mod_file_and_target() |
| atest_utils.build() | load
v V
+--------------------------+ +--------------------------+
| module-info.json | | foo.json |
| module_bp_cc_deps.json | | module_bp_cc_deps.json |
| module_bp_java_deps.json | | module_bp_java_deps.json |
+--------------------------+ +--------------------------+
| |
| _merge_soong_info() <--------------------+
v
+============================+
| $ANDROID_PRODUCT_OUT |
| /atest_merged_dep.json |--> load as module info.
+============================+
Returns:
Dict of module name to module info and dict of module path to module info.
"""
if not self.mod_info_file_path:
self.module_info_target, file_path = _discover_mod_file_and_target(
self.force_build)
self.mod_info_file_path = Path(file_path)
# Even undergone a rebuild after _discover_mod_file_and_target(), merge
# atest_merged_dep.json only when module_deps_infos actually change so
# that Atest can decrease disk I/O and ensure data accuracy at all.
self.update_merge_info = self.need_merge_module_info()
start = time.time()
if self.update_merge_info:
# Load the $ANDROID_PRODUCT_OUT/module-info.json for merging.
module_info_json = atest_utils.load_json_safely(self.mod_info_file_path)
name_to_module_info = self._merge_build_system_infos(module_info_json)
duration = time.time() - start
logging.debug('Merging module info took %ss', duration)
metrics.LocalDetectEvent(
detect_type=DetectType.MODULE_MERGE_MS, result=int(duration*1000))
else:
# Load $ANDROID_PRODUCT_OUT/atest_merged_dep.json directly.
name_to_module_info = atest_utils.load_json_safely(self.merged_dep_path)
duration = time.time() - start
logging.debug('Loading module info took %ss', duration)
metrics.LocalDetectEvent(
detect_type=DetectType.MODULE_LOAD_MS, result=int(duration*1000))
_add_missing_variant_modules(name_to_module_info)
logging.debug('Loading %s as module-info.', self.merged_dep_path)
return name_to_module_info, get_path_to_module_info(name_to_module_info)
def _save_module_info_timestamp(self):
"""Dump the timestamp of essential module info files.
* module-info.json
* module_bp_cc_deps.json
* module_bp_java_deps.json
"""
dirname = atest_utils.get_host_out('indexes')
if not dirname.is_dir():
dirname.mkdir(parents=True)
timestamp = {}
for json_file in [self.mod_info_file_path,
self.java_dep_path,
self.cc_dep_path]:
timestamp.update(
{str(json_file): json_file.stat().st_mtime}
)
timestamp_file = dirname.joinpath('modules.stp')
with open(timestamp_file, 'w', encoding='utf8') as _file:
json.dump(timestamp, _file)
def need_merge_module_info(self):
"""Check if need to merge module info json files.
There are 2 scienarios that atest_merged_dep.json will be updated.
1. One of the checksum of module-info.json, module_bp_java_deps.json and
module_cc_java_deps.json have changed.
2. atest_merged_deps.json does not exist.
If fits one of above scienarios, it is recognized to update.
Returns:
True if one of the scienarios reaches, False otherwise.
"""
if not self.merged_dep_path.is_file():
return True
timestamp_file = atest_utils.get_host_out('indexes/modules.stp')
data = atest_utils.load_json_safely(timestamp_file)
for f in [self.mod_info_file_path,
self.java_dep_path,
self.cc_dep_path]:
if f.stat().st_mtime != data.get(str(f), ''):
return True
return False
def _merge_build_system_infos(self, name_to_module_info,
java_bp_info_path=None, cc_bp_info_path=None):
"""Merge the content of module-info.json and CC/Java dependency files
to name_to_module_info.
Args:
name_to_module_info: Dict of module name to module info dict.
java_bp_info_path: String of path to java dep file to load up.
Used for testing.
cc_bp_info_path: String of path to cc dep file to load up.
Used for testing.
Returns:
Dict of updated name_to_module_info.
"""
# Merge _JAVA_DEP_INFO
if not java_bp_info_path:
java_bp_info_path = self.java_dep_path
java_bp_infos = atest_utils.load_json_safely(java_bp_info_path)
if java_bp_infos:
logging.debug('Merging Java build info: %s', java_bp_info_path)
name_to_module_info = merge_soong_info(
name_to_module_info, java_bp_infos)
# Merge _CC_DEP_INFO
if not cc_bp_info_path:
cc_bp_info_path = self.cc_dep_path
cc_bp_infos = atest_utils.load_json_safely(cc_bp_info_path)
if cc_bp_infos:
logging.debug('Merging CC build info: %s', cc_bp_info_path)
# CC's dep json format is different with java.
# Below is the example content:
# {
# "clang": "${ANDROID_ROOT}/bin/clang",
# "clang++": "${ANDROID_ROOT}/bin/clang++",
# "modules": {
# "ACameraNdkVendorTest": {
# "path": [
# "frameworks/av/camera/ndk"
# ],
# "srcs": [
# "frameworks/tests/AImageVendorTest.cpp",
# "frameworks/tests/ACameraManagerTest.cpp"
# ],
name_to_module_info = merge_soong_info(
name_to_module_info, cc_bp_infos.get('modules', {}))
# If $ANDROID_PRODUCT_OUT was not created in pyfakefs, simply return it
# without dumping atest_merged_dep.json in real.
# Adds the key into module info as a unique ID.
for key, info in name_to_module_info.items():
info[constants.MODULE_INFO_ID] = key
if not self.merged_dep_path.parent.is_dir():
return name_to_module_info
# b/178559543 saving merged module info in a temp file and copying it to
# atest_merged_dep.json can eliminate the possibility of accessing it
# concurrently and resulting in invalid JSON format.
with tempfile.NamedTemporaryFile() as temp_file:
with open(temp_file.name, 'w', encoding='utf-8') as _temp:
json.dump(name_to_module_info, _temp, indent=0)
shutil.copy(temp_file.name, self.merged_dep_path)
return name_to_module_info
def get_testable_modules(self, suite=None):
"""Return the testable modules of the given suite name.
Atest does not index testable modules against compatibility_suites. When
suite was given, or the index file was interrupted, always run
_get_testable_modules() and re-index.
Args:
suite: A string of suite name.
Returns:
If suite is not given, return all the testable modules in module
info, otherwise return only modules that belong to the suite.
"""
modules = set()
start = time.time()
if self.module_index_proc:
self.module_index_proc.join()
if self.module_index.is_file():
if not suite:
with open(self.module_index, 'rb') as cache:
try:
modules = pickle.load(cache, encoding="utf-8")
except UnicodeDecodeError:
modules = pickle.load(cache)
# when module indexing was interrupted.
except EOFError:
pass
else:
modules = self._get_testable_modules(suite=suite)
# If the modules.idx does not exist or invalid for any reason, generate
# a new one arbitrarily.
if not modules:
if not suite:
modules = self._get_testable_modules(index=True)
else:
modules = self._get_testable_modules(index=True, suite=suite)
duration = time.time() - start
metrics.LocalDetectEvent(
detect_type=DetectType.TESTABLE_MODULES,
result=int(duration))
return modules
def _get_testable_modules(self, index=False, suite=None):
"""Return all available testable modules and index them.
Args:
index: boolean that determines running _index_testable_modules().
suite: string for the suite name.
Returns:
Set of all testable modules.
"""
modules = _get_testable_modules(
self.name_to_module_info, self.path_to_module_info, suite)
if index:
self._index_testable_modules(modules)
return modules
def _index_testable_modules(self, content):
"""Dump testable modules.
Args:
content: An object that will be written to the index file.
"""
logging.debug(r'Indexing testable modules... '
r'(This is required whenever module-info.json '
r'was rebuilt.)')
Path(self.module_index).parent.mkdir(parents=True, exist_ok=True)
with open(self.module_index, 'wb') as cache:
try:
pickle.dump(content, cache, protocol=2)
except IOError:
logging.error('Failed in dumping %s', cache)
os.remove(cache)
class ModuleInfo:
"""Class that offers fast/easy lookup for Module related details."""
def __init__(
self,
name_to_module_info: Dict[str, Any]=None,
path_to_module_info: Dict[str, Any]=None,
module_info_target: str=None,
mod_info_file_path: Path=None,
get_testable_modules: Callable=None,
):
"""Initialize the ModuleInfo object.
Load up the module-info.json file and initialize the helper vars.
Note that module-info.json does not contain all module dependencies,
therefore, Atest needs to accumulate dependencies defined in bp files.
+----------------------+ +----------------------------+
| $ANDROID_PRODUCT_OUT | |$ANDROID_BUILD_TOP/out/soong|
| /module-info.json | | /module_bp_java_deps.json |
+-----------+----------+ +-------------+--------------+
| _merge_soong_info() |
+------------------------------+
|
v
+----------------------------+ +----------------------------+
|tempfile.NamedTemporaryFile | |$ANDROID_BUILD_TOP/out/soong|
+-------------+--------------+ | /module_bp_cc_deps.json |
| +-------------+--------------+
| _merge_soong_info() |
+-------------------------------+
|
+-------|
v
+============================+
| $ANDROID_PRODUCT_OUT |
| /atest_merged_dep.json |--> load as module info.
+============================+
Args:
name_to_module_info: Dict of name to module info.
path_to_module_info: Dict of path to module info.
module_info_target: Target name of module-info.json.
mod_info_file_path: Path of module-info.json.
get_testable_modules: Function to get all testable modules.
"""
self.root_dir = os.environ.get(constants.ANDROID_BUILD_TOP)
self.roboleaf_tests = {}
self.name_to_module_info = name_to_module_info or {}
self.path_to_module_info = path_to_module_info or {}
self.module_info_target = module_info_target
self.mod_info_file_path = mod_info_file_path
self._get_testable_modules = get_testable_modules
def is_module(self, name):
"""Return True if name is a module, False otherwise."""
info = self.get_module_info(name)
# From aosp/2293302 it started merging all modules' dependency in bp
# even the module is not be exposed to make, and those modules could not
# be treated as a build target using m. Only treat input name as module
# if it also has the module_name attribute which means it could be a
# build target for m.
if info and info.get(constants.MODULE_NAME):
return True
return False
def get_paths(self, name):
"""Return paths of supplied module name, Empty list if non-existent."""
info = self.get_module_info(name)
if info:
return info.get(constants.MODULE_PATH, [])
return []
def get_module_names(self, rel_module_path):
"""Get the modules that all have module_path.
Args:
rel_module_path: path of module in module-info.json
Returns:
List of module names.
"""
return _get_module_names(self.path_to_module_info, rel_module_path)
def get_module_info(self, mod_name):
"""Return dict of info for given module name, None if non-existence."""
return self.name_to_module_info.get(mod_name)
@staticmethod
def is_suite_in_compatibility_suites(suite, mod_info):
"""Check if suite exists in the compatibility_suites of module-info.
Args:
suite: A string of suite name.
mod_info: Dict of module info to check.
Returns:
True if it exists in mod_info, False otherwise.
"""
if not isinstance(mod_info, dict):
return False
return suite in mod_info.get(
constants.MODULE_COMPATIBILITY_SUITES, [])
def get_testable_modules(self, suite=None):
return self._get_testable_modules(suite)
@staticmethod
def is_tradefed_testable_module(info: Dict[str, Any]) -> bool:
"""Check whether the module is a Tradefed executable test."""
if not info:
return False
if not info.get(constants.MODULE_INSTALLED, []):
return False
return ModuleInfo.has_test_config(info)
@staticmethod
def is_mobly_module(info: Dict[str, Any]) -> bool:
"""Check whether the module is a Mobly test.
Note: Only python_test_host modules marked with a test_options tag of
"mobly" is considered a Mobly module.
Args:
info: Dict of module info to check.
Returns:
True if this is a Mobly test module, False otherwise.
"""
return constants.MOBLY_TEST_OPTIONS_TAG in info.get(
constants.MODULE_TEST_OPTIONS_TAGS, [])
def is_testable_module(self, info: Dict[str, Any]) -> bool:
"""Check if module is something we can test.
A module is testable if:
- it's a tradefed testable module, or
- it's a Mobly module, or
- it's a robolectric module (or shares path with one).
Args:
info: Dict of module info to check.
Returns:
True if we can test this module, False otherwise.
"""
return _is_testable_module(
self.name_to_module_info, self.path_to_module_info, info)
@staticmethod
def has_test_config(info: Dict[str, Any]) -> bool:
"""Validate if this module has a test config.
A module can have a test config in the following manner:
- test_config be set in module-info.json.
- Auto-generated config via the auto_test_config key
in module-info.json.
Args:
info: Dict of module info to check.
Returns:
True if this module has a test config, False otherwise.
"""
return bool(info.get(constants.MODULE_TEST_CONFIG, []) or
info.get('auto_test_config', []))
def is_legacy_robolectric_test(self, info: Dict[str, Any]) -> bool:
"""Return whether the module_name is a legacy Robolectric test"""
return _is_legacy_robolectric_test(
self.name_to_module_info, self.path_to_module_info, info)
def get_robolectric_test_name(self, info: Dict[str, Any]) -> str:
"""Returns runnable robolectric module name.
This method is for legacy robolectric tests and returns one of associated
modules. The pattern is determined by the amount of shards:
10 shards:
FooTests -> RunFooTests0, RunFooTests1 ... RunFooTests9
No shard:
FooTests -> RunFooTests
Arg:
info: Dict of module info to check.
Returns:
String of the first-matched associated module that belongs to the
actual robolectric module, None if nothing has been found.
"""
return _get_robolectric_test_name(
self.name_to_module_info, self.path_to_module_info, info)
def is_robolectric_test(self, module_name):
"""Check if the given module is a robolectric test.
Args:
module_name: String of module to check.
Returns:
Boolean whether it's a robotest or not.
"""
if self.get_robolectric_type(module_name):
return True
return False
def get_robolectric_type(self, module_name: str) -> int:
"""Check if the given module is a robolectric test and return type of it.
Robolectric declaration is converting from Android.mk to Android.bp, and
in the interim Atest needs to support testing both types of tests.
The modern robolectric tests defined by 'android_robolectric_test' in an
Android.bp file can can be run in Tradefed Test Runner:
SettingsRoboTests -> Tradefed Test Runner
Legacy tests defined in an Android.mk can only run with the 'make' way.
SettingsRoboTests -> make RunSettingsRoboTests0
To determine whether the test is a modern/legacy robolectric test:
1. If the 'robolectric-test` in the compatibility_suites, it's a
modern one, otherwise it's a legacy test. This is accurate since
aosp/2308586 already set the test suite of `robolectric-test`
for all `modern` Robolectric tests in Soong.
2. Traverse all modules share the module path. If one of the
modules has a ROBOLECTRIC class, it's a legacy robolectric test.
Args:
module_name: String of module to check.
Returns:
0: not a robolectric test.
1: a modern robolectric test(defined in Android.bp)
2: a legacy robolectric test(defined in Android.mk)
"""
info = self.get_module_info(module_name)
if not info:
return 0
# Some Modern mode Robolectric test has related module which compliant
# with the Legacy Robolectric test. In this case, the Modern mode
# Robolectric tests should be prior to the Legacy mode.
if self.is_modern_robolectric_test(info):
return constants.ROBOTYPE_MODERN
if self.is_legacy_robolectric_test(info):
return constants.ROBOTYPE_LEGACY
return 0
def get_instrumentation_target_apps(self, module_name: str) -> Dict:
"""Return target APKs of an instrumentation test.
Returns:
A dict of target module and target APK(s). e.g.
{"FooService": {"/path/to/the/FooService.apk"}}
"""
# 1. Determine the actual manifest filename from an Android.bp(if any)
manifest = self.get_filepath_from_module(module_name,
'AndroidManifest.xml')
bpfile = self.get_filepath_from_module(module_name, 'Android.bp')
if bpfile.is_file():
bp_info = atest_utils.get_bp_content(bpfile, 'android_test')
if not bp_info or not bp_info.get(module_name):
return {}
manifest = self.get_filepath_from_module(
module_name,
bp_info.get(module_name).get('manifest'))
xml_info = atest_utils.get_manifest_info(manifest)
# 2. Translate package name to a module name.
package = xml_info.get('package')
target_package = xml_info.get('target_package')
# Ensure it's an instrumentation test(excluding self-instrmented)
if target_package and package != target_package:
logging.debug('Found %s an instrumentation test.', module_name)
metrics.LocalDetectEvent(
detect_type=DetectType.FOUND_INSTRUMENTATION_TEST, result=1)
target_module = self.get_target_module_by_pkg(
package=target_package,
search_from=manifest.parent)
if target_module:
return self.get_artifact_map(target_module)
return {}
# pylint: disable=anomalous-backslash-in-string
def get_target_module_by_pkg(self, package: str, search_from: Path) -> str:
"""Translate package name to the target module name.
This method is dedicated to determine the target module by translating
a package name.
Phase 1: Find out possible manifest files among parent directories.
Phase 2. Look for the defined package fits the given name, and ensure
it is not a persistent app.
Phase 3: Translate the manifest path to possible modules. A valid module
must fulfill:
1. The 'class' type must be ['APPS'].
2. It is not a Robolectric test.
Returns:
A string of module name.
"""
xmls = []
for pth in search_from.parents:
if pth == Path(self.root_dir):
break
for name in os.listdir(pth):
if pth.joinpath(name).is_file():
match = re.match('.*AndroidManifest.*\.xml$', name)
if match:
xmls.append(os.path.join(pth, name))
possible_modules = []
for xml in xmls:
rel_dir = str(Path(xml).relative_to(self.root_dir).parent)
logging.debug('Looking for package "%s" in %s...', package, xml)
xml_info = atest_utils.get_manifest_info(xml)
if xml_info.get('package') == package:
if xml_info.get('persistent'):
logging.debug('%s is a persistent app.', package)
continue
for _m in self.path_to_module_info.get(rel_dir):
possible_modules.append(_m)
if possible_modules:
for mod in possible_modules:
name = mod.get('module_name')
if (mod.get('class') == ['APPS'] and
not self.is_robolectric_test(name)):
return name
return ''
def get_artifact_map(self, module_name: str) -> Dict:
"""Get the installed APK path of the given module."""
target_mod_info = self.get_module_info(module_name)
artifact_map = {}
if target_mod_info:
apks = set()
artifacts = target_mod_info.get('installed')
for artifact in artifacts:
if Path(artifact).suffix == '.apk':
apks.add(os.path.join(self.root_dir, artifact))
artifact_map.update({module_name: apks})
return artifact_map
def is_auto_gen_test_config(self, module_name):
"""Check if the test config file will be generated automatically.
Args:
module_name: A string of the module name.
Returns:
True if the test config file will be generated automatically.
"""
if self.is_module(module_name):
mod_info = self.get_module_info(module_name)
auto_test_config = mod_info.get('auto_test_config', [])
return auto_test_config and auto_test_config[0]
return False
@staticmethod
def is_legacy_robolectric_class(info: Dict[str, Any]) -> bool:
"""Check if the class is `ROBOLECTRIC`
This method is for legacy robolectric tests that the associated modules
contain:
'class': ['ROBOLECTRIC']
Args:
info: ModuleInfo to check.
Returns:
True if the attribute class in mod_info is ROBOLECTRIC, False
otherwise.
"""
if info:
module_classes = info.get(constants.MODULE_CLASS, [])
return (module_classes and
module_classes[0] == constants.MODULE_CLASS_ROBOLECTRIC)
return False
def is_native_test(self, module_name):
"""Check if the input module is a native test.
Args:
module_name: A string of the module name.
Returns:
True if the test is a native test, False otherwise.
"""
mod_info = self.get_module_info(module_name)
return constants.MODULE_CLASS_NATIVE_TESTS in mod_info.get(
constants.MODULE_CLASS, [])
def has_mainline_modules(self,
module_name: str, mainline_binaries: List[str]) -> bool:
"""Check if the mainline modules are in module-info.
Args:
module_name: A string of the module name.
mainline_binaries: A list of mainline module binaries.
Returns:
True if mainline_binaries is in module-info, False otherwise.
"""
mod_info = self.get_module_info(module_name)
# Check 'test_mainline_modules' attribute of the module-info.json.
mm_in_mf = mod_info.get(constants.MODULE_MAINLINE_MODULES, [])
ml_modules_set = set(mainline_binaries)
if mm_in_mf:
return contains_same_mainline_modules(
ml_modules_set, set(mm_in_mf))
for test_config in mod_info.get(constants.MODULE_TEST_CONFIG, []):
# Check the value of 'mainline-param' in the test config.
if not self.is_auto_gen_test_config(module_name):
return contains_same_mainline_modules(
ml_modules_set,
atest_utils.get_mainline_param(
os.path.join(self.root_dir, test_config)))
# Unable to verify mainline modules in an auto-gen test config.
logging.debug('%s is associated with an auto-generated test config.',
module_name)
return True
return False
def get_filepath_from_module(self, module_name: str, filename: str) -> Path:
"""Return absolute path of the given module and filename."""
mod_path = self.get_paths(module_name)
if mod_path:
return Path(self.root_dir).joinpath(mod_path[0], filename)
return Path()
def get_module_dependency(self, module_name, depend_on=None):
"""Get the dependency sets for input module.
Recursively find all the dependencies of the input module.
Args:
module_name: String of module to check.
depend_on: The list of parent dependencies.
Returns:
Set of dependency modules.
"""
if not depend_on:
depend_on = set()
deps = set()
mod_info = self.get_module_info(module_name)
if not mod_info:
return deps
mod_deps = set(mod_info.get(constants.MODULE_DEPENDENCIES, []))
# Remove item in deps if it already in depend_on:
mod_deps = mod_deps - depend_on
deps = deps.union(mod_deps)
for mod_dep in mod_deps:
deps = deps.union(set(self.get_module_dependency(
mod_dep, depend_on=depend_on.union(deps))))
return deps
def get_install_module_dependency(self, module_name, depend_on=None):
"""Get the dependency set for the given modules with installed path.
Args:
module_name: String of module to check.
depend_on: The list of parent dependencies.
Returns:
Set of dependency modules which has installed path.
"""
install_deps = set()
deps = self.get_module_dependency(module_name, depend_on)
logging.debug('%s depends on: %s', module_name, deps)
for module in deps:
mod_info = self.get_module_info(module)
if mod_info and mod_info.get(constants.MODULE_INSTALLED, []):
install_deps.add(module)
logging.debug('modules %s required by %s were not installed',
install_deps, module_name)
return install_deps
def is_unit_test(self, mod_info):
"""Return True if input module is unit test, False otherwise.
Args:
mod_info: ModuleInfo to check.
Returns:
True if input module is unit test, False otherwise.
"""
return mod_info.get(constants.MODULE_IS_UNIT_TEST, '') == 'true'
def is_host_unit_test(self, info: Dict[str, Any]) -> bool:
"""Return True if input module is host unit test, False otherwise.
Args:
info: ModuleInfo to check.
Returns:
True if input module is host unit test, False otherwise.
"""
return self.is_tradefed_testable_module(info) and \
self.is_suite_in_compatibility_suites('host-unit-tests', info)
def is_modern_robolectric_test(self, info: Dict[str, Any]) -> bool:
"""Return whether 'robolectric-tests' is in 'compatibility_suites'."""
return self.is_tradefed_testable_module(info) and \
self.is_robolectric_test_suite(info)
def is_robolectric_test_suite(self, mod_info) -> bool:
"""Return True if 'robolectric-tests' in the compatibility_suites.
Args:
mod_info: ModuleInfo to check.
Returns:
True if the 'robolectric-tests' is in the compatibility_suites,
False otherwise.
"""
return self.is_suite_in_compatibility_suites('robolectric-tests',
mod_info)
def is_device_driven_test(self, mod_info):
"""Return True if input module is device driven test, False otherwise.
Args:
mod_info: ModuleInfo to check.
Returns:
True if input module is device driven test, False otherwise.
"""
if self.is_robolectric_test_suite(mod_info):
return False
return self.is_tradefed_testable_module(mod_info) and \
'DEVICE' in mod_info.get(constants.MODULE_SUPPORTED_VARIANTS, [])
def is_host_driven_test(self, mod_info):
"""Return True if input module is host driven test, False otherwise.
Args:
mod_info: ModuleInfo to check.
Returns:
True if input module is host driven test, False otherwise.
"""
return self.is_tradefed_testable_module(mod_info) and \
'HOST' in mod_info.get(constants.MODULE_SUPPORTED_VARIANTS, [])
def _any_module(self, _: Module) -> bool:
return True
def get_all_tests(self):
"""Get a list of all the module names which are tests."""
return self._get_all_modules(type_predicate=self.is_testable_module)
def get_all_unit_tests(self):
"""Get a list of all the module names which are unit tests."""
return self._get_all_modules(type_predicate=self.is_unit_test)
def get_all_host_unit_tests(self):
"""Get a list of all the module names which are host unit tests."""
return self._get_all_modules(type_predicate=self.is_host_unit_test)
def get_all_device_driven_tests(self):
"""Get a list of all the module names which are device driven tests."""
return self._get_all_modules(type_predicate=self.is_device_driven_test)
def _get_all_modules(self, type_predicate=None):
"""Get a list of all the module names that passed the predicate."""
modules = []
type_predicate = type_predicate or self._any_module
for mod_name, mod_info in self.name_to_module_info.items():
if mod_info.get(constants.MODULE_NAME, '') == mod_name:
if type_predicate(mod_info):
modules.append(mod_name)
return modules
def get_modules_by_path_in_srcs(self, path: str) -> Set:
"""Get the module name that the given path belongs to.(in 'srcs')
Args:
path: Relative path to ANDROID_BUILD_TOP of a file.
Returns:
A set of string for matched module names, empty set if nothing find.
"""
modules = set()
for _, mod_info in self.name_to_module_info.items():
if str(path) in mod_info.get(constants.MODULE_SRCS, []):
modules.add(mod_info.get(constants.MODULE_NAME))
return modules
def get_modules_by_include_deps(
self, deps: Set[str],
testable_module_only: bool = False) -> Set[str]:
"""Get the matched module names for the input dependencies.
Args:
deps: A set of string for dependencies.
testable_module_only: Option if only want to get testable module.
Returns:
A set of matched module names for the input dependencies.
"""
modules = set()
for mod_name in (self.get_testable_modules() if testable_module_only
else self.name_to_module_info.keys()):
mod_info = self.get_module_info(mod_name)
if mod_info and deps.intersection(
set(mod_info.get(constants.MODULE_DEPENDENCIES, []))):
modules.add(mod_info.get(constants.MODULE_NAME))
return modules
def get_installed_paths(self, module_name: str) -> [Path]:
mod_info = self.get_module_info(module_name)
if not mod_info:
return []
def _to_abs_path(p):
if os.path.isabs(p):
return Path(p)
return Path(os.getenv(constants.ANDROID_BUILD_TOP), p)
return [_to_abs_path(p) for p in mod_info.get('installed', [])]
def merge_soong_info(name_to_module_info, mod_bp_infos):
"""Merge the dependency and srcs in mod_bp_infos to name_to_module_info.
Args:
name_to_module_info: Dict of module name to module info dict.
mod_bp_infos: Dict of module name to bp's module info dict.
Returns:
Dict of updated name_to_module_info.
"""
merge_items = [constants.MODULE_DEPENDENCIES, constants.MODULE_SRCS,
constants.MODULE_LIBS, constants.MODULE_STATIC_LIBS,
constants.MODULE_STATIC_DEPS, constants.MODULE_PATH]
for module_name, dep_info in mod_bp_infos.items():
mod_info = name_to_module_info.setdefault(module_name, {})
for merge_item in merge_items:
dep_info_values = dep_info.get(merge_item, [])
mod_info_values = mod_info.get(merge_item, [])
mod_info_values.extend(dep_info_values)
mod_info_values.sort()
# deduplicate values just in case.
mod_info_values = list(dict.fromkeys(mod_info_values))
name_to_module_info[
module_name][merge_item] = mod_info_values
return name_to_module_info
def _discover_mod_file_and_target(force_build):
"""Find the module file.
Args:
force_build: Boolean to indicate if we should rebuild the
module_info file regardless of the existence of it.
Returns:
Tuple of module_info_target and path to the module-info.json.
"""
logging.debug('Probing and validating module info...')
root_dir = Path(os.getenv(constants.ANDROID_BUILD_TOP))
out_dir = Path(os.getenv(constants.ANDROID_PRODUCT_OUT))
module_file_path = out_dir.joinpath(_MODULE_INFO)
# If OUT_DIR/OUT_DIR_COMMON_BASE was set outside of the root_dir, use
# absolute path; otherwise use relative path as the target name.
if out_dir.is_relative_to(root_dir):
module_info_target = str(module_file_path.relative_to(root_dir))
else:
logging.debug('User customized out dir!')
module_file_path = out_dir.joinpath(_MODULE_INFO)
module_info_target = str(module_file_path)
if force_build or not module_file_path.is_file():
atest_utils.build_module_info_target(module_info_target)
return module_info_target, module_file_path
def _add_missing_variant_modules(name_to_module_info: Dict[str, Module]):
missing_modules = {}
# Android's build system automatically adds a suffix for some build module
# variants. For example, a module-info entry for a module originally named
# 'HelloWorldTest' might appear as 'HelloWorldTest_32' and which Atest would
# not be able to find. We add such entries if not already present so they
# can be looked up using their declared module name.
for mod_name, mod_info in name_to_module_info.items():
declared_module_name = mod_info.get(constants.MODULE_NAME, mod_name)
if declared_module_name in name_to_module_info:
continue
missing_modules.setdefault(declared_module_name, mod_info)
name_to_module_info.update(missing_modules)
def contains_same_mainline_modules(mainline_modules: Set[str], module_lists: Set[str]):
"""Check if mainline modules listed on command line is
the same set as config.
Args:
mainline_modules: A list of mainline modules from triggered test.
module_lists: A list of concatenate mainline module string from test configs.
Returns
True if the set mainline modules from triggered test is in the test configs.
"""
for module_string in module_lists:
if mainline_modules == set(module_string.split('+')):
return True
return False
def get_path_to_module_info(name_to_module_info):
"""Return the path_to_module_info dict.
Args:
name_to_module_info: Dict of module name to module info dict.
Returns:
Dict of module path to module info dict.
"""
path_to_module_info = {}
for mod_name, mod_info in name_to_module_info.items():
# Cross-compiled and multi-arch modules actually all belong to
# a single target so filter out these extra modules.
if mod_name != mod_info.get(constants.MODULE_NAME, ''):
continue
for path in mod_info.get(constants.MODULE_PATH, []):
mod_info[constants.MODULE_NAME] = mod_name
# There could be multiple modules in a path.
if path in path_to_module_info:
path_to_module_info[path].append(mod_info)
else:
path_to_module_info[path] = [mod_info]
return path_to_module_info
def _get_module_names(path_to_module_info, rel_module_path):
"""Get the modules that all have module_path.
Args:
path_to_module_info: Dict of path to module info.
rel_module_path: path of module in module-info.json.
Returns:
List of module names.
"""
return [m.get(constants.MODULE_NAME)
for m in path_to_module_info.get(rel_module_path, [])]
def _get_robolectric_test_name(
name_to_module_info: Dict[str, Dict],
path_to_module_info: Dict[str, Dict],
info: Dict[str, Any]) -> str:
"""Returns runnable robolectric module name.
This method is for legacy robolectric tests and returns one of associated
modules. The pattern is determined by the amount of shards:
10 shards:
FooTests -> RunFooTests0, RunFooTests1 ... RunFooTests9
No shard:
FooTests -> RunFooTests
Arg:
name_to_module_info: Dict of name to module info.
path_to_module_info: Dict of path to module info.
info: Dict of module info to check.
Returns:
String of the first-matched associated module that belongs to the
actual robolectric module, None if nothing has been found.
"""
if not info:
return ''
module_paths = info.get(constants.MODULE_PATH, [])
if not module_paths:
return ''
filtered_module_names = [
name
for name in _get_module_names(path_to_module_info, module_paths[0])
if name.startswith("Run")
]
return next(
(
name
for name in filtered_module_names
if ModuleInfo.is_legacy_robolectric_class(name_to_module_info.get(name))
),
'',
)
def _is_legacy_robolectric_test(
name_to_module_info: Dict[str, Dict],
path_to_module_info: Dict[str, Dict],
info: Dict[str, Any]) -> bool:
"""Return whether the module_name is a legacy Robolectric test"""
if ModuleInfo.is_tradefed_testable_module(info):
return False
return bool(_get_robolectric_test_name(
name_to_module_info, path_to_module_info, info))
def _is_testable_module(
name_to_module_info: Dict[str, Dict],
path_to_module_info: Dict[str, Dict],
info: Dict[str, Any]) -> bool:
"""Check if module is something we can test.
A module is testable if:
- it's a tradefed testable module, or
- it's a Mobly module, or
- it's a robolectric module (or shares path with one).
Args:
name_to_module_info: Dict of name to module info.
path_to_module_info: Dict of path to module info.
info: Dict of module info to check.
Returns:
True if we can test this module, False otherwise.
"""
if not info:
return False
if ModuleInfo.is_tradefed_testable_module(info):
return True
if ModuleInfo.is_mobly_module(info):
return True
if _is_legacy_robolectric_test(name_to_module_info, path_to_module_info, info):
return True
return False
def _get_testable_modules(
name_to_module_info: Dict[str, Dict],
path_to_module_info: Dict[str, Dict],
suite: bool=None):
modules = set()
begin = time.time()
for _, info in name_to_module_info.items():
if _is_testable_module(name_to_module_info, path_to_module_info, info):
testable_module = info.get(constants.MODULE_NAME)
if testable_module:
modules.add(testable_module)
logging.debug('Probing all testable modules took %ss',
time.time() - begin)
if suite:
_modules = set()
for module_name in modules:
info = name_to_module_info.get(module_name)
if ModuleInfo.is_suite_in_compatibility_suites(suite, info):
testable_module = info.get(constants.MODULE_NAME)
if testable_module:
_modules.add(testable_module)
return _modules
return modules